ValveSubject：Rx 的排队主题，具有内置缓冲，open/close 操作

Question

我经常运行遇到需要某种阀门结构来控制反应管道流量的情况。通常，在基于网络的应用程序中，我需要根据连接状态 open/close 请求流。

此阀门主题应支持opening/closing流，并以先进先出顺序输出交付。阀门关闭时应缓冲输入值。

A ConcurrentQueue 或 BlockingCollection 通常用于此类场景，但这会立即将线程引入图片中。我一直在寻找这个问题的纯反应性解决方案。

Answer 1

这里主要是基于Buffer()和BehaviorSubject的实现。行为主体跟踪阀门的 open/close 状态。阀门的打开开始缓冲 windows，阀门的关闭关闭那些 windows。缓冲运算符的输出是 "re-injected" 到输入（这样即使观察者自己也可以关闭阀门）：

/// <summary>
/// Subject offering Open() and Close() methods, with built-in buffering.
/// Note that closing the valve in the observer is supported.
/// </summary>
/// <remarks>As is the case with other Rx subjects, this class is not thread-safe, in that
/// order of elements in the output is indeterministic in the case of concurrent operation 
/// of Open()/Close()/OnNext()/OnError(). To guarantee strict order of delivery even in the 
/// case of concurrent access, <see cref="ValveSubjectExtensions.Synchronize{T}(NEXThink.Finder.Utils.Rx.IValveSubject{T})"/> can be used.</remarks>
/// <typeparam name="T">Elements type</typeparam>
public class ValveSubject<T> : IValveSubject<T>
{
    private enum Valve
    {
        Open,
        Closed
    }

    private readonly Subject<T> input = new Subject<T>();
    private readonly BehaviorSubject<Valve> valveSubject = new BehaviorSubject<Valve>(Valve.Open);
    private readonly Subject<T> output = new Subject<T>();

    public ValveSubject()
    {
        var valveOperations = valveSubject.DistinctUntilChanged();
        input.Buffer(
            bufferOpenings: valveOperations.Where(v => v == Valve.Closed),
            bufferClosingSelector: _ => valveOperations.Where(v => v == Valve.Open))
            .SelectMany(t => t).Subscribe(input);
        input.Where(t => valveSubject.Value == Valve.Open).Subscribe(output);
    }

    public bool IsOpen
    {
        get { return valveSubject.Value == Valve.Open; }
    }

    public bool IsClosed
    {
        get { return valveSubject.Value == Valve.Closed; }
    }

    public void OnNext(T value)
    {
        input.OnNext(value);
    }

    public void OnError(Exception error)
    {
        input.OnError(error);
    }

    public void OnCompleted()
    {
        output.OnCompleted();
        input.OnCompleted();
        valveSubject.OnCompleted();
    }

    public IDisposable Subscribe(IObserver<T> observer)
    {
        return output.Subscribe(observer);
    }

    public void Open()
    {
        valveSubject.OnNext(Valve.Open);
    }

    public void Close()
    {
        valveSubject.OnNext(Valve.Closed);
    }
}

public interface IValveSubject<T>:ISubject<T>
{
    void Open();

    void Close();
}

冲洗阀门的其他方法有时很有用，例如删除不再相关的剩余请求。这是一个建立在先例之上的实现，adapter-style:

/// <summary>
/// Subject with same semantics as <see cref="ValveSubject{T}"/>, but adding flushing out capability 
/// which allows clearing the valve of any remaining elements before closing.
/// </summary>
/// <typeparam name="T">Elements type</typeparam>
public class FlushableValveSubject<T> : IFlushableValveSubject<T>
{
    private readonly BehaviorSubject<ValveSubject<T>> valvesSubject = new BehaviorSubject<ValveSubject<T>>(new ValveSubject<T>());

    private ValveSubject<T> CurrentValve
    {
        get { return valvesSubject.Value; }
    }

    public bool IsOpen
    {
        get { return CurrentValve.IsOpen; }
    }

    public bool IsClosed
    {
        get { return CurrentValve.IsClosed; }
    }

    public void OnNext(T value)
    {
        CurrentValve.OnNext(value);
    }

    public void OnError(Exception error)
    {
        CurrentValve.OnError(error);
    }

    public void OnCompleted()
    {
        CurrentValve.OnCompleted();
        valvesSubject.OnCompleted();
    }

    public IDisposable Subscribe(IObserver<T> observer)
    {
        return valvesSubject.Switch().Subscribe(observer);
    }

    public void Open()
    {
        CurrentValve.Open();
    }

    public void Close()
    {
        CurrentValve.Close();
    }

    /// <summary>
    /// Discards remaining elements in the valve and reset the valve into a closed state
    /// </summary>
    /// <returns>Replayable observable with any remaining elements</returns>
    public IObservable<T> FlushAndClose()
    {
        var previousValve = CurrentValve;
        valvesSubject.OnNext(CreateClosedValve());
        var remainingElements = new ReplaySubject<T>();
        previousValve.Subscribe(remainingElements);
        previousValve.Open();
        return remainingElements;
    }

    private static ValveSubject<T> CreateClosedValve()
    {
        var valve = new ValveSubject<T>();
        valve.Close();
        return valve;
    }
}

public interface IFlushableValveSubject<T> : IValveSubject<T>
{
    IObservable<T> FlushAndClose();
}

如评论中所述，这些主题在并发操作的情况下不再保证传递顺序的意义上不是"thread-safe"。以与标准 Rx Subject、Subject.Synchronize() (https://msdn.microsoft.com/en-us/library/hh211643%28v=vs.103%29.aspx) 类似的方式，我们可以引入一些在阀门周围提供锁定的扩展：

public static class ValveSubjectExtensions
{
    public static IValveSubject<T> Synchronize<T>(this IValveSubject<T> valve)
    {
        return Synchronize(valve, new object());
    }

    public static IValveSubject<T> Synchronize<T>(this IValveSubject<T> valve, object gate)
    {
        return new SynchronizedValveAdapter<T>(valve, gate);
    }

    public static IFlushableValveSubject<T> Synchronize<T>(this IFlushableValveSubject<T> valve)
    {
        return Synchronize(valve, new object());
    }

    public static IFlushableValveSubject<T> Synchronize<T>(this IFlushableValveSubject<T> valve, object gate)
    {
        return new SynchronizedFlushableValveAdapter<T>(valve, gate);
    }
}

internal class SynchronizedValveAdapter<T> : IValveSubject<T>
{
    private readonly object gate;
    private readonly IValveSubject<T> valve;

    public SynchronizedValveAdapter(IValveSubject<T> valve, object gate)
    {
        this.valve = valve;
        this.gate = gate;
    }

    public void OnNext(T value)
    {
        lock (gate)
        {
            valve.OnNext(value);    
        }
    }

    public void OnError(Exception error)
    {
        lock (gate)
        {
            valve.OnError(error);
        }
    }

    public void OnCompleted()
    {
        lock (gate)
        {
            valve.OnCompleted();
        }
    }

    public IDisposable Subscribe(IObserver<T> observer)
    {
        return valve.Subscribe(observer);
    }

    public void Open()
    {
        lock (gate)
        {
            valve.Open();
        }
    }

    public void Close()
    {
        lock (gate)
        {
            valve.Close();
        }
    }
 }

 internal class SynchronizedFlushableValveAdapter<T> : SynchronizedValveAdapter<T>, IFlushableValveSubject<T>
 {
    private readonly object gate;
    private readonly IFlushableValveSubject<T> valve;

    public SynchronizedFlushableValveAdapter(IFlushableValveSubject<T> valve, object gate)
        : base(valve, gate)
    {
        this.valve = valve;
        this.gate = gate;
    }

    public IObservable<T> FlushAndClose()
    {
        lock (gate)
        {
            return valve.FlushAndClose();
        }
    }
}

Answer 2

这是我使用延迟运算符的实现：

source.delay(new Func1<Integer, Observable<Boolean>>() {
    @Override
    public Observable<Boolean> call(Integer integer) {
        return valve.filter(new Func1<Boolean, Boolean>() {
            @Override
            public Boolean call(Boolean aBoolean) {
                return aBoolean;
            }
        });
    }
})
.toBlocking()
.subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer integer) {
        System.out.println("out: " + integer);
    }
});

想法是将所有源排放延迟到 "valve opens"。如果阀门已经打开，则不会延迟物品的排放。

Rx valve gist

ValveSubject：Rx 的排队主题，具有内置缓冲，open/close 操作

ValveSubject: a queuing subject for Rx with built-in buffering, open/close operations

c#

system.reactive