每个订阅产生一个项目的 Observable - 每个处理程序一个事件

Question

假设我想公开一个 observable 立即通知，如果 Internet 连接现在可用，或者如果设备未连接到 Internet，则通知将在可用时推出：

IObservable<DateTime> InternetBecameAvailableSignalledOncePerSubscriber { get; }

此外，每个订阅应该只有一个通知，而不需要订阅者执行 .Take(1) 或类似的操作。

即依赖于 Internet 资源的客户端将使用这个 observable 现在或 Internet 可用时立即执行某些操作，但不会多次执行此操作 - 如果 Internet 变得不可用且可用，则不会再向该订阅者发出信号第二次..

如何使用 Reactive Extensions (Rx) 实现这一点？

• 请忽略互联网可以这样使用的有缺陷的概念，将其视为可访问 google.com 或您喜欢的任何风格的实现

Answer 1

这应该可以通过 Rx 轻松解决。问题是您如何知道互联网是否可用？它是基于其他消费者的订阅，是基于另一种方法（如Connect() 方法）还是某个正在推送给您的事件（如 WCF 通道状态更改事件）？

根据这个答案，您似乎只需要封装您的 Take(1) 和 Replay(1)。

public class IServiceClient
{
    IObservable<DateTime> LastConnnected { get; }
}

public class ServiceClient : IServiceClient, IDisposable
{
    private readonly IDisposable _connection;
    private readonly IObservable<DateTime> _lastConnnected;

    public ServiceClient()
    {
        //Question 1) where does the 'Connected' sequence come from i.e. what is it that tells you that you have internet connectivity?
        //Question 2) When should the subscription be made to 'Connected'? Here I cheat and do it in the ctor, not great.
        var connected = Connected.Replay(1)
                                .Where(isConnected=>isConnected)
                                .Take(1)
                                .Select(_=>DateTime.UtcNow);

        _lastConnnected = connected;
        _connection = connected.Connect();
    }

    public IObservable<DateTime> LastConnnected{ get {return _lastConnnected; } }

    public void Dispose()
    {
        _connection.Dispose();
    }
}

这确实给您留下了一些其他问题需要回答，例如是什么告诉您您是否有 Internet 连接以及对此的资源管理计划是什么？

更新代码

public interface IServiceClient
{
    IObservable<DateTime> LastConnnected { get; }
}

public class ServiceClient : IServiceClient, IDisposable
{
    private readonly IDisposable _connection;
    private readonly IObservable<bool> _lastConnnected;

    public ServiceClient(IObservable<ConnectionState> connectedStates)
    {
        var cachedStates = connectedStates.Select(state=>state.IsConnected).Replay(1);
        _lastConnnected = cachedStates;
        _connection = cachedStates.Connect();
    }

    public IObservable<DateTime> LastConnnected
    { 
        get 
        {
            return _lastConnnected.StartWith(IsConnected())
                                  .Where(isConnected=>isConnected)
                                  .Take(1)
                                  .Select(_=>DateTime.UtcNow); 
        } 
    }

    //....

    public void Dispose()
    {
        _connection.Dispose();
    }
}