使主题同步不那么痛苦

Question

我想同步对BehaviorSubject<T> 的访问，所以我希望使用Subject.Synchronize。但是，我对这个界面有几个痛点，我想知道我是否错过了一种更令人满意的做事方式。

首先，我不得不同时存储原始主题和同步主题。这是因为我有时会在BehaviorSubject<T> 上使用Value 属性。也是因为Synchronize的返回值不是一次性的，所以我必须存储原始主题的一个实例才能正确处理。

其次，Synchronize的返回值为ISubject<TSource, TResult>，与ISubject<T>不兼容。

所以我最终得到了这样的代码：

public class SomeClass
{
    private readonly BehaviorSubject<string> something;
    private readonly ISubject<string, string> synchronizedSomething;

    public SomeClass()
    {
        this.something = new BehaviorSubject<string>(null);

        // having to provide the string type here twice is annoying
        this.synchronizedSomething = Subject.Synchronize<string, string>(this.something);
    }

    // must remember to use synchronizedSomething here (I forgot and had to edit my question again, showing how easy it is to screw this up)
    public IObservable<string> Something => this.synchronizedSomething.AsObservable();

    // could be called from any thread
    public void SomeMethod()
    {
        // do some work

        // also must be careful to use synchronizedSomething here
        this.synchronizedSomething.OnNext("some calculated value");
    }

    public void Dispose()
    {
        // synchronizedSomething is not disposable, so we must dispose the original subject
        this.something.Dispose();
    }
}

我是否缺少更清洁/更好的方法？为了清楚起见，我希望能够做的是这样的事情（伪代码）：

public class SomeClass
{
    private readonly IBehaviorSubject<string> something;

    public SomeClass()
    {
        this.something = new BehaviorSubject<string>(null).Synchronized();
    }

    public IObservable<string> Something => this.something.AsObservable();

    // could be called from any thread
    public void SomeMethod()
    {
        // do some work

        this.something.OnNext("some calculated value");
    }

    public void Dispose()
    {
        this.something.Dispose();
    }
}

Answer 1

我从您发布的代码示例中得到了一些注释

1. IBehaviorSubject<string> 不是 Rx.NET 中定义的类型。也许你的意思是ISubject<string>？
2. 您将null 作为默认值传递给BehaviorSubject<T>，通常当我看到这个时，用户实际上只是想要ReplaySubject<string>(1)。这取决于您在代码库中的某处是否有 Where(x=>x!=null) 或 Skip(1) 作为补偿行为。
3. 也许您想使用静态方法Subject.Synchronize(ISubject<T>) 而不是扩展方法.Synchronized()？

这可能是您上面示例代码的合适替代品。

public class SomeClass
{
    //Exposed as ISubject as I can't see usage of `Value` and `TryGetValue` are not present.
    private readonly ISubject<string> something;

    public SomeClass()
    {
        var source = new BehaviorSubject<string>(null);
        //Maybe this is actually what you want?
        //var source = new ReplaySubject<string>(1);
        this.something = Subject.Synchronize(source);
    }

    public IObservable<string> Something => this.something.AsObservable();

    // could be called from any thread
    public void SomeMethod()
    {
        // do some work

        this.something.OnNext("some calculated value");
    }
}

Answer 2

public class SynchronizeBehaviorSubject<T> : ISubject<T>, IDisposable
{
    private readonly BehaviorSubject<T> _source;
    private readonly ISubject<T> _sourceSync;

    public SynchronizeBehaviorSubject(BehaviorSubject<T> source)
    {
        _source = source;
        _sourceSync = source.Synchronize();
    }

    public void OnCompleted() => _sourceSync.OnCompleted();

    public void OnError(Exception error) => _sourceSync.OnError(error);

    public void OnNext(T value) => _sourceSync.OnNext(value);

    public IDisposable Subscribe(IObserver<T> observer) => _sourceSync.Subscribe(observer);

    public T Value => _source.Value;

    public bool HasObservers => _source.HasObservers;
    public void Dispose() => _source.Dispose();
    public bool IsDisposed => _source.IsDisposed;
}

public static class ReactiveEx
{

    public static ISubject<T> Synchronize<T>(this ISubject<T> source) =>
        Subject.Synchronize(source);

    public static SynchronizeBehaviorSubject<T> Synchronize<T>(this BehaviorSubject<T> source) =>
        new SynchronizeBehaviorSubject<T>(source);
}

用法：

private readonly SynchronizeBehaviorSubject<bool> _isBool 
                    = new BehaviorSubject(false).Synchronize();

bool isBool = _isBool.Value;

如果你不需要Value，甚至是ISubject<T>