1
0

ThrottleFrame.cs 3.9 KB

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