TakeWhile.cs 3.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134
  1. using System;
  2. namespace UniRx.Operators
  3. {
  4. internal class TakeWhileObservable<T> : OperatorObservableBase<T>
  5. {
  6. readonly IObservable<T> source;
  7. readonly Func<T, bool> predicate;
  8. readonly Func<T, int, bool> predicateWithIndex;
  9. public TakeWhileObservable(IObservable<T> source, Func<T, bool> predicate)
  10. : base(source.IsRequiredSubscribeOnCurrentThread())
  11. {
  12. this.source = source;
  13. this.predicate = predicate;
  14. }
  15. public TakeWhileObservable(IObservable<T> source, Func<T, int, bool> predicateWithIndex)
  16. : base(source.IsRequiredSubscribeOnCurrentThread())
  17. {
  18. this.source = source;
  19. this.predicateWithIndex = predicateWithIndex;
  20. }
  21. protected override IDisposable SubscribeCore(IObserver<T> observer, IDisposable cancel)
  22. {
  23. if (predicate != null)
  24. {
  25. return new TakeWhile(this, observer, cancel).Run();
  26. }
  27. else
  28. {
  29. return new TakeWhile_(this, observer, cancel).Run();
  30. }
  31. }
  32. class TakeWhile : OperatorObserverBase<T, T>
  33. {
  34. readonly TakeWhileObservable<T> parent;
  35. public TakeWhile(TakeWhileObservable<T> parent, IObserver<T> observer, IDisposable cancel) : base(observer, cancel)
  36. {
  37. this.parent = parent;
  38. }
  39. public IDisposable Run()
  40. {
  41. return parent.source.Subscribe(this);
  42. }
  43. public override void OnNext(T value)
  44. {
  45. bool isPassed;
  46. try
  47. {
  48. isPassed = parent.predicate(value);
  49. }
  50. catch (Exception ex)
  51. {
  52. try { observer.OnError(ex); } finally { Dispose(); }
  53. return;
  54. }
  55. if (isPassed)
  56. {
  57. observer.OnNext(value);
  58. }
  59. else
  60. {
  61. try { observer.OnCompleted(); } finally { Dispose(); }
  62. }
  63. }
  64. public override void OnError(Exception error)
  65. {
  66. try { observer.OnError(error); } finally { Dispose(); }
  67. }
  68. public override void OnCompleted()
  69. {
  70. try { observer.OnCompleted(); } finally { Dispose(); }
  71. }
  72. }
  73. class TakeWhile_ : OperatorObserverBase<T, T>
  74. {
  75. readonly TakeWhileObservable<T> parent;
  76. int index = 0;
  77. public TakeWhile_(TakeWhileObservable<T> parent, IObserver<T> observer, IDisposable cancel) : base(observer, cancel)
  78. {
  79. this.parent = parent;
  80. }
  81. public IDisposable Run()
  82. {
  83. return parent.source.Subscribe(this);
  84. }
  85. public override void OnNext(T value)
  86. {
  87. bool isPassed;
  88. try
  89. {
  90. isPassed = parent.predicateWithIndex(value, index++);
  91. }
  92. catch (Exception ex)
  93. {
  94. try { observer.OnError(ex); } finally { Dispose(); }
  95. return;
  96. }
  97. if (isPassed)
  98. {
  99. observer.OnNext(value);
  100. }
  101. else
  102. {
  103. try { observer.OnCompleted(); } finally { Dispose(); }
  104. }
  105. }
  106. public override void OnError(Exception error)
  107. {
  108. try { observer.OnError(error); } finally { Dispose(); }
  109. }
  110. public override void OnCompleted()
  111. {
  112. try { observer.OnCompleted(); } finally { Dispose(); }
  113. }
  114. }
  115. }
  116. }