RefCount.cs 2.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677
  1. using System;
  2. using UniRx.Operators;
  3. namespace UniRx.Operators
  4. {
  5. internal class RefCountObservable<T> : OperatorObservableBase<T>
  6. {
  7. readonly IConnectableObservable<T> source;
  8. readonly object gate = new object();
  9. int refCount = 0;
  10. IDisposable connection;
  11. public RefCountObservable(IConnectableObservable<T> source)
  12. : base(source.IsRequiredSubscribeOnCurrentThread())
  13. {
  14. this.source = source;
  15. }
  16. protected override IDisposable SubscribeCore(IObserver<T> observer, IDisposable cancel)
  17. {
  18. return new RefCount(this, observer, cancel).Run();
  19. }
  20. class RefCount : OperatorObserverBase<T, T>
  21. {
  22. readonly RefCountObservable<T> parent;
  23. public RefCount(RefCountObservable<T> parent, IObserver<T> observer, IDisposable cancel) : base(observer, cancel)
  24. {
  25. this.parent = parent;
  26. }
  27. public IDisposable Run()
  28. {
  29. var subcription = parent.source.Subscribe(this);
  30. lock (parent.gate)
  31. {
  32. if (++parent.refCount == 1)
  33. {
  34. parent.connection = parent.source.Connect();
  35. }
  36. }
  37. return Disposable.Create(() =>
  38. {
  39. subcription.Dispose();
  40. lock (parent.gate)
  41. {
  42. if (--parent.refCount == 0)
  43. {
  44. parent.connection.Dispose();
  45. }
  46. }
  47. });
  48. }
  49. public override void OnNext(T value)
  50. {
  51. base.observer.OnNext(value);
  52. }
  53. public override void OnError(Exception error)
  54. {
  55. try { observer.OnError(error); }
  56. finally { Dispose(); }
  57. }
  58. public override void OnCompleted()
  59. {
  60. try { observer.OnCompleted(); }
  61. finally { Dispose(); }
  62. }
  63. }
  64. }
  65. }