ThrottleFirstFrame.cs 3.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123
  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 ThrottleFirstFrameObservable<T> : OperatorObservableBase<T>
  10. {
  11. readonly IObservable<T> source;
  12. readonly int frameCount;
  13. readonly FrameCountType frameCountType;
  14. public ThrottleFirstFrameObservable(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 ThrottleFirstFrame(this, observer, cancel).Run();
  23. }
  24. class ThrottleFirstFrame : OperatorObserverBase<T, T>
  25. {
  26. readonly ThrottleFirstFrameObservable<T> parent;
  27. readonly object gate = new object();
  28. bool open = true;
  29. SerialDisposable cancelable;
  30. ThrottleFirstFrameTick tick;
  31. public ThrottleFirstFrame(ThrottleFirstFrameObservable<T> parent, IObserver<T> observer, IDisposable cancel) : base(observer, cancel)
  32. {
  33. this.parent = parent;
  34. }
  35. public IDisposable Run()
  36. {
  37. tick = new ThrottleFirstFrameTick(this);
  38. cancelable = new SerialDisposable();
  39. var subscription = parent.source.Subscribe(this);
  40. return StableCompositeDisposable.Create(cancelable, subscription);
  41. }
  42. void OnNext()
  43. {
  44. lock (gate)
  45. {
  46. open = true;
  47. }
  48. }
  49. public override void OnNext(T value)
  50. {
  51. lock (gate)
  52. {
  53. if (!open) return;
  54. observer.OnNext(value);
  55. open = false;
  56. }
  57. var d = new SingleAssignmentDisposable();
  58. cancelable.Disposable = d;
  59. d.Disposable = UnityObservable.TimerFrame(parent.frameCount, parent.frameCountType)
  60. .Subscribe(tick);
  61. }
  62. public override void OnError(Exception error)
  63. {
  64. cancelable.Dispose();
  65. lock (gate)
  66. {
  67. try { observer.OnError(error); } finally { Dispose(); }
  68. }
  69. }
  70. public override void OnCompleted()
  71. {
  72. cancelable.Dispose();
  73. lock (gate)
  74. {
  75. try { observer.OnCompleted(); } finally { Dispose(); }
  76. }
  77. }
  78. // immutable, can share.
  79. class ThrottleFirstFrameTick : IObserver<long>
  80. {
  81. readonly ThrottleFirstFrame parent;
  82. public ThrottleFirstFrameTick(ThrottleFirstFrame parent)
  83. {
  84. this.parent = parent;
  85. }
  86. public void OnCompleted()
  87. {
  88. }
  89. public void OnError(Exception error)
  90. {
  91. }
  92. public void OnNext(long _)
  93. {
  94. lock (parent.gate)
  95. {
  96. parent.open = true;
  97. }
  98. }
  99. }
  100. }
  101. }
  102. }