从NetworkStream读取会破坏缓冲区

Question

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

然而，当我用ReadUntilClosedObservable2读取数据时，数据到达没有问题。

我想使用ReadUntilClosedObservable1，因为在ReadUntilClosedObservable2中反复读取流正在燃烧CPU。

如何使消息保持同步顺序?

更新:

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

我刚刚注意到readToEnd to在完成前一个工作之前一次又一次地被触发。难道不需要同步吗?如果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操作符:

• FromAsyncPattern:从异步呼叫签名生成"IObservable工厂"
• Observable.Defer:允许你使用上面的IObservable工厂来生成每个订阅者的可观察对象
• Observable.While:允许您在IObservable
上"Reinvoke until I say when"

(编辑:修改为使用NetworkStream示例)

(doubleEDIT:根据注释修改)

试试这个-除非我猜错了，它或多或少就是你想要的:

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()));
    }
}