使用 Subject 解耦 Observable 订阅和初始化

Question

我有一个公开IObservable 状态的 API。但是这个状态取决于一个底层的 observable 源，它必须通过Init 初始化。

我想做的是保护用户不必按正确的顺序做事：按照目前的情况，如果他们在执行Init 之前尝试订阅Status，他们会得到一个例外因为它们的源没有初始化。

所以我有了使用Subject 将两者解耦的天才想法：订阅我的Status 的外部用户只是订阅了主题，然后当他们调用Init 时，我订阅了底层服务使用我的主题。

代码中的想法

private ISubject<bool> _StatusSubject = new Subject<bool>();
public IObservable<bool> Status { get { return _StatusSubject; } }

public void Init() 
{
    _Connection = new Connection();
    Underlying.GetDeferredObservable(_Connection).Subscribe(_StatusSubject);
}

但是，从对虚拟项目的测试来看，问题在于初始化 通过订阅 Subject 来“唤醒”我的底层 Observable，即使还没有人订阅该主题。如果可能的话，我想避免这种情况，但我不确定如何...

（我也注意到 the received wisdom 说“一般规则是，如果你使用一个主题，那么你做错了什么”）

Answer 1

您似乎缺少的概念是如何知道某人何时开始收听并仅初始化您的基础来源。通常您使用Observable.Create 或其同级之一（Defer、Using、...）来执行此操作。

以下是没有Subject 的方法：

private IObservable<bool> _status = Observable.Defer(() =>
{
    _Connection = new Connection();
    return Underlying.GetDeferredObservable(_Connection);
};

public IObservable<bool> Status { get { return _status; } }

Defer 在有人真正订阅之前不会调用初始化代码。

但这有几个潜在的问题：

1. 每个观察者都会建立一个新的连接
2. 当观察者取消订阅时，连接不会被清理。

第二个问题很容易解决，所以让我们先解决这个问题。假设您的 Connection 是一次性的，在这种情况下您可以这样做：

private IObservable<bool> _status = Observable
    .Using(() => new Connection(),
           connection => Underlying.GetDeferredObservable(connection));

public IObservable<bool> Status { get { return _status; } }

通过这个迭代，每当有人订阅时，都会创建一个新的Connection 并传递给第二个 Lamba 方法来构造 observable。每当观察者取消订阅时，Connection 就是Disposed。如果Connection 不是IDisposable，那么您可以使用Disposable.Create(Action) 创建一个IDisposable，它将运行您需要运行的任何操作来清理连接。

你仍然有每个观察者创建一个新连接的问题。我们可以使用Publish 和RefCount 来解决这个问题：

private IObservable<bool> _status = Observable
    .Using(() => new Connection(),
           connection => Underlying.GetDeferredObservable(connection))
    .Publish()
    .RefCount();

public IObservable<bool> Status { get { return _status; } }

现在，当 first 观察者订阅时，将创建连接并订阅底层的 observable。后续观察者将共享连接并获取当前状态。当 last 观察者取消订阅时，连接将被释放并关闭一切。如果之后有其他观察者订阅，一切都会重新开始。

在底层，Publish 实际上是使用Subject 来共享单个可观察源。而RefCount 正在跟踪目前有多少观察者正在观察。

Answer 2

我可能在这里过于简单化了，但让我按要求使用Subject：

你的Thingy：

public class Thingy
{
    private BehaviorSubject<bool> _statusSubject = new BehaviorSubject<bool>(false);    
    public IObservable<bool> Status
    {
        get
        {
            return _statusSubject;
        }
    }

    public void Init()
    {
        var c = new object();
        new Underlying().GetDeferredObservable(c).Subscribe(_statusSubject);
    }
}

伪造的Underlying：

public class Underlying
{
    public IObservable<bool> GetDeferredObservable(object connection)
    {
        return Observable.DeferAsync<bool>(token => {
            return Task.Factory.StartNew(() => {
                Console.WriteLine("UNDERLYING ENGAGED");
                Thread.Sleep(1000);
                // Let's pretend there's some static on the line...
                return Observable.Return(true)
                    .Concat(Observable.Return(false))
                    .Concat(Observable.Return(true));
            }, token);
        });
    }
}

安全带：

void Main()
{
    var thingy = new Thingy();
    using(thingy.Status.Subscribe(stat => Console.WriteLine("Status:{0}", stat)))
    {
        Console.WriteLine("Waiting three seconds to Init...");
        Thread.Sleep(3000);
        thingy.Init();
        Console.ReadLine();
    }
}

输出：

Status:False
Waiting three seconds to Init...
UNDERLYING ENGAGED
Status:True
Status:False
Status:True

Answer 3

嗯，玩过这个，我不认为我可以只用一个主题。

尚未完成测试/尝试，但这是我目前提出的似乎可行的方法，但它并不能保护我免受主题问题的影响，因为我仍在内部使用它。

public class ObservableRouter<T> : IObservable<T>
{
    ISubject<T> _Subject = new Subject<T>();
    Dictionary<IObserver<T>, IDisposable> _ObserverSubscriptions 
                               = new Dictionary<IObserver<T>, IDisposable>();
    IObservable<T> _ObservableSource;
    IDisposable _SourceSubscription;

    //Note that this can happen before or after SetSource
    public IDisposable Subscribe(IObserver<T> observer)
    {
        _ObserverSubscriptions.Add(observer, _Subject.Subscribe(observer));
        IfReadySubscribeToSource();
        return Disposable.Create(() => UnsubscribeObserver(observer));
    }

    //Note that this can happen before or after Subscribe
    public void SetSource(IObservable<T> observable)
    {
        if(_ObserverSubscriptions.Count > 0 && _ObservableSource != null) 
                  throw new InvalidOperationException("Already routed!");
        _ObservableSource = observable;
        IfReadySubscribeToSource();
    }

    private void IfReadySubscribeToSource()
    {
        if(_SourceSubscription == null &&
           _ObservableSource != null && 
           _ObserverSubscriptions.Count > 0)
        {
            _SourceSubscription = _ObservableSource.Subscribe(_Subject);
        }
    }

    private void UnsubscribeObserver(IObserver<T> observer)
    {
        _ObserverSubscriptions[observer].Dispose();
        _ObserverSubscriptions.Remove(observer);
        if(_ObserverSubscriptions.Count == 0)
        {
            _SourceSubscription.Dispose();
            _SourceSubscription = null;
        }
    }
}