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
|
// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
using System.Collections.Generic;
namespace System.Reactive
{
class EventSource<T> : IEventSource<T>
{
private readonly IObservable<T> _source;
private readonly Dictionary<Delegate, Stack<IDisposable>> _subscriptions;
private readonly Action<Action<T>, /*object,*/ T> _invokeHandler;
public EventSource(IObservable<T> source, Action<Action<T>, /*object,*/ T> invokeHandler)
{
_source = source;
_invokeHandler = invokeHandler;
_subscriptions = new Dictionary<Delegate, Stack<IDisposable>>();
}
public event Action<T> OnNext
{
add
{
var gate = new object();
var isAdded = false;
var isDone = false;
var remove = new Action(() =>
{
lock (gate)
{
if (isAdded)
Remove(value);
else
isDone = true;
}
});
//
// [OK] Use of unsafe Subscribe: non-pretentious wrapper of an observable in an event; exceptions can occur during +=.
//
var d = _source.Subscribe/*Unsafe*/(
x => _invokeHandler(value, /*this,*/ x),
ex => { remove(); ex.Throw(); },
() => remove()
);
lock (gate)
{
if (!isDone)
{
Add(value, d);
isAdded = true;
}
}
}
remove
{
Remove(value);
}
}
private void Add(Delegate handler, IDisposable disposable)
{
lock (_subscriptions)
{
var l = new Stack<IDisposable>();
if (!_subscriptions.TryGetValue(handler, out l))
_subscriptions[handler] = l = new Stack<IDisposable>();
l.Push(disposable);
}
}
private void Remove(Delegate handler)
{
var d = default(IDisposable);
lock (_subscriptions)
{
var l = new Stack<IDisposable>();
if (_subscriptions.TryGetValue(handler, out l))
{
d = l.Pop();
if (l.Count == 0)
_subscriptions.Remove(handler);
}
}
if (d != null)
d.Dispose();
}
}
}
|
