2
0

Merge.cs 8.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264
  1. using System;
  2. using System.Collections.Generic;
  3. namespace UniRx.Operators
  4. {
  5. internal class MergeObservable<T> : OperatorObservableBase<T>
  6. {
  7. private readonly IObservable<IObservable<T>> sources;
  8. private readonly int maxConcurrent;
  9. public MergeObservable(IObservable<IObservable<T>> sources, bool isRequiredSubscribeOnCurrentThread)
  10. : base(isRequiredSubscribeOnCurrentThread)
  11. {
  12. this.sources = sources;
  13. }
  14. public MergeObservable(IObservable<IObservable<T>> sources, int maxConcurrent, bool isRequiredSubscribeOnCurrentThread)
  15. : base(isRequiredSubscribeOnCurrentThread)
  16. {
  17. this.sources = sources;
  18. this.maxConcurrent = maxConcurrent;
  19. }
  20. protected override IDisposable SubscribeCore(IObserver<T> observer, IDisposable cancel)
  21. {
  22. if (maxConcurrent > 0)
  23. {
  24. return new MergeConcurrentObserver(this, observer, cancel).Run();
  25. }
  26. else
  27. {
  28. return new MergeOuterObserver(this, observer, cancel).Run();
  29. }
  30. }
  31. class MergeOuterObserver : OperatorObserverBase<IObservable<T>, T>
  32. {
  33. readonly MergeObservable<T> parent;
  34. CompositeDisposable collectionDisposable;
  35. SingleAssignmentDisposable sourceDisposable;
  36. object gate = new object();
  37. bool isStopped = false;
  38. public MergeOuterObserver(MergeObservable<T> parent, IObserver<T> observer, IDisposable cancel) : base(observer, cancel)
  39. {
  40. this.parent = parent;
  41. }
  42. public IDisposable Run()
  43. {
  44. collectionDisposable = new CompositeDisposable();
  45. sourceDisposable = new SingleAssignmentDisposable();
  46. collectionDisposable.Add(sourceDisposable);
  47. sourceDisposable.Disposable = parent.sources.Subscribe(this);
  48. return collectionDisposable;
  49. }
  50. public override void OnNext(IObservable<T> value)
  51. {
  52. var disposable = new SingleAssignmentDisposable();
  53. collectionDisposable.Add(disposable);
  54. var collectionObserver = new Merge(this, disposable);
  55. disposable.Disposable = value.Subscribe(collectionObserver);
  56. }
  57. public override void OnError(Exception error)
  58. {
  59. lock (gate)
  60. {
  61. try { observer.OnError(error); } finally { Dispose(); };
  62. }
  63. }
  64. public override void OnCompleted()
  65. {
  66. isStopped = true;
  67. if (collectionDisposable.Count == 1)
  68. {
  69. lock (gate)
  70. {
  71. try { observer.OnCompleted(); } finally { Dispose(); };
  72. }
  73. }
  74. else
  75. {
  76. sourceDisposable.Dispose();
  77. }
  78. }
  79. class Merge : OperatorObserverBase<T, T>
  80. {
  81. readonly MergeOuterObserver parent;
  82. readonly IDisposable cancel;
  83. public Merge(MergeOuterObserver parent, IDisposable cancel)
  84. : base(parent.observer, cancel)
  85. {
  86. this.parent = parent;
  87. this.cancel = cancel;
  88. }
  89. public override void OnNext(T value)
  90. {
  91. lock (parent.gate)
  92. {
  93. base.observer.OnNext(value);
  94. }
  95. }
  96. public override void OnError(Exception error)
  97. {
  98. lock (parent.gate)
  99. {
  100. try { observer.OnError(error); } finally { Dispose(); };
  101. }
  102. }
  103. public override void OnCompleted()
  104. {
  105. parent.collectionDisposable.Remove(cancel);
  106. if (parent.isStopped && parent.collectionDisposable.Count == 1)
  107. {
  108. lock (parent.gate)
  109. {
  110. try { observer.OnCompleted(); } finally { Dispose(); };
  111. }
  112. }
  113. }
  114. }
  115. }
  116. class MergeConcurrentObserver : OperatorObserverBase<IObservable<T>, T>
  117. {
  118. readonly MergeObservable<T> parent;
  119. CompositeDisposable collectionDisposable;
  120. SingleAssignmentDisposable sourceDisposable;
  121. object gate = new object();
  122. bool isStopped = false;
  123. // concurrency
  124. Queue<IObservable<T>> q;
  125. int activeCount;
  126. public MergeConcurrentObserver(MergeObservable<T> parent, IObserver<T> observer, IDisposable cancel) : base(observer, cancel)
  127. {
  128. this.parent = parent;
  129. }
  130. public IDisposable Run()
  131. {
  132. q = new Queue<IObservable<T>>();
  133. activeCount = 0;
  134. collectionDisposable = new CompositeDisposable();
  135. sourceDisposable = new SingleAssignmentDisposable();
  136. collectionDisposable.Add(sourceDisposable);
  137. sourceDisposable.Disposable = parent.sources.Subscribe(this);
  138. return collectionDisposable;
  139. }
  140. public override void OnNext(IObservable<T> value)
  141. {
  142. lock (gate)
  143. {
  144. if (activeCount < parent.maxConcurrent)
  145. {
  146. activeCount++;
  147. Subscribe(value);
  148. }
  149. else
  150. {
  151. q.Enqueue(value);
  152. }
  153. }
  154. }
  155. public override void OnError(Exception error)
  156. {
  157. lock (gate)
  158. {
  159. try { observer.OnError(error); } finally { Dispose(); };
  160. }
  161. }
  162. public override void OnCompleted()
  163. {
  164. lock (gate)
  165. {
  166. isStopped = true;
  167. if (activeCount == 0)
  168. {
  169. try { observer.OnCompleted(); } finally { Dispose(); };
  170. }
  171. else
  172. {
  173. sourceDisposable.Dispose();
  174. }
  175. }
  176. }
  177. void Subscribe(IObservable<T> innerSource)
  178. {
  179. var disposable = new SingleAssignmentDisposable();
  180. collectionDisposable.Add(disposable);
  181. var collectionObserver = new Merge(this, disposable);
  182. disposable.Disposable = innerSource.Subscribe(collectionObserver);
  183. }
  184. class Merge : OperatorObserverBase<T, T>
  185. {
  186. readonly MergeConcurrentObserver parent;
  187. readonly IDisposable cancel;
  188. public Merge(MergeConcurrentObserver parent, IDisposable cancel)
  189. : base(parent.observer, cancel)
  190. {
  191. this.parent = parent;
  192. this.cancel = cancel;
  193. }
  194. public override void OnNext(T value)
  195. {
  196. lock (parent.gate)
  197. {
  198. base.observer.OnNext(value);
  199. }
  200. }
  201. public override void OnError(Exception error)
  202. {
  203. lock (parent.gate)
  204. {
  205. try { observer.OnError(error); } finally { Dispose(); };
  206. }
  207. }
  208. public override void OnCompleted()
  209. {
  210. parent.collectionDisposable.Remove(cancel);
  211. lock (parent.gate)
  212. {
  213. if (parent.q.Count > 0)
  214. {
  215. var source = parent.q.Dequeue();
  216. parent.Subscribe(source);
  217. }
  218. else
  219. {
  220. parent.activeCount--;
  221. if (parent.isStopped && parent.activeCount == 0)
  222. {
  223. try { observer.OnCompleted(); } finally { Dispose(); };
  224. }
  225. }
  226. }
  227. }
  228. }
  229. }
  230. }
  231. }