使用 RX 创建 IO 绑定 observable

Question

我有一个工作线程进行阻塞调用 (ReadFrame) 从套接字（IO 绑定）读取传入数据。 线程运行一个循环，将数据输入Subject， 消费者可以观察到。

private void ReadLoop()
{
    while (!IsDisposed)
    {
        var frame = _Socket.ReadFrame();
        _ReceivedFrames.OnNext(frame);
    }
}

我想知道是否有更 RX 的方式来做到这一点。

这是我做的一个尝试（玩具示例）：

var src = Observable
         .Repeat(Unit.Default)
         .Select(_ =>
         {
             Thread.Sleep(1000);              // simulated blocking ReadFrame call
             return "data read from socket";
         })
         .SubscribeOn(ThreadPoolScheduler.Instance) // avoid blocking when subscribing
         .ObserveOn(NewThreadScheduler.Default)     // spin up new thread (?)
         .Publish()
         .RefCount();

var d = src.Subscribe(s => s.Dump()); // simulated consumer

Console.ReadLine();  // simulated main running code

d.Dispose();  // tear down

我正在努力正确使用ObserveOn、SubscribeOn 和调度程序。 玩具示例似乎有效，但我不确定线程​​的生命周期是否得到正确管理。

阅读器线程是否因d.Dispose() 调用而关闭？
我是否需要创建一个新线程？
我应该改用Observable.Create 吗？如何？

下面是@Enigmativity 要求的附加信息：

ReadLoop() 方法是符合以下接口的类的一部分：

public interface ICanSocket : IDisposable
{
    IObservable<CanFrame> ReceivedFrames { get; }
    IObserver<CanFrame>   FramesToSend   { get; }
}

当父 ICanSocket 被释放时，它的成员 _Socket 被释放（关闭）。

Answer 1

执行此操作的最“Rxy”方式是使用Rxx，它具有执行异步 I/O 的可观察方法。

您的主要担忧似乎是：

• 订阅时，不要阻塞订阅者线程（也就是在后台线程上运行 I/O 线程）
• 当调用者退订时，停止 I/O 线程

解决这些问题的一种方法是只使用异步 Create 方法：

// just use Task.Run to "background" the work
var src = Observable
    .Create<CanFrame>((observer, cancellationToken) => Task.Run(() =>
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            var frame = _Socket.ReadFrame();
            if (frame == null) // end of stream?
            {
                // will send a Completed event
                return;
            }

            observer.OnNext(frame);
        }
    }));

var d = src.Subscribe(s => s.Dump());
Console.ReadLine();
d.Dispose();