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
|
// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
using System.Threading;
namespace System.Reactive
{
/// <summary>
/// Abstract base class for implementations of the IObserver<T> interface.
/// </summary>
/// <remarks>This base class enforces the grammar of observers where OnError and OnCompleted are terminal messages.</remarks>
/// <typeparam name="T">The type of the elements in the sequence.</typeparam>
public abstract class ObserverBase<T> : IObserver<T>, IDisposable
{
private int isStopped;
/// <summary>
/// Creates a new observer in a non-stopped state.
/// </summary>
protected ObserverBase()
{
isStopped = 0;
}
/// <summary>
/// Notifies the observer of a new element in the sequence.
/// </summary>
/// <param name="value">Next element in the sequence.</param>
public void OnNext(T value)
{
if (isStopped == 0)
OnNextCore(value);
}
/// <summary>
/// Implement this method to react to the receival of a new element in the sequence.
/// </summary>
/// <param name="value">Next element in the sequence.</param>
/// <remarks>This method only gets called when the observer hasn't stopped yet.</remarks>
protected abstract void OnNextCore(T value);
/// <summary>
/// Notifies the observer that an exception has occurred.
/// </summary>
/// <param name="error">The error that has occurred.</param>
/// <exception cref="ArgumentNullException"><paramref name="error"/> is null.</exception>
public void OnError(Exception error)
{
if (error == null)
throw new ArgumentNullException("error");
if (Interlocked.Exchange(ref isStopped, 1) == 0)
{
OnErrorCore(error);
}
}
/// <summary>
/// Implement this method to react to the occurrence of an exception.
/// </summary>
/// <param name="error">The error that has occurred.</param>
/// <remarks>This method only gets called when the observer hasn't stopped yet, and causes the observer to stop.</remarks>
[System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Naming", "CA1716:IdentifiersShouldNotMatchKeywords", MessageId = "Error", Justification = "Same name as in the IObserver<T> definition of OnError in the BCL.")]
protected abstract void OnErrorCore(Exception error);
/// <summary>
/// Notifies the observer of the end of the sequence.
/// </summary>
public void OnCompleted()
{
if (Interlocked.Exchange(ref isStopped, 1) == 0)
{
OnCompletedCore();
}
}
/// <summary>
/// Implement this method to react to the end of the sequence.
/// </summary>
/// <remarks>This method only gets called when the observer hasn't stopped yet, and causes the observer to stop.</remarks>
protected abstract void OnCompletedCore();
/// <summary>
/// Disposes the observer, causing it to transition to the stopped state.
/// </summary>
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
/// <summary>
/// Core implementation of IDisposable.
/// </summary>
/// <param name="disposing">true if the Dispose call was triggered by the IDisposable.Dispose method; false if it was triggered by the finalizer.</param>
protected virtual void Dispose(bool disposing)
{
if (disposing)
{
isStopped = 1;
}
}
internal bool Fail(Exception error)
{
if (Interlocked.Exchange(ref isStopped, 1) == 0)
{
OnErrorCore(error);
return true;
}
return false;
}
}
}
|
