0

ReadUntilClosedObservable1を使用してNetworkStreamからデータを読み取ると、読み取られたデータの一部のブロックがオーバーラップするように、返されたデータが破損します。

ただし、ReadUntilClosedObservable2を使用してデータを読み取ると、データは問題なく到着します。

ReadUntilClosedObservable2のストリームから繰り返し読み取ると、CPUが消費されるため、ReadUntilClosedObservable1を使用したいと思います。

メッセージを同期順に取得するにはどうすればよいですか？

アップデート：

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

1 に答える 1

0

いやいやいや…そんなことしないで…

Async + Rx は、機能するための直感的ではないセットアップの 1 つですが、試みているものよりもかなり単純です。キー ビットは、3 つの異なる Rx 演算子です。

  • FromAsyncPattern:IObservable非同期呼び出し署名から「ファクトリー」を生成します
  • Observable.DeferIObservable：上記のファクトリを使用して、サブスクライバーごとにオブザーバブルを生成できます
  • Observable.While：「いつ言うまで再起動する」ことができますIObservable

（編集：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()));
    }
}
于 2013-01-21T17:29:24.983 に答える