小编shi*_*ipr的帖子

经过对StackOverflow的大量努力和研究 - 其中大部分已经过时,因为Reactive Extensions代码最近发生了变化 - 我终于能够消除此Observable方法从套接字读取数据的所有编译错误,我理解这一点代码比我最初做的要好得多.但还不完全.有人可以用英语回复给我,回答两三个问题吗？

是从这种方法中提取的缓冲数据(或者如果我有错误,应该怎么做？)？是否有不再需要它的部分？虽然我真的喜欢与业务代码分离,并且只用一两种方法保留所有套接字代码,但有没有更好的方法(解耦和可读)？

    public static IObservable<int> WhenDataReceived(this Socket socket, int byteCount, SocketFlags flags = SocketFlags.None)
    {
        Contract.Requires(byteCount > 0);

        return Observable.Create<int>(
            observer =>
            {
                byte[] buffer = new byte[byteCount];
                int remainder = byteCount;
                bool shutdown = false;

                return Observable.Defer<int>(() =>
                        Task.Factory.FromAsync<int>(socket.BeginReceive(buffer, buffer.Length - remainder, remainder, flags,
                        (result) =>
                        {
                            var read = (int)result.AsyncState;
                            remainder -= read;

                            if (read == 0)
                                shutdown = true;
                        },
                        null), socket.EndReceive).ToObservable())
                    .TakeWhile(_ => remainder > 0 && !shutdown)
                    .TakeLast(1)
                    .Subscribe(
                        observer.OnNext, …