使用TCPCLIENT和反应性扩展从流中读取持续的地下语

Question

考虑以下代码：

internal class Program
{
    private static void Main(string[] args)
    {
        var client = new TcpClient();
        client.ConnectAsync("localhost", 7105).Wait();
        var stream = client.GetStream();
        var observable = stream.ReadDataObservable().Repeat();
        var s = from d in observable.Buffer(4)
                let headerLength = IPAddress.NetworkToHostOrder(BitConverter.ToInt16(d.ToArray(), 2))
                let b = observable.Take(headerLength)
                select b.ToEnumerable().ToArray();
        s.Subscribe(a => Console.WriteLine("{0}", a));
        Console.ReadLine();
    }
}
public static class Extensions
{
    public static IObservable<byte> ReadDataObservable(this Stream stream)
    {
        return Observable.Defer(async () =>
        {
            var buffer = new byte[1024];
            var readBytes = await stream.ReadAsync(buffer, 0, buffer.Length);
            return buffer.Take(readBytes).ToObservable();
        });
    }
}

基本上，我想用反应性扩展来解析我收到的消息。消息的标题使用缓冲区（4）正确解析，我获得了剩余的消息的长度。出现的问题是，当我进行stream.take（headerLength）时，代码会重新评估整个"链"，并试图从流中获取新消息，而不是返回其余的字节，这些字节已经从流中读过。更确切地说，第一个readasync（...）返回38个字节，缓冲区（4）返回其中的第四个，可观察到的。带有readasync的消息。

问题是，如何确保可观察到的（headerLength）接收已经读取的34个字节，而不是尝试从流中读取新消息？我已经四处寻找解决方案，但是我真的不知道如何实现这一目标。

编辑：此解决方案（使用反应性扩展（RX）用于套接字编程实用吗？）不是我想要的。这并不是读取流中可用的所有可用的东西（直至缓冲），而是从中延续的副流。对我来说，这个解决方案似乎不是从流中阅读的一种非常有效的方法，因此我的问题。

Answer 1

这种方法无法正常工作。问题是您使用可观察的方式。Buffer不会读取4个字节并退出，它将不断读取4个字节块。Take形成第二个订阅，该订阅将读取重叠字节。您会发现将流直接解析到消息中要容易得多。

以下代码也付出了很大的努力来正确清理。

假设您的Message就是这样（添加了用于测试的ToString）：

public class Message
{
    public byte[] PayLoad;
    public override string ToString()
    {
        return Encoding.UTF8.GetString(PayLoad);
    }
}

您已经获得了Stream，然后您可以按照以下方式解析它。首先，一种从流中读取确切数量字节的方法：

public async static Task ReadExactBytesAsync(
    Stream stream, byte[] buffer, CancellationToken ct)
{
    var count = buffer.Length;
    var totalBytesRemaining = count;
    var totalBytesRead = 0;
    while (totalBytesRemaining != 0)
    {
        var bytesRead = await stream.ReadAsync(
            buffer, totalBytesRead, totalBytesRemaining, ct);
        ct.ThrowIfCancellationRequested();
        totalBytesRead += bytesRead;
        totalBytesRemaining -= bytesRead;
    }
}

然后将流转换为 IObservable<Message>：

public static IObservable<Message> ReadMessages(
    Stream sourceStream,
    IScheduler scheduler = null)
{
    int subscribed = 0;
    scheduler = scheduler ?? Scheduler.Default;
    return Observable.Create<Message>(o =>
    {
        // first check there is only one subscriber
        // (multiple stream readers would cause havoc)
        int previous = Interlocked.CompareExchange(ref subscribed, 1, 0);
        if (previous != 0)
            o.OnError(new Exception(
                "Only one subscriber is allowed for each stream."));
        // we will return a disposable that cleans
        // up both the scheduled task below and
        // the source stream
        var dispose = new CompositeDisposable
        {
            Disposable.Create(sourceStream.Dispose)
        };
        // use async scheduling to get nice imperative code
        var schedule = scheduler.ScheduleAsync(async (ctrl, ct) =>
        {
            // store the header here each time
            var header = new byte[4];
            // loop until cancellation requested
            while (!ct.IsCancellationRequested)
            {                        
                try
                {
                    // read the exact number of bytes for a header
                    await ReadExactBytesAsync(sourceStream, header, ct);
                }
                catch (OperationCanceledException)
                {
                    throw;
                }
                catch (Exception ex)
                {
                    // pass through any problem in the stream and quit
                    o.OnError(new InvalidDataException("Error in stream.", ex));
                    return;
                }                   
                ct.ThrowIfCancellationRequested();
                var bodyLength = IPAddress.NetworkToHostOrder(
                    BitConverter.ToInt16(header, 2));
                // create buffer to read the message
                var payload = new byte[bodyLength];
                // read exact bytes as before
                try
                {
                    await ReadExactBytesAsync(sourceStream, payload, ct);
                }
                catch (OperationCanceledException)
                {
                    throw;
                }
                catch (Exception ex)
                {
                    o.OnError(new InvalidDataException("Error in stream.", ex));
                    return;
                }
                // create a new message and send it to client
                var message = new Message { PayLoad = payload };
                o.OnNext(message);
            }
            // wrap things up
            ct.ThrowIfCancellationRequested();
            o.OnCompleted();
        });
        // return the suscription handle
        dispose.Add(schedule);
        return dispose;
    });
}

编辑 - 我使用的非常骇人听闻的测试代码：

private static void Main(string[] args)
{
    var listener = new TcpListener(IPAddress.Any, 12873);
    listener.Start();
    var listenTask = listener.AcceptTcpClientAsync();
    listenTask.ContinueWith((Task<TcpClient> t) =>
    {
        var client = t.Result;
        var stream = client.GetStream();
        const string messageText = "Hello World!";                
        var body = Encoding.UTF8.GetBytes(messageText);                
        var header = BitConverter.GetBytes(
            IPAddress.HostToNetworkOrder(body.Length));
        for (int i = 0; i < 5; i++)
        {
            stream.Write(header, 0, 4);
            stream.Write(body, 0, 4);
            stream.Flush();
            // deliberate nasty delay
            Thread.Sleep(2000);
            stream.Write(body, 4, body.Length - 4);
            stream.Flush();
        }
        stream.Close();
        listener.Stop();
    });

    var tcpClient = new TcpClient();
    tcpClient.Connect(new IPEndPoint(IPAddress.Loopback, 12873));
    var clientStream = tcpClient.GetStream();
    ReadMessages(clientStream).Subscribe(
        Console.WriteLine,
        ex => Console.WriteLine("Error: " + ex.Message),
        () => Console.WriteLine("Done!"));
    Console.ReadLine();
}

总结

您需要考虑设置读取的超时，以防服务器死亡，并且服务器应发送某种"结束消息"。目前，此方法将不断尝试接收字节。由于您还没有指定它，我还没有包含类似的内容 - 但是如果您这样做，那么我只写了 break in the while循环会导致OnCompleted发送。

Answer 2

我想这里需要什么是qActive：基于RX.NET的可疑反应性TCP服务器提供商

服务器

Observable
    .Interval(TimeSpan.FromSeconds(1))
    .ServeQbservableTcp(new IPEndPoint(IPAddress.Loopback, 3205))
    .Subscribe();

客户端

var datasourceAddress = new IPEndPoint(IPAddress.Loopback, 3205);
var datasource = new TcpQbservableClient<long>(datasourceAddress);
(
     from value in datasource.Query()
     //The code below is actually executed on the server
     where value <= 5 || value >= 8
     select value
)
.Subscribe(Console.WriteLine);

对此有什么想法是，客户可以说出他们收到的数据的频率和频率，并且服务器仍然可以限制和控制何时，何时和返回多少数据。

有关此https://github.com/rxdave/qactive

的更多信息

另一个博客。样本

https://sachabarbs.wordpress.com/2016/12/23/rx-over-the-wire/