ObserveOn<TSource>
using System.Runtime.CompilerServices;
using System.Threading;
namespace System.Reactive.Concurrency
{
[System.Runtime.CompilerServices.NullableContext(1)]
[System.Runtime.CompilerServices.Nullable(0)]
internal static class ObserveOn<[System.Runtime.CompilerServices.Nullable(2)] TSource>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
1
})]
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)
{
return new ObserveOnObserverNew<TSource>(_scheduler, observer);
}
protected override void Run(ObserveOnObserverNew<TSource> sink)
{
sink.Run(_source);
}
}
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
1
})]
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)
{
return new ObserveOnObserverLongRunning<TSource>(_scheduler, observer);
}
protected override void Run(ObserveOnObserverLongRunning<TSource> sink)
{
sink.Run(_source);
}
}
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
0
})]
internal sealed class Context : Producer<TSource, Context._>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})]
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)
{
_context.OperationStarted();
SetUpstream(ObservableExtensions.SubscribeSafe<TSource>(source, (IObserver<TSource>)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, null);
}
[System.Runtime.CompilerServices.NullableContext(2)]
private void OnNextPosted(object value)
{
ForwardOnNext((TSource)value);
}
[System.Runtime.CompilerServices.NullableContext(2)]
private void OnErrorPosted(object error)
{
ForwardOnError((Exception)error);
}
[System.Runtime.CompilerServices.NullableContext(2)]
private void OnCompletedPosted(object ignored)
{
ForwardOnCompleted();
}
}
private readonly IObservable<TSource> _source;
private readonly SynchronizationContext _context;
public Context(IObservable<TSource> source, SynchronizationContext context)
{
_source = source;
_context = context;
}
[return: System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0
})]
protected override _ CreateSink(IObserver<TSource> observer)
{
return new _(_context, observer);
}
protected override void Run([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0
})] _ sink)
{
sink.Run(_source);
}
}
}
}