Concat.cs 4.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142
  1. using System;
  2. using System.Collections.Generic;
  3. namespace UniRx.Operators
  4. {
  5. // needs to more improvement
  6. internal class ConcatObservable<T> : OperatorObservableBase<T>
  7. {
  8. readonly IEnumerable<IObservable<T>> sources;
  9. public ConcatObservable(IEnumerable<IObservable<T>> sources)
  10. : base(true)
  11. {
  12. this.sources = sources;
  13. }
  14. public IObservable<T> Combine(IEnumerable<IObservable<T>> combineSources)
  15. {
  16. return new ConcatObservable<T>(CombineSources(this.sources, combineSources));
  17. }
  18. static IEnumerable<IObservable<T>> CombineSources(IEnumerable<IObservable<T>> first, IEnumerable<IObservable<T>> second)
  19. {
  20. foreach (var item in first)
  21. {
  22. yield return item;
  23. }
  24. foreach (var item in second)
  25. {
  26. yield return item;
  27. }
  28. }
  29. protected override IDisposable SubscribeCore(IObserver<T> observer, IDisposable cancel)
  30. {
  31. return new Concat(this, observer, cancel).Run();
  32. }
  33. class Concat : OperatorObserverBase<T, T>
  34. {
  35. readonly ConcatObservable<T> parent;
  36. readonly object gate = new object();
  37. bool isDisposed;
  38. IEnumerator<IObservable<T>> e;
  39. SerialDisposable subscription;
  40. Action nextSelf;
  41. public Concat(ConcatObservable<T> parent, IObserver<T> observer, IDisposable cancel)
  42. : base(observer, cancel)
  43. {
  44. this.parent = parent;
  45. }
  46. public IDisposable Run()
  47. {
  48. isDisposed = false;
  49. e = parent.sources.GetEnumerator();
  50. subscription = new SerialDisposable();
  51. var schedule = Scheduler.DefaultSchedulers.TailRecursion.Schedule(RecursiveRun);
  52. return StableCompositeDisposable.Create(schedule, subscription, Disposable.Create(() =>
  53. {
  54. lock (gate)
  55. {
  56. this.isDisposed = true;
  57. this.e.Dispose();
  58. }
  59. }));
  60. }
  61. void RecursiveRun(Action self)
  62. {
  63. lock (gate)
  64. {
  65. this.nextSelf = self;
  66. if (isDisposed) return;
  67. var current = default(IObservable<T>);
  68. var hasNext = false;
  69. var ex = default(Exception);
  70. try
  71. {
  72. hasNext = e.MoveNext();
  73. if (hasNext)
  74. {
  75. current = e.Current;
  76. if (current == null) throw new InvalidOperationException("sequence is null.");
  77. }
  78. else
  79. {
  80. e.Dispose();
  81. }
  82. }
  83. catch (Exception exception)
  84. {
  85. ex = exception;
  86. e.Dispose();
  87. }
  88. if (ex != null)
  89. {
  90. try { observer.OnError(ex); }
  91. finally { Dispose(); }
  92. return;
  93. }
  94. if (!hasNext)
  95. {
  96. try { observer.OnCompleted(); }
  97. finally { Dispose(); }
  98. return;
  99. }
  100. var source = current;
  101. var d = new SingleAssignmentDisposable();
  102. subscription.Disposable = d;
  103. d.Disposable = source.Subscribe(this);
  104. }
  105. }
  106. public override void OnNext(T value)
  107. {
  108. base.observer.OnNext(value);
  109. }
  110. public override void OnError(Exception error)
  111. {
  112. try { observer.OnError(error); }
  113. finally { Dispose(); }
  114. }
  115. public override void OnCompleted()
  116. {
  117. this.nextSelf();
  118. }
  119. }
  120. }
  121. }