2
0

Distinct.cs 4.1 KB

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