BatchFrame.cs 9.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282
  1. using System;
  2. using System.Collections.Generic;
  3. namespace UniRx.Operators
  4. {
  5. internal class BatchFrameObservable<T> : OperatorObservableBase<IList<T>>
  6. {
  7. readonly IObservable<T> source;
  8. readonly int frameCount;
  9. readonly FrameCountType frameCountType;
  10. public BatchFrameObservable(IObservable<T> source, int frameCount, FrameCountType frameCountType)
  11. : base(source.IsRequiredSubscribeOnCurrentThread())
  12. {
  13. this.source = source;
  14. this.frameCount = frameCount;
  15. this.frameCountType = frameCountType;
  16. }
  17. protected override IDisposable SubscribeCore(IObserver<IList<T>> observer, IDisposable cancel)
  18. {
  19. return new BatchFrame(this, observer, cancel).Run();
  20. }
  21. class BatchFrame : OperatorObserverBase<T, IList<T>>
  22. {
  23. readonly BatchFrameObservable<T> parent;
  24. readonly object gate = new object();
  25. readonly BooleanDisposable cancellationToken = new BooleanDisposable();
  26. readonly System.Collections.IEnumerator timer;
  27. bool isRunning;
  28. bool isCompleted;
  29. List<T> list;
  30. public BatchFrame(BatchFrameObservable<T> parent, IObserver<IList<T>> observer, IDisposable cancel) : base(observer, cancel)
  31. {
  32. this.parent = parent;
  33. this.timer = new ReusableEnumerator(this);
  34. }
  35. public IDisposable Run()
  36. {
  37. list = new List<T>();
  38. var sourceSubscription = parent.source.Subscribe(this);
  39. return StableCompositeDisposable.Create(sourceSubscription, cancellationToken);
  40. }
  41. public override void OnNext(T value)
  42. {
  43. lock (gate)
  44. {
  45. if (isCompleted) return;
  46. list.Add(value);
  47. if (!isRunning)
  48. {
  49. isRunning = true;
  50. timer.Reset(); // reuse
  51. switch (parent.frameCountType)
  52. {
  53. case FrameCountType.Update:
  54. MainThreadDispatcher.StartUpdateMicroCoroutine(timer);
  55. break;
  56. case FrameCountType.FixedUpdate:
  57. MainThreadDispatcher.StartFixedUpdateMicroCoroutine(timer);
  58. break;
  59. case FrameCountType.EndOfFrame:
  60. MainThreadDispatcher.StartEndOfFrameMicroCoroutine(timer);
  61. break;
  62. default:
  63. break;
  64. }
  65. }
  66. }
  67. }
  68. public override void OnError(Exception error)
  69. {
  70. try { observer.OnError(error); } finally { Dispose(); }
  71. }
  72. public override void OnCompleted()
  73. {
  74. List<T> currentList;
  75. lock (gate)
  76. {
  77. isCompleted = true;
  78. currentList = list;
  79. }
  80. if (currentList.Count != 0)
  81. {
  82. observer.OnNext(currentList);
  83. }
  84. try { observer.OnCompleted(); } finally { Dispose(); }
  85. }
  86. // reuse, no gc allocate
  87. class ReusableEnumerator : System.Collections.IEnumerator
  88. {
  89. readonly BatchFrame parent;
  90. int currentFrame;
  91. public ReusableEnumerator(BatchFrame parent)
  92. {
  93. this.parent = parent;
  94. }
  95. public object Current
  96. {
  97. get { return null; }
  98. }
  99. public bool MoveNext()
  100. {
  101. if (parent.cancellationToken.IsDisposed) return false;
  102. List<T> currentList;
  103. lock (parent.gate)
  104. {
  105. if (currentFrame++ == parent.parent.frameCount)
  106. {
  107. if (parent.isCompleted) return false;
  108. currentList = parent.list;
  109. parent.list = new List<T>();
  110. parent.isRunning = false;
  111. // exit lock
  112. }
  113. else
  114. {
  115. return true;
  116. }
  117. }
  118. parent.observer.OnNext(currentList);
  119. return false;
  120. }
  121. public void Reset()
  122. {
  123. currentFrame = 0;
  124. }
  125. }
  126. }
  127. }
  128. internal class BatchFrameObservable : OperatorObservableBase<Unit>
  129. {
  130. readonly IObservable<Unit> source;
  131. readonly int frameCount;
  132. readonly FrameCountType frameCountType;
  133. public BatchFrameObservable(IObservable<Unit> source, int frameCount, FrameCountType frameCountType)
  134. : base(source.IsRequiredSubscribeOnCurrentThread())
  135. {
  136. this.source = source;
  137. this.frameCount = frameCount;
  138. this.frameCountType = frameCountType;
  139. }
  140. protected override IDisposable SubscribeCore(IObserver<Unit> observer, IDisposable cancel)
  141. {
  142. return new BatchFrame(this, observer, cancel).Run();
  143. }
  144. class BatchFrame : OperatorObserverBase<Unit, Unit>
  145. {
  146. readonly BatchFrameObservable parent;
  147. readonly object gate = new object();
  148. readonly BooleanDisposable cancellationToken = new BooleanDisposable();
  149. readonly System.Collections.IEnumerator timer;
  150. bool isRunning;
  151. bool isCompleted;
  152. public BatchFrame(BatchFrameObservable parent, IObserver<Unit> observer, IDisposable cancel) : base(observer, cancel)
  153. {
  154. this.parent = parent;
  155. this.timer = new ReusableEnumerator(this);
  156. }
  157. public IDisposable Run()
  158. {
  159. var sourceSubscription = parent.source.Subscribe(this);
  160. return StableCompositeDisposable.Create(sourceSubscription, cancellationToken);
  161. }
  162. public override void OnNext(Unit value)
  163. {
  164. lock (gate)
  165. {
  166. if (!isRunning)
  167. {
  168. isRunning = true;
  169. timer.Reset(); // reuse
  170. switch (parent.frameCountType)
  171. {
  172. case FrameCountType.Update:
  173. MainThreadDispatcher.StartUpdateMicroCoroutine(timer);
  174. break;
  175. case FrameCountType.FixedUpdate:
  176. MainThreadDispatcher.StartFixedUpdateMicroCoroutine(timer);
  177. break;
  178. case FrameCountType.EndOfFrame:
  179. MainThreadDispatcher.StartEndOfFrameMicroCoroutine(timer);
  180. break;
  181. default:
  182. break;
  183. }
  184. }
  185. }
  186. }
  187. public override void OnError(Exception error)
  188. {
  189. try { observer.OnError(error); } finally { Dispose(); }
  190. }
  191. public override void OnCompleted()
  192. {
  193. bool running;
  194. lock (gate)
  195. {
  196. running = isRunning;
  197. isCompleted = true;
  198. }
  199. if (running)
  200. {
  201. observer.OnNext(Unit.Default);
  202. }
  203. try { observer.OnCompleted(); } finally { Dispose(); }
  204. }
  205. // reuse, no gc allocate
  206. class ReusableEnumerator : System.Collections.IEnumerator
  207. {
  208. readonly BatchFrame parent;
  209. int currentFrame;
  210. public ReusableEnumerator(BatchFrame parent)
  211. {
  212. this.parent = parent;
  213. }
  214. public object Current
  215. {
  216. get { return null; }
  217. }
  218. public bool MoveNext()
  219. {
  220. if (parent.cancellationToken.IsDisposed) return false;
  221. lock (parent.gate)
  222. {
  223. if (currentFrame++ == parent.parent.frameCount)
  224. {
  225. if (parent.isCompleted) return false;
  226. parent.isRunning = false;
  227. // exit lock
  228. }
  229. else
  230. {
  231. return true;
  232. }
  233. }
  234. parent.observer.OnNext(Unit.Default);
  235. return false;
  236. }
  237. public void Reset()
  238. {
  239. currentFrame = 0;
  240. }
  241. }
  242. }
  243. }
  244. }