c# - 从 NetworkStream 读取会破坏缓冲区

Question

使用 ReadUntilClosedObservable1 从 NetworkStream 读取数据时，返回的数据已损坏，就像某些读取数据块重叠一样。

但是，当我使用 ReadUntilClosedObservable2 读取数据时，数据到达时没有问题。

我想使用 ReadUntilClosedObservable1，因为在 ReadUntilClosedObservable2 中重复读取流正在消耗 CPU。

如何按同步顺序获取消息？

更新：

return Observable.Timer(TimeSpan.Zero, interval, TaskPoolScheduler.Default)
                    .SelectMany(_ => readToEnd)
                    .Where(dataChunk => dataChunk.Length > 0);

我只是注意到readToEndto 在完成上一份工作之前被一次又一次地解雇。不需要同步吗？如果Observable.Timer是问题，如果没有它，我怎样才能达到相同的效果，间隔阅读但不等待就开始？

public static IObservable<int> ReadObservable(this Stream stream, byte[] buffer
                                              ,int offset, int count)
{
    return stream.ReadAsync(buffer, offset, count)
                    .ToObservable();
}

public static IObservable<byte[]> ReadObservable(this Stream stream,
                                                 int bufferSize)
{
    var buffer = new byte[bufferSize];

    return stream.ReadObservable(buffer, 0, buffer.Length)
                    .Select(cbRead =>
                                {
                                    if (cbRead == 0)
                                    {
                                        return new byte[0];
                                    }

                                    if (cbRead == buffer.Length)
                                    {
                                        return buffer;
                                    }

                                    var dataChunk = new byte[cbRead];

                                    Buffer.BlockCopy(buffer, 0, dataChunk,
                                                     0, cbRead);

                                    return dataChunk;
                                });
}

public static IObservable<byte[]> ReadUntilClosedObservable1(this NetworkStream
                                     stream, int bufferSize, TimeSpan interval)
{
    var readToEnd = Observable.Defer(() => stream.ReadObservable(bufferSize))
                                .DoWhile(() => stream.DataAvailable)
                                .ToList()
                                .Select(dataChunks =>
                                    {
                                        var buffer = new List<byte>();

                                        foreach (var dataChunk in dataChunks)
                                        {
                                            buffer.AddRange(dataChunk);
                                        }

                                        return buffer.ToArray();
                                    });

    return Observable.Timer(TimeSpan.Zero, interval, TaskPoolScheduler.Default)
                        .SelectMany(_ => readToEnd)
                        .Where(dataChunk => dataChunk.Length > 0);
}

public static IObservable<byte[]> ReadUntilClosedObservable2(this Stream stream
                                                             ,int bufferSize)
{
    return Observable.Defer(() => stream.ReadObservable(bufferSize))
                        .Repeat()
                        .Where(dataChunk => dataChunk.Length > 0);
}

Answer 1

哦，不，不……不要那样做……

Async + Rx 是一种更...非直观的设置来开始工作，但它比您尝试的要简单得多。关键位是三个不同的 Rx 运算符：

• FromAsyncPatternIObservable：从异步调用签名生成“工厂”
• Observable.Defer：允许您使用上述IObservable工厂为每个订阅者生成 observables
• Observable.While：允许您“在我说什么时候重新调用”IObservable

（编辑：更改为使用NetworkStream示例）

（双重编辑：根据评论更改）

试试这个 - 除非我错过了我的猜测，它或多或少是你正在尝试的：

void Main()
{    
    // We'll feed this to listener
    var message = "Yo mamma said you like messages like this";

    var listenerTask = Task
        .Factory
        .StartNew(() => 
            {
                var bufferSize = 1024;
                var localhost = new IPAddress(new byte[]{127,0,0,1});
                var listener = new TcpListener(localhost, 11201);
                listener.Start();
                var incomingClient = listener.AcceptTcpClient();
                var clientStream = incomingClient.GetStream();
                // our buffered reader
                var observer = clientStream.ReadObservable(bufferSize);
                var compareBuffer = observer
                    // Take while we're getting data and the client
                    // is still connected
                    .TakeWhile(returnBuffer => returnBuffer.Length > 0 && 
                        incomingClient.Connected)
                    // In between read blocks, respond back to the client
                    // No need for fanciness here, just normal async writeback
                    .Do(returnBuffer => clientStream.BeginWrite(
                             returnBuffer, 
                             0, 
                             returnBuffer.Length, 
                             ar => clientStream.EndWrite(ar), 
                             null))
                    .ToEnumerable()
                    .SelectMany (returnBuffer => returnBuffer)
                    .ToArray();
                listener.Stop();
                Console.WriteLine(
                     "Listener thinks it was told... {0}", 
                     Encoding.ASCII.GetString(compareBuffer));
            });

    var clientTask = Task.Factory.StartNew(
        () => 
        {
            var client = new TcpClient();
            client.Connect("localhost", 11201);
            var random = new Random();
            var outStream = client.GetStream();
            var bytesToSend = Encoding.ASCII.GetBytes(message);
            foreach(byte toSend in bytesToSend)
            {
                // send a character over
                outStream.WriteByte(toSend);

                // Listener should parrot us...
                int goOn = outStream.ReadByte();
                if(goOn != toSend)
                {
                    Console.WriteLine(
                        "Huh. Listener echoed wrong. I said: {0}, they said {1}", 
                         toSend, 
                         goOn);
                    break;
                }
                Console.WriteLine("I said: {0}, they said {1}", toSend, goOn);

                // Take a little nap (simulate latency, etc)
                Thread.Sleep(random.Next(200));
            }
            client.Close();
        });

    Task.WaitAll(listenerTask, clientTask);
}


public static class Ext
{        
    public static IObservable<byte[]> ReadObservable(this Stream stream, int bufferSize)
    {        
        // to hold read data
        var buffer = new byte[bufferSize];
        // Step 1: async signature => observable factory
        var asyncRead = Observable.FromAsyncPattern<byte[], int, int, int>(
            stream.BeginRead, 
            stream.EndRead);
        return Observable.While(
            // while there is data to be read
            () => stream.CanRead, 
            // iteratively invoke the observable factory, which will
            // "recreate" it such that it will start from the current
            // stream position - hence "0" for offset
            Observable.Defer(() => asyncRead(buffer, 0, bufferSize))
                .Select(readBytes => buffer.Take(readBytes).ToArray()));
    }
}