2
0

Start.cs 2.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101
  1. using System;
  2. namespace UniRx.Operators
  3. {
  4. internal class StartObservable<T> : OperatorObservableBase<T>
  5. {
  6. readonly Action action;
  7. readonly Func<T> function;
  8. readonly IScheduler scheduler;
  9. readonly TimeSpan? startAfter;
  10. public StartObservable(Func<T> function, TimeSpan? startAfter, IScheduler scheduler)
  11. : base(scheduler == Scheduler.CurrentThread)
  12. {
  13. this.function = function;
  14. this.startAfter = startAfter;
  15. this.scheduler = scheduler;
  16. }
  17. public StartObservable(Action action, TimeSpan? startAfter, IScheduler scheduler)
  18. : base(scheduler == Scheduler.CurrentThread)
  19. {
  20. this.action = action;
  21. this.startAfter = startAfter;
  22. this.scheduler = scheduler;
  23. }
  24. protected override IDisposable SubscribeCore(IObserver<T> observer, IDisposable cancel)
  25. {
  26. if (startAfter != null)
  27. {
  28. return scheduler.Schedule(startAfter.Value, new StartObserver(this, observer, cancel).Run);
  29. }
  30. else
  31. {
  32. return scheduler.Schedule(new StartObserver(this, observer, cancel).Run);
  33. }
  34. }
  35. class StartObserver : OperatorObserverBase<T, T>
  36. {
  37. readonly StartObservable<T> parent;
  38. public StartObserver(StartObservable<T> parent, IObserver<T> observer, IDisposable cancel) : base(observer, cancel)
  39. {
  40. this.parent = parent;
  41. }
  42. public void Run()
  43. {
  44. var result = default(T);
  45. try
  46. {
  47. if (parent.function != null)
  48. {
  49. result = parent.function();
  50. }
  51. else
  52. {
  53. parent.action();
  54. }
  55. }
  56. catch (Exception exception)
  57. {
  58. try { observer.OnError(exception); }
  59. finally { Dispose(); }
  60. return;
  61. }
  62. OnNext(result);
  63. try { observer.OnCompleted(); }
  64. finally { Dispose(); }
  65. }
  66. public override void OnNext(T value)
  67. {
  68. try
  69. {
  70. base.observer.OnNext(value);
  71. }
  72. catch
  73. {
  74. Dispose();
  75. throw;
  76. }
  77. }
  78. public override void OnError(Exception error)
  79. {
  80. try { observer.OnError(error); }
  81. finally { Dispose(); }
  82. }
  83. public override void OnCompleted()
  84. {
  85. try { observer.OnCompleted(); }
  86. finally { Dispose(); }
  87. }
  88. }
  89. }
  90. }