130 lines
4.7 KiB
C#
130 lines
4.7 KiB
C#
// Licensed to the .NET Foundation under one or more agreements.
|
|
// The .NET Foundation licenses this file to you under the MIT License.
|
|
// See the LICENSE file in the project root for more information.
|
|
|
|
using System.Threading;
|
|
|
|
namespace System.Reactive.Concurrency
|
|
{
|
|
internal static class ObserveOn<TSource>
|
|
{
|
|
/// <summary>
|
|
/// The new ObserveOn operator run with an IScheduler in a lock-free manner.
|
|
/// </summary>
|
|
internal sealed class Scheduler : Producer<TSource, ObserveOnObserverNew<TSource>>
|
|
{
|
|
private readonly IObservable<TSource> _source;
|
|
private readonly IScheduler _scheduler;
|
|
|
|
public Scheduler(IObservable<TSource> source, IScheduler scheduler)
|
|
{
|
|
_source = source;
|
|
_scheduler = scheduler;
|
|
}
|
|
|
|
protected override ObserveOnObserverNew<TSource> CreateSink(IObserver<TSource> observer) => new(_scheduler, observer);
|
|
|
|
protected override void Run(ObserveOnObserverNew<TSource> sink) => sink.Run(_source);
|
|
}
|
|
|
|
/// <summary>
|
|
/// The new ObserveOn operator run with an ISchedulerLongRunning in a mostly lock-free manner.
|
|
/// </summary>
|
|
internal sealed class SchedulerLongRunning : Producer<TSource, ObserveOnObserverLongRunning<TSource>>
|
|
{
|
|
private readonly IObservable<TSource> _source;
|
|
private readonly ISchedulerLongRunning _scheduler;
|
|
|
|
public SchedulerLongRunning(IObservable<TSource> source, ISchedulerLongRunning scheduler)
|
|
{
|
|
_source = source;
|
|
_scheduler = scheduler;
|
|
}
|
|
|
|
protected override ObserveOnObserverLongRunning<TSource> CreateSink(IObserver<TSource> observer) => new(_scheduler, observer);
|
|
|
|
protected override void Run(ObserveOnObserverLongRunning<TSource> sink) => sink.Run(_source);
|
|
}
|
|
|
|
internal sealed class Context : Producer<TSource, Context._>
|
|
{
|
|
private readonly IObservable<TSource> _source;
|
|
private readonly SynchronizationContext _context;
|
|
|
|
public Context(IObservable<TSource> source, SynchronizationContext context)
|
|
{
|
|
_source = source;
|
|
_context = context;
|
|
}
|
|
|
|
protected override _ CreateSink(IObserver<TSource> observer) => new(_context, observer);
|
|
|
|
protected override void Run(_ sink) => sink.Run(_source);
|
|
|
|
internal sealed class _ : IdentitySink<TSource>
|
|
{
|
|
private readonly SynchronizationContext _context;
|
|
|
|
public _(SynchronizationContext context, IObserver<TSource> observer)
|
|
: base(observer)
|
|
{
|
|
_context = context;
|
|
}
|
|
|
|
public override void Run(IObservable<TSource> source)
|
|
{
|
|
//
|
|
// The interactions with OperationStarted/OperationCompleted below allow
|
|
// for test frameworks to wait until a whole sequence is observed, running
|
|
// asserts on a per-message level. Also, for ASP.NET pages, the use of the
|
|
// built-in synchronization context would allow processing to finished in
|
|
// its entirety before moving on with the page lifecycle.
|
|
//
|
|
_context.OperationStarted();
|
|
|
|
SetUpstream(source.SubscribeSafe(this));
|
|
}
|
|
|
|
protected override void Dispose(bool disposing)
|
|
{
|
|
if (disposing)
|
|
{
|
|
_context.OperationCompleted();
|
|
}
|
|
base.Dispose(disposing);
|
|
}
|
|
|
|
public override void OnNext(TSource value)
|
|
{
|
|
_context.Post(OnNextPosted, value);
|
|
}
|
|
|
|
public override void OnError(Exception error)
|
|
{
|
|
_context.Post(OnErrorPosted, error);
|
|
}
|
|
|
|
public override void OnCompleted()
|
|
{
|
|
_context.Post(OnCompletedPosted, state: null);
|
|
}
|
|
|
|
private void OnNextPosted(object? value)
|
|
{
|
|
ForwardOnNext((TSource)value!);
|
|
}
|
|
|
|
private void OnErrorPosted(object? error)
|
|
{
|
|
ForwardOnError((Exception)error!);
|
|
}
|
|
|
|
private void OnCompletedPosted(object? ignored)
|
|
{
|
|
ForwardOnCompleted();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|