TimeoutFrame.cs 4.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142
  1. using System;
  2. #if UniRxLibrary
  3. using UnityObservable = UniRx.ObservableUnity;
  4. #else
  5. using UnityObservable = UniRx.Observable;
  6. #endif
  7. namespace UniRx.Operators
  8. {
  9. internal class TimeoutFrameObservable<T> : OperatorObservableBase<T>
  10. {
  11. readonly IObservable<T> source;
  12. readonly int frameCount;
  13. readonly FrameCountType frameCountType;
  14. public TimeoutFrameObservable(IObservable<T> source, int frameCount, FrameCountType frameCountType) : base(source.IsRequiredSubscribeOnCurrentThread())
  15. {
  16. this.source = source;
  17. this.frameCount = frameCount;
  18. this.frameCountType = frameCountType;
  19. }
  20. protected override IDisposable SubscribeCore(IObserver<T> observer, IDisposable cancel)
  21. {
  22. return new TimeoutFrame(this, observer, cancel).Run();
  23. }
  24. class TimeoutFrame : OperatorObserverBase<T, T>
  25. {
  26. readonly TimeoutFrameObservable<T> parent;
  27. readonly object gate = new object();
  28. ulong objectId = 0ul;
  29. bool isTimeout = false;
  30. SingleAssignmentDisposable sourceSubscription;
  31. SerialDisposable timerSubscription;
  32. public TimeoutFrame(TimeoutFrameObservable<T> parent, IObserver<T> observer, IDisposable cancel) : base(observer, cancel)
  33. {
  34. this.parent = parent;
  35. }
  36. public IDisposable Run()
  37. {
  38. sourceSubscription = new SingleAssignmentDisposable();
  39. timerSubscription = new SerialDisposable();
  40. timerSubscription.Disposable = RunTimer(objectId);
  41. sourceSubscription.Disposable = parent.source.Subscribe(this);
  42. return StableCompositeDisposable.Create(timerSubscription, sourceSubscription);
  43. }
  44. IDisposable RunTimer(ulong timerId)
  45. {
  46. return UnityObservable.TimerFrame(parent.frameCount, parent.frameCountType)
  47. .Subscribe(new TimeoutFrameTick(this, timerId));
  48. }
  49. public override void OnNext(T value)
  50. {
  51. ulong useObjectId;
  52. bool timeout;
  53. lock (gate)
  54. {
  55. timeout = isTimeout;
  56. objectId++;
  57. useObjectId = objectId;
  58. }
  59. if (timeout) return;
  60. timerSubscription.Disposable = Disposable.Empty; // cancel old timer
  61. observer.OnNext(value);
  62. timerSubscription.Disposable = RunTimer(useObjectId);
  63. }
  64. public override void OnError(Exception error)
  65. {
  66. bool timeout;
  67. lock (gate)
  68. {
  69. timeout = isTimeout;
  70. objectId++;
  71. }
  72. if (timeout) return;
  73. timerSubscription.Dispose();
  74. try { observer.OnError(error); } finally { Dispose(); }
  75. }
  76. public override void OnCompleted()
  77. {
  78. bool timeout;
  79. lock (gate)
  80. {
  81. timeout = isTimeout;
  82. objectId++;
  83. }
  84. if (timeout) return;
  85. timerSubscription.Dispose();
  86. try { observer.OnCompleted(); } finally { Dispose(); }
  87. }
  88. class TimeoutFrameTick : IObserver<long>
  89. {
  90. readonly TimeoutFrame parent;
  91. readonly ulong timerId;
  92. public TimeoutFrameTick(TimeoutFrame parent, ulong timerId)
  93. {
  94. this.parent = parent;
  95. this.timerId = timerId;
  96. }
  97. public void OnCompleted()
  98. {
  99. }
  100. public void OnError(Exception error)
  101. {
  102. }
  103. public void OnNext(long _)
  104. {
  105. lock (parent.gate)
  106. {
  107. if (parent.objectId == timerId)
  108. {
  109. parent.isTimeout = true;
  110. }
  111. }
  112. if (parent.isTimeout)
  113. {
  114. try { parent.observer.OnError(new TimeoutException()); } finally { parent.Dispose(); }
  115. }
  116. }
  117. }
  118. }
  119. }
  120. }