Каков наиболее сукцинный способ записи функции GetMessages
с помощью Rx:
static void Main()
{
Socket socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
var messages = GetMessages(socket, IPAddress.Loopback, 4000);
messages.Subscribe(x => Console.WriteLine(x));
Console.ReadKey();
}
static IObservable<string> GetMessages(Socket socket, IPAddress addr, int port)
{
var whenConnect = Observable.FromAsyncPattern<IPAddress, int>(socket.BeginConnect, socket.EndConnect)(addr, port);
// now will receive a stream of messages
// each message is prefixed with an 4 bytes/Int32 indicating it length.
// the rest of the message is a string
// ????????????? Now What ?????????????
}
Простой сервер как драйвер для вышеуказанного примера: http://gist.github.com/452893#file_program.cs
Использование Rx для программирования сокетов
Я изучаю использование Reactive Extensions для какой-либо работы по программированию сокетов, которую я выполняю. Моей мотивацией для этого было бы то, что это каким-то образом упростит код. Будет ли это означать меньше кода, чем меньше вложенных в эти строки.
Однако пока это не выглядит так:
- Я не нашел очень много примеров использования Rx с сокетами
- пример
sЯ нашел не менее сложный, чем мой существующий код BeginXXXX, EndXXXX - Хотя
Observable
имеет методы расширения для FromAsyncPattern, это не распространяется наSocketEventArgs
Async API.
Текущее нерабочее решение
Вот что я до сих пор. Это не работает, он терпит неудачу с переполнением стека (heh). Я не понял семантики, чтобы создать IObservable
, который будет читать указанное количество байтов.
static IObservable<int> GetMessages(Socket socket, IPAddress addr, int port)
{
var whenConnect = Observable.FromAsyncPattern<IPAddress, int>(socket.BeginConnect, socket.EndConnect)(addr, port);
// keep reading until we get the first 4 bytes
byte[] buffer = new byte[1024];
var readAsync = Observable.FromAsyncPattern<byte[], int, int, SocketFlags, int>(socket.BeginReceive, socket.EndReceive);
IObservable<int> readBytes = null;
var temp = from totalRead in Observable.Defer(() => readBytes)
where totalRead < 4
select readAsync(buffer, totalRead, totalRead - 4, SocketFlags.None);
readBytes = temp.SelectMany(x => x).Sum();
var nowDoSomethingElse = readBytes.SkipUntil(whenConnect);
}