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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
#if !NO_PERF
using System;
using System.Collections.Generic;
using System.Reactive;
using System.Reactive.Threading;
using System.Threading;
namespace System.Reactive.Linq.Observαble
{
class Collect<TSource, TResult> : PushToPullAdapter<TSource, TResult>
{
private readonly Func<TResult> _getInitialCollector;
private readonly Func<TResult, TSource, TResult> _merge;
private readonly Func<TResult, TResult> _getNewCollector;
public Collect(IObservable<TSource> source, Func<TResult> getInitialCollector, Func<TResult, TSource, TResult> merge, Func<TResult, TResult> getNewCollector)
: base(source)
{
_getInitialCollector = getInitialCollector;
_merge = merge;
_getNewCollector = getNewCollector;
}
protected override PushToPullSink<TSource, TResult> Run(IDisposable subscription)
{
var sink = new _(this, subscription);
sink.Initialize();
return sink;
}
class _ : PushToPullSink<TSource, TResult>
{
private readonly Collect<TSource, TResult> _parent;
public _(Collect<TSource, TResult> parent, IDisposable subscription)
: base(subscription)
{
_parent = parent;
}
private object _gate;
private TResult _collector;
private bool _hasFailed;
private Exception _error;
private bool _hasCompleted;
private bool _done;
public void Initialize()
{
_gate = new object();
_collector = _parent._getInitialCollector();
}
public override void OnNext(TSource value)
{
lock (_gate)
{
try
{
_collector = _parent._merge(_collector, value);
}
catch (Exception ex)
{
_error = ex;
_hasFailed = true;
base.Dispose();
}
}
}
public override void OnError(Exception error)
{
base.Dispose();
lock (_gate)
{
_error = error;
_hasFailed = true;
}
}
public override void OnCompleted()
{
base.Dispose();
lock (_gate)
{
_hasCompleted = true;
}
}
public override bool TryMoveNext(out TResult current)
{
lock (_gate)
{
if (_hasFailed)
{
current = default(TResult);
_error.Throw();
}
else
{
if (_hasCompleted)
{
if (_done)
{
current = default(TResult);
return false;
}
current = _collector;
_done = true;
}
else
{
current = _collector;
try
{
_collector = _parent._getNewCollector(current);
}
catch
{
base.Dispose();
throw;
}
}
}
return true;
}
}
}
}
}
#endif
|
