Window<TSource, TWindowClosing>
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Subjects;
using System.Runtime.CompilerServices;
namespace System.Reactive.Linq.ObservableImpl
{
[System.Runtime.CompilerServices.NullableContext(1)]
[System.Runtime.CompilerServices.Nullable(0)]
internal static class Window<[System.Runtime.CompilerServices.Nullable(2)] TSource, [System.Runtime.CompilerServices.Nullable(2)] TWindowClosing>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
1,
0,
0
})]
internal sealed class Selector : Producer<IObservable<TSource>, Selector._>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
1
})]
internal sealed class _ : Sink<TSource, IObservable<TSource>>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1
})]
private sealed class WindowClosingObserver : SafeObserver<TWindowClosing>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})]
private readonly _ _parent;
public WindowClosingObserver([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})] _ parent)
{
_parent = parent;
}
public override void OnNext(TWindowClosing value)
{
_parent.CloseWindow(this);
}
public override void OnError(Exception error)
{
_parent.OnError(error);
}
public override void OnCompleted()
{
_parent.CloseWindow(this);
}
}
private readonly object _gate = new object();
private readonly AsyncLock _windowGate = new AsyncLock();
private readonly SerialDisposable _m = new SerialDisposable();
private readonly Func<IObservable<TWindowClosing>> _windowClosingSelector;
private Subject<TSource> _window;
[System.Runtime.CompilerServices.Nullable(2)]
private RefCountDisposable _refCountDisposable;
public _([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})] Selector parent, IObserver<IObservable<TSource>> observer)
: base(observer)
{
_windowClosingSelector = parent._windowClosingSelector;
_window = new Subject<TSource>();
}
public override void Run(IObservable<TSource> source)
{
CompositeDisposable compositeDisposable = new CompositeDisposable(2) {
_m
};
_refCountDisposable = new RefCountDisposable(compositeDisposable);
NextWindow();
compositeDisposable.Add(ObservableExtensions.SubscribeSafe<TSource>(source, (IObserver<TSource>)this));
_windowGate.Wait<_>(this, (Action<_>)delegate(_ this) {
this.CreateWindowClose();
});
SetUpstream(_refCountDisposable);
}
private void NextWindow()
{
WindowObservable<TSource> value = new WindowObservable<TSource>(_window, _refCountDisposable);
ForwardOnNext(value);
}
private void CreateWindowClose()
{
IObservable<TWindowClosing> source;
try {
source = _windowClosingSelector();
} catch (Exception error) {
lock (_gate) {
ForwardOnError(error);
}
return;
}
WindowClosingObserver windowClosingObserver = new WindowClosingObserver(this);
_m.Disposable = windowClosingObserver;
windowClosingObserver.SetResource(ObservableExtensions.SubscribeSafe<TWindowClosing>(source, (IObserver<TWindowClosing>)windowClosingObserver));
}
private void CloseWindow(IDisposable closingSubscription)
{
closingSubscription.Dispose();
lock (_gate) {
_window.OnCompleted();
_window = new Subject<TSource>();
NextWindow();
}
_windowGate.Wait<_>(this, (Action<_>)delegate(_ this) {
this.CreateWindowClose();
});
}
public override void OnNext(TSource value)
{
lock (_gate) {
_window.OnNext(value);
}
}
public override void OnError(Exception error)
{
lock (_gate) {
_window.OnError(error);
ForwardOnError(error);
}
}
public override void OnCompleted()
{
lock (_gate) {
_window.OnCompleted();
ForwardOnCompleted();
}
}
}
private readonly IObservable<TSource> _source;
private readonly Func<IObservable<TWindowClosing>> _windowClosingSelector;
public Selector(IObservable<TSource> source, Func<IObservable<TWindowClosing>> windowClosingSelector)
{
_source = source;
_windowClosingSelector = windowClosingSelector;
}
[return: System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})]
protected override _ CreateSink(IObserver<IObservable<TSource>> observer)
{
return new _(this, observer);
}
protected override void Run([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})] _ sink)
{
sink.Run(_source);
}
}
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
1,
0,
0
})]
internal sealed class Boundaries : Producer<IObservable<TSource>, Boundaries._>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
0,
1,
1,
1
})]
internal sealed class _ : Sink<TSource, IObservable<TSource>>
{
[System.Runtime.CompilerServices.Nullable(0)]
private sealed class WindowClosingObserver : IObserver<TWindowClosing>
{
[System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})]
private readonly _ _parent;
public WindowClosingObserver([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})] _ parent)
{
_parent = parent;
}
public void OnNext(TWindowClosing value)
{
lock (_parent._gate) {
_parent._window.OnCompleted();
_parent._window = new Subject<TSource>();
_parent.NextWindow();
}
}
public void OnError(Exception error)
{
_parent.OnError(error);
}
public void OnCompleted()
{
_parent.OnCompleted();
}
}
private readonly object _gate = new object();
private Subject<TSource> _window;
[System.Runtime.CompilerServices.Nullable(2)]
private RefCountDisposable _refCountDisposable;
public _(IObserver<IObservable<TSource>> observer)
: base(observer)
{
_window = new Subject<TSource>();
}
public void Run([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})] Boundaries parent)
{
CompositeDisposable compositeDisposable = new CompositeDisposable(2);
_refCountDisposable = new RefCountDisposable(compositeDisposable);
NextWindow();
compositeDisposable.Add(ObservableExtensions.SubscribeSafe<TSource>(parent._source, (IObserver<TSource>)this));
compositeDisposable.Add(ObservableExtensions.SubscribeSafe<TWindowClosing>(parent._windowBoundaries, (IObserver<TWindowClosing>)new WindowClosingObserver(this)));
SetUpstream(_refCountDisposable);
}
private void NextWindow()
{
WindowObservable<TSource> value = new WindowObservable<TSource>(_window, _refCountDisposable);
ForwardOnNext(value);
}
public override void OnNext(TSource value)
{
lock (_gate) {
_window.OnNext(value);
}
}
public override void OnError(Exception error)
{
lock (_gate) {
_window.OnError(error);
ForwardOnError(error);
}
}
public override void OnCompleted()
{
lock (_gate) {
_window.OnCompleted();
ForwardOnCompleted();
}
}
}
private readonly IObservable<TSource> _source;
private readonly IObservable<TWindowClosing> _windowBoundaries;
public Boundaries(IObservable<TSource> source, IObservable<TWindowClosing> windowBoundaries)
{
_source = source;
_windowBoundaries = windowBoundaries;
}
[return: System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})]
protected override _ CreateSink(IObserver<IObservable<TSource>> observer)
{
return new _(observer);
}
protected override void Run([System.Runtime.CompilerServices.Nullable(new byte[] {
1,
0,
0
})] _ sink)
{
sink.Run(this);
}
}
}
}