Skip to content

Review join observer #647

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Jun 26, 2018
Merged

Review join observer #647

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c25b172
Save a SingleAssignmentDisposable.
danielcweber Jun 21, 2018
77e590a
Style fixes as suggested by .editorconfig.
danielcweber Jun 21, 2018
ae74858
Add a note about the copying of a list that is enumerated.
danielcweber Jun 26, 2018
946d9b5
Merge branch 'master' into ReviewJoinObserver
danielcweber Jun 26, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 22 additions & 23 deletions Rx.NET/Source/src/System.Reactive/Joins/JoinObserver.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,32 +16,31 @@ internal interface IJoinObserver : IDisposable

internal sealed class JoinObserver<T> : ObserverBase<Notification<T>>, IJoinObserver
{
private object gate;
private readonly IObservable<T> source;
private readonly Action<Exception> onError;
private List<ActivePlan> activePlans;
private object _gate;
private readonly IObservable<T> _source;
private readonly Action<Exception> _onError;
private readonly List<ActivePlan> _activePlans;
public Queue<Notification<T>> Queue { get; private set; }
private readonly SingleAssignmentDisposable subscription;
private bool isDisposed;
private IDisposable _subscription;
private bool _isDisposed;

public JoinObserver(IObservable<T> source, Action<Exception> onError)
{
this.source = source;
this.onError = onError;
_source = source;
_onError = onError;
Queue = new Queue<Notification<T>>();
subscription = new SingleAssignmentDisposable();
activePlans = new List<ActivePlan>();
_activePlans = new List<ActivePlan>();
}

public void AddActivePlan(ActivePlan activePlan)
{
activePlans.Add(activePlan);
_activePlans.Add(activePlan);
}

public void Subscribe(object gate)
{
this.gate = gate;
subscription.Disposable = source.Materialize().SubscribeSafe(this);
_gate = gate;
Disposable.SetSingle(ref _subscription, _source.Materialize().SubscribeSafe(this));
}

public void Dequeue()
Expand All @@ -51,18 +50,18 @@ public void Dequeue()

protected override void OnNextCore(Notification<T> notification)
{
lock (gate)
lock (_gate)
{
if (!isDisposed)
if (!_isDisposed)
{
if (notification.Kind == NotificationKind.OnError)
{
onError(notification.Exception);
_onError(notification.Exception);
return;
}

Queue.Enqueue(notification);
foreach (var activePlan in activePlans.ToArray())
foreach (var activePlan in _activePlans.ToArray()) // Working on a copy since _activePlans might change while iterating.
activePlan.Match();
}
}
Expand All @@ -78,22 +77,22 @@ protected override void OnCompletedCore()

internal void RemoveActivePlan(ActivePlan activePlan)
{
activePlans.Remove(activePlan);
if (activePlans.Count == 0)
_activePlans.Remove(activePlan);
if (_activePlans.Count == 0)
Dispose();
}

protected override void Dispose(bool disposing)
{
base.Dispose(disposing);

if (!isDisposed)
if (!_isDisposed)
{
if (disposing)
subscription.Dispose();
Disposable.TryDispose(ref _subscription);

isDisposed = true;
_isDisposed = true;
}
}
}
}
}