blob: e565a422fcda627dc9351d18304cd78697ef2cf6 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
|
using System.Reactive.Disposables;
namespace System.Reactive
{
class AutoDetachObserver<T> : ObserverBase<T>
{
private readonly IObserver<T> observer;
private readonly SingleAssignmentDisposable m = new SingleAssignmentDisposable();
public AutoDetachObserver(IObserver<T> observer)
{
this.observer = observer;
}
public IDisposable Disposable
{
set { m.Disposable = value; }
}
protected override void OnNextCore(T value)
{
//
// Safeguarding of the pipeline against rogue observers is required for proper
// resource cleanup. Consider the following example:
//
// var xs = Observable.Interval(TimeSpan.FromSeconds(1));
// var ys = <some random sequence>;
// var res = xs.CombineLatest(ys, (x, y) => x + y);
//
// The marble diagram of the query above looks as follows:
//
// xs -----0-----1-----2-----3-----4-----5-----6-----7-----8-----9---...
// | | | | | | | | |
// ys --------4--+--5--+-----+--2--+--1--+-----+-----+--0--+-----+---...
// | | | | | | | | | | | | | |
// v v v v v v v v v v v v v v
// res --------4--5--6--7-----8--5--6--5--6-----7-----8--7--8-----9---...
// |
// @#&
//
// Notice the free-threaded nature of Rx, where messages on the resulting sequence
// are produced by either of the two input sequences to CombineLatest.
//
// Now assume an exception happens in the OnNext callback for the observer of res,
// at the indicated point marked with @#& above. The callback runs in the context
// of ys, so the exception will take down the scheduler thread of ys. This by
// itself is a problem (that can be mitigated by a Catch operator on IScheduler),
// but notice how the timer that produces xs is kept alive.
//
// The safe-guarding code below ensures the acquired resources are disposed when
// the user callback throws.
//
var __noError = false;
try
{
observer.OnNext(value);
__noError = true;
}
finally
{
if (!__noError)
Dispose();
}
}
protected override void OnErrorCore(Exception exception)
{
try
{
observer.OnError(exception);
}
finally
{
Dispose();
}
}
protected override void OnCompletedCore()
{
try
{
observer.OnCompleted();
}
finally
{
Dispose();
}
}
protected override void Dispose(bool disposing)
{
base.Dispose(disposing);
if (disposing)
m.Dispose();
}
}
}
|
