1
0

Subject.cs 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169
  1. using System;
  2. using System.Collections.Generic;
  3. using System.Text;
  4. using UniRx.InternalUtil;
  5. namespace UniRx
  6. {
  7. public sealed class Subject<T> : ISubject<T>, IDisposable, IOptimizedObservable<T>
  8. {
  9. object observerLock = new object();
  10. bool isStopped;
  11. bool isDisposed;
  12. Exception lastError;
  13. IObserver<T> outObserver = EmptyObserver<T>.Instance;
  14. public bool HasObservers
  15. {
  16. get
  17. {
  18. return !(outObserver is EmptyObserver<T>) && !isStopped && !isDisposed;
  19. }
  20. }
  21. public void OnCompleted()
  22. {
  23. IObserver<T> old;
  24. lock (observerLock)
  25. {
  26. ThrowIfDisposed();
  27. if (isStopped) return;
  28. old = outObserver;
  29. outObserver = EmptyObserver<T>.Instance;
  30. isStopped = true;
  31. }
  32. old.OnCompleted();
  33. }
  34. public void OnError(Exception error)
  35. {
  36. if (error == null) throw new ArgumentNullException("error");
  37. IObserver<T> old;
  38. lock (observerLock)
  39. {
  40. ThrowIfDisposed();
  41. if (isStopped) return;
  42. old = outObserver;
  43. outObserver = EmptyObserver<T>.Instance;
  44. isStopped = true;
  45. lastError = error;
  46. }
  47. old.OnError(error);
  48. }
  49. public void OnNext(T value)
  50. {
  51. outObserver.OnNext(value);
  52. }
  53. public IDisposable Subscribe(IObserver<T> observer)
  54. {
  55. if (observer == null) throw new ArgumentNullException("observer");
  56. var ex = default(Exception);
  57. lock (observerLock)
  58. {
  59. ThrowIfDisposed();
  60. if (!isStopped)
  61. {
  62. var listObserver = outObserver as ListObserver<T>;
  63. if (listObserver != null)
  64. {
  65. outObserver = listObserver.Add(observer);
  66. }
  67. else
  68. {
  69. var current = outObserver;
  70. if (current is EmptyObserver<T>)
  71. {
  72. outObserver = observer;
  73. }
  74. else
  75. {
  76. outObserver = new ListObserver<T>(new ImmutableList<IObserver<T>>(new[] { current, observer }));
  77. }
  78. }
  79. return new Subscription(this, observer);
  80. }
  81. ex = lastError;
  82. }
  83. if (ex != null)
  84. {
  85. observer.OnError(ex);
  86. }
  87. else
  88. {
  89. observer.OnCompleted();
  90. }
  91. return Disposable.Empty;
  92. }
  93. public void Dispose()
  94. {
  95. lock (observerLock)
  96. {
  97. isDisposed = true;
  98. outObserver = DisposedObserver<T>.Instance;
  99. }
  100. }
  101. void ThrowIfDisposed()
  102. {
  103. if (isDisposed) throw new ObjectDisposedException("");
  104. }
  105. public bool IsRequiredSubscribeOnCurrentThread()
  106. {
  107. return false;
  108. }
  109. class Subscription : IDisposable
  110. {
  111. readonly object gate = new object();
  112. Subject<T> parent;
  113. IObserver<T> unsubscribeTarget;
  114. public Subscription(Subject<T> parent, IObserver<T> unsubscribeTarget)
  115. {
  116. this.parent = parent;
  117. this.unsubscribeTarget = unsubscribeTarget;
  118. }
  119. public void Dispose()
  120. {
  121. lock (gate)
  122. {
  123. if (parent != null)
  124. {
  125. lock (parent.observerLock)
  126. {
  127. var listObserver = parent.outObserver as ListObserver<T>;
  128. if (listObserver != null)
  129. {
  130. parent.outObserver = listObserver.Remove(unsubscribeTarget);
  131. }
  132. else
  133. {
  134. parent.outObserver = EmptyObserver<T>.Instance;
  135. }
  136. unsubscribeTarget = null;
  137. parent = null;
  138. }
  139. }
  140. }
  141. }
  142. }
  143. }
  144. }