Skip to content
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

Jump to bottom

Save some allocations in ToTaskObserver. #643

Merged
merged 2 commits into from
Jun 25, 2018
Merged

Save some allocations in ToTaskObserver. #643

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fc4a964
Save some allocations in ToTaskObserver.
danielcweber Jun 22, 2018
711b7c0
Merge branch 'master' into ReviewToTask
danielcweber Jun 25, 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
26 changes: 11 additions & 15 deletions Rx.NET/Source/src/System.Reactive/Threading/Tasks/TaskObservableExtensions.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -286,43 +286,41 @@ public static Task<TResult> ToTask<TResult>(this IObservable<TResult> observable
return observable.ToTask(cancellationToken, state: null);
}

private sealed class ToTaskObserver<TResult> : IObserver<TResult>
private sealed class ToTaskObserver<TResult> : SafeObserver<TResult>
{
private readonly CancellationToken _ct;
private readonly IDisposable _disposable;
private readonly TaskCompletionSource<TResult> _tcs;
private readonly CancellationTokenRegistration _ctr = default(CancellationTokenRegistration);

private bool _hasValue;
private TResult _lastValue;

public ToTaskObserver(TaskCompletionSource<TResult> tcs, IDisposable disposable, CancellationToken ct)
public ToTaskObserver(TaskCompletionSource<TResult> tcs, CancellationToken ct)
{
_ct = ct;
_tcs = tcs;
_disposable = disposable;

if (ct.CanBeCanceled)
{
_ctr = ct.Register(Cancel);
_ctr = ct.Register(@this => ((ToTaskObserver<TResult>)@this).Cancel(), this);
}
}

public void OnNext(TResult value)
public override void OnNext(TResult value)
{
_hasValue = true;
_lastValue = value;
}

public void OnError(Exception error)
public override void OnError(Exception error)
{
_tcs.TrySetException(error);

_ctr.Dispose(); // no null-check needed (struct)
_disposable.Dispose();
Dispose();
}

public void OnCompleted()
public override void OnCompleted()
{
if (_hasValue)
{
Expand All @@ -334,12 +332,12 @@ public void OnCompleted()
}

_ctr.Dispose(); // no null-check needed (struct)
_disposable.Dispose();
Dispose();
}

private void Cancel()
{
_disposable.Dispose();
Dispose();
#if HAS_TPL46
_tcs.TrySetCanceled(_ct);
#else
Expand All @@ -364,9 +362,7 @@ public static Task<TResult> ToTask<TResult>(this IObservable<TResult> observable

var tcs = new TaskCompletionSource<TResult>(state);

var disposable = new SingleAssignmentDisposable();

var taskCompletionObserver = new ToTaskObserver<TResult>(tcs, disposable, cancellationToken);
var taskCompletionObserver = new ToTaskObserver<TResult>(tcs, cancellationToken);

//
// Subtle race condition: if the source completes before we reach the line below, the SingleAssigmentDisposable
Expand All @@ -382,7 +378,7 @@ public static Task<TResult> ToTask<TResult>(this IObservable<TResult> observable
// exception handling logic here for the reason explained above. We cannot afford to throw here
// and as a result never set the TaskCompletionSource, so we tunnel everything through here.
//
disposable.Disposable = observable.Subscribe/*Unsafe*/(taskCompletionObserver);
taskCompletionObserver.SetResource(observable.Subscribe/*Unsafe*/(taskCompletionObserver));
}
catch (Exception ex)
{
Expand Down