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
|
// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
#if !NO_PERF && !NO_CDS
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Reactive.Disposables;
using System.Threading;
namespace System.Reactive.Linq.ObservableImpl
{
class GetEnumerator<TSource> : IEnumerator<TSource>, IObserver<TSource>
{
private readonly ConcurrentQueue<TSource> _queue;
private TSource _current;
private Exception _error;
private bool _done;
private bool _disposed;
private readonly SemaphoreSlim _gate;
private readonly SingleAssignmentDisposable _subscription;
public GetEnumerator()
{
_queue = new ConcurrentQueue<TSource>();
_gate = new SemaphoreSlim(0);
_subscription = new SingleAssignmentDisposable();
}
public IEnumerator<TSource> Run(IObservable<TSource> source)
{
//
// [OK] Use of unsafe Subscribe: non-pretentious exact mirror with the dual GetEnumerator method.
//
_subscription.Disposable = source.Subscribe/*Unsafe*/(this);
return this;
}
public void OnNext(TSource value)
{
_queue.Enqueue(value);
_gate.Release();
}
public void OnError(Exception error)
{
_error = error;
_subscription.Dispose();
_gate.Release();
}
public void OnCompleted()
{
_done = true;
_subscription.Dispose();
_gate.Release();
}
public bool MoveNext()
{
_gate.Wait();
if (_disposed)
throw new ObjectDisposedException("");
if (_queue.TryDequeue(out _current))
return true;
_error.ThrowIfNotNull();
Debug.Assert(_done);
_gate.Release(); // In the (rare) case the user calls MoveNext again we shouldn't block!
return false;
}
public TSource Current
{
get { return _current; }
}
object Collections.IEnumerator.Current
{
get { return _current; }
}
public void Dispose()
{
_subscription.Dispose();
_disposed = true;
_gate.Release();
}
public void Reset()
{
throw new NotSupportedException();
}
}
}
#endif
|
