使用RX observable创建“去抖动”效果

Question

我有一个FTP客户端，我想要连接到FTP服务器，除非（比方说）一分钟没有活动。我想使用Observable来做这件事。

这是一个非常愚蠢的Linqpad脚本，演示了这个概念：

async Task Main()
{
    var client = new Client();
    client.Connect();

    var debounce = new Subject<int>();
    debounce
        .Throttle(TimeSpan.FromSeconds(1))
        .Subscribe(eventNumber => client.Disconnect(eventNumber));

    // Something uses the FTP client
    debounce.OnNext(1);
    await Task.Delay(200);

    // Something else uses the FTP client
    debounce.OnNext(2);
    await Task.Delay(300);

    // No activity, the client will disconnect
    await Task.Delay(1000);
}

public class Client
{
    public void Connect() => Console.WriteLine("Connected");
    public void Disconnect(int eventNumber) => Console.WriteLine($"Disconnected: {eventNumber}");
}

这很有效 - 客户端在事件“2”之后断开连接。

问题：有更好的方法吗？或者更准确地说，如果不使用Subject，有没有更好的方法呢？

编辑

这是一个更加充实的类本 - 有效地，它订阅了一个observable，它会告诉它一些需要下载的文件;如果没有文件通过一些超时，那么我希望客户端断开连接。

public class MyClassThatDownloadsViaFtp
{
    private IObserver<Unit> _debouncer;
    private FtpClient _client;

    public MyClassThatDownloadsViaFtp(IObservable<FileToDownload> filesToDownloadViaFtp)
    {
        filesToDownloadViaFtp.Subscribe(DownloadFileViaFtp);

        // Disconnect after a minute of activity
        _debouncer = new Subject<Unit>();
        _debouncer
            .Throttle(TimeSpan.FromMinutes(1))
            .Subscribe(_ => DisconnectFtpClient());
    }

    public void DownloadFileViaFtp(FileToDownload file)
    {
        if (_client == null) _client = ConnectFtpClient();

        // Signal that the client is doing some work to prevent disconnect
        _debouncer.OnNext(Unit.Default);
        _client.Download(file.PathOnFtpServer);
    }

    // implementation irrelivent
    private FtpClient ConnectFtpClient() => new FtpClient();
    private FtpClient DisconnectFtpClient() => _client = null;
}

我想通了，因为我有一个源流，它可能更容易限制它以达到相同的效果（如下）;但是，在我没有可以限制的源流的情况下，我仍然想知道最好的方法。

public class MyClassThatDownloadsViaFtp 
{
    private FtpClient _client;

    public MyClassThatDownloadsViaFtp(IObservable<FileToDownload> filesToDownloadViaFtp)
    {
        filesToDownloadViaFtp
            .Select(DownloadFileViaFtp)
            .Throttle(TimeSpan.FromMinutes(1))
            .Subscribe(_ => DisconnectFtpClient());
    }

    public Unit DownloadFileViaFtp(FileToDownload file)
    {
        if (_client == null) _client = ConnectFtpClient();
        _client.Download(file.PathOnFtpServer);

        return Unit.Default;
    }

    // implementation irrelivent
    private FtpClient ConnectFtpClient() => new FtpClient();
    private FtpClient DisconnectFtpClient() => _client = null; 
}

Answer 1

你基本上回答了你的问题：

public MyClassThatDownloadsViaFtp(IObservable<FileToDownload> filesToDownloadViaFtp)
{
    filesToDownloadViaFtp
        .Select(DownloadFileViaFtp)
        .Throttle(TimeSpan.FromMinutes(1))
        .Subscribe(_ => DisconnectFtpClient());
}

如果你没有像filesToDownloadViaFtp这样的方便的流，那么从Observable.Create或Observable.FromEvent或Observable.FromEventPattern等创建一个。

一个狡辩：Select理想情况下运行没有副作用，DownloadFileViaFtp非常有副作用。在Subscribe电话中，副作用最好。