c# - ValveSubject : a queuing subject for Rx with built-in buffering, 打开/关闭操作

Question

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

此阀主题应支持打开/关闭流，并以 FIFO 顺序输出交付。阀门关闭时应缓冲输入值。

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

Answer 1

这里主要基于Buffer()和BehaviorSubject实现。行为主体跟踪阀门的打开/关闭状态。阀门的打开启动缓冲窗口，阀门的关闭关闭这些窗口。缓冲运算符的输出被“重新注入(inject)”到输入中(这样即使观察者自己也可以关闭阀门):

/// <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();
}

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

/// <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();
}

如评论中所述，这些主题不是“线程安全的”，因为在并发操作的情况下不再保证传递顺序。以与标准 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();
        }
    }
}