Асинхронное сетевое программирование с использованием Reactive Extensions

После выполнения некоторых (более или менее) «низкого уровня» асинхронном socketпрограммировании лет назад (в основе события асинхронной модели ^(EAP) мода) и в последнее время перемещения «вверх» к TcpListener(Asynchronous Programming Model ^(APM) ) , а затем пытаясь перейти к async/await(Асинхронный шаблон на основе задач ^(TAP) ), у меня почти всегда было то, что мне постоянно приходилось беспокоиться об этом «низкоуровневом слесарном деле». Итак, я подумал; почему бы не RXпопробовать ( Reactive Extensions ), поскольку он может более точно соответствовать моей проблемной области.

Много кода, который я пишу, имеет отношение ко многим клиентам, подключающимся через Tcp к моему приложению, которые затем запускают двустороннюю (асинхронную) связь. Клиент или сервер может в любой момент решить, что сообщение должно быть отправлено, и сделать это, так что это не ваша классическая request/responseустановка, а скорее двусторонняя «линия» в режиме реального времени, открытая для обеих сторон, для отправки того, что они хотят всякий раз, когда они хотят. (Если у кого-нибудь есть приличное имя, чтобы описать это, я был бы рад услышать это!).

«Протокол» отличается в зависимости от приложения (и на самом деле не имеет отношения к моему вопросу). У меня, однако, есть начальный вопрос:

Учитывая, что работает только один «сервер», но он должен отслеживать множество (обычно тысячи) соединений (например, клиентов), каждое из которых имеет (из-за отсутствия лучшего описания) свой собственный «конечный автомат», чтобы отслеживать свои внутренние штаты и т.д., какой подход вы бы предпочли? EAP / TAP / APM? Будет ли RX даже рассматриваться как вариант? Если нет, то почему?

Итак, мне нужно работать с Async, так как a) это не протокол запроса / ответа, поэтому я не могу иметь поток / клиента в вызове "ожидание сообщения" или вызов "блокирование сообщения" (однако, если отправка блокировка для этого клиента, только я мог жить с ним) и б) мне нужно обрабатывать много одновременных соединений. Я не вижу способа сделать это (надежно) с помощью блокировки вызовов.

Большинство моих приложений связаны с VoiP; будь то SIP сообщения от SIP ентов или УАТС (связанных) передачи сообщений из приложений , таких как FreeSwitch / OpenSIPS и т.д. , но вы можете, в его простейшей форме, попытайтесь представить себе «чат» сервер пытается обрабатывать много «чат» клиентов. Большинство протоколов основаны на тексте (ASCII).

Итак, после реализации множества различных комбинаций вышеупомянутых методов, я хотел бы упростить свою работу, создав объект, который я могу просто создать, рассказать о том, что IPEndpointслушать, и попросить его рассказать мне, когда происходит что-то интересное (что я обычно использовать события для, поэтому некоторые EAP обычно смешиваются с двумя другими методами). Класс не должен пытаться «понять» протокол; он должен просто обрабатывать входящие / исходящие строки. И, таким образом, следя за тем, чтобы RX надеялся, что это (в конце концов) упростит работу, я создал новую «скрипку» с нуля:

using System;
using System.Collections.Concurrent;
using System.Net;
using System.Net.Sockets;
using System.Reactive.Linq;
using System.Text;

class Program
{
    static void Main(string[] args)
    {
        var f = new FiddleServer(new IPEndPoint(IPAddress.Any, 8084));
        f.Start();
        Console.ReadKey();
        f.Stop();
        Console.ReadKey();
    }
}

public class FiddleServer
{
    private TcpListener _listener;
    private ConcurrentDictionary<ulong, FiddleClient> _clients;
    private ulong _currentid = 0;

    public IPEndPoint LocalEP { get; private set; }

    public FiddleServer(IPEndPoint localEP)
    {
        this.LocalEP = localEP;
        _clients = new ConcurrentDictionary<ulong, FiddleClient>();
    }

    public void Start()
    {
        _listener = new TcpListener(this.LocalEP);
        _listener.Start();
        Observable.While(() => true, Observable.FromAsync(_listener.AcceptTcpClientAsync)).Subscribe(
            //OnNext
            tcpclient =>
            {
                //Create new FSClient with unique ID
                var fsclient = new FiddleClient(_currentid++, tcpclient);
                //Keep track of clients
                _clients.TryAdd(fsclient.ClientId, fsclient);
                //Initialize connection
                fsclient.Send("connect\n\n");

                Console.WriteLine("Client {0} accepted", fsclient.ClientId);
            },
            //OnError
            ex =>
            {

            },
            //OnComplete
            () =>
            {
                Console.WriteLine("Client connection initialized");
                //Accept new connections
                _listener.AcceptTcpClientAsync();
            }
        );
        Console.WriteLine("Started");
    }

    public void Stop()
    {
        _listener.Stop();
        Console.WriteLine("Stopped");
    }

    public void Send(ulong clientid, string rawmessage)
    {
        FiddleClient fsclient;
        if (_clients.TryGetValue(clientid, out fsclient))
        {
            fsclient.Send(rawmessage);
        }
    }
}

public class FiddleClient
{
    private TcpClient _tcpclient;

    public ulong ClientId { get; private set; }

    public FiddleClient(ulong id, TcpClient tcpclient)
    {
        this.ClientId = id;
        _tcpclient = tcpclient;
    }

    public void Send(string rawmessage)
    {
        Console.WriteLine("Sending {0}", rawmessage);
        var data = Encoding.ASCII.GetBytes(rawmessage);
        _tcpclient.GetStream().WriteAsync(data, 0, data.Length);    //Write vs WriteAsync?
    }
}

_{^{Я знаю, что в этой "скрипке" есть небольшая деталь, специфичная для реализации; в этом случае я работаю с FreeSwitch ESL, поэтому "connect\n\n"при рефакторинге к более общему подходу следует удалить фиддл.}}

_{^{Я также знаю, что мне нужно реорганизовать анонимные методы в методы частного экземпляра класса Server; Я просто не уверен, какое соглашение (например, " OnSomething", например) использовать для их имен методов?}}

Это моя основа / отправная точка / основа (которая нуждается в некоторой «подстройке»). У меня есть несколько вопросов по этому поводу:

Смотри выше вопрос "1"
Я на правильном пути? Или мои «дизайнерские» решения несправедливы?
По параллелизму: справится ли это с тысячами клиентов (без разбора / обработки фактических сообщений)
Об исключениях: я не уверен, как получить исключения, созданные в клиентах "на сервер" ("RX-мудрый"); что было бы хорошим способом?
Теперь я могу получить любого подключенного клиента из моего класса сервера (используя его ClientId), если предположить, что я так или иначе выставляю клиентов и вызываю методы для них напрямую. Я также могу вызывать методы через класс Server (например, Send(clientId, rawmessage)метод (тогда как последний подход был бы «удобным» методом для быстрого получения сообщения другой стороне).
Я не совсем уверен, куда (и как) идти отсюда:
- а) мне нужно обрабатывать входящие сообщения; как бы это настроить? Я могу получить поток конечно, но где я буду обрабатывать получение полученных байтов? Я думаю, что мне нужен какой-то "ObservableStream" - то, на что я могу подписаться? Я бы положил это в FiddleClientили FiddleServer?
- б) Предполагая, что я хочу избежать использования события до тех пор, пока эти FiddleClient/ FiddleServerклассы не будут реализованы более конкретно, чтобы адаптировать их обработку протокола для конкретного приложения и т. д. с использованием более конкретных FooClient/ FooServerклассов: как мне перейти от получения данных в базовых «Fiddle-классах» к их более конкретные аналоги?

Статьи / ссылки, которые я уже прочитал / снял / использовал для справки:

c# .net tcp RobIII
источник

Взгляните на существующую библиотеку ReactiveSockets

Flagbug

Я не ищу библиотеки или ссылки (хотя для справки они ценятся), но для ввода / совета / помощи по моим вопросам и общей настройке. Я хочу научиться улучшать свой собственный код и лучше понять, в каком направлении это принимать, взвесить все за и против и т. Д. Не обращаясь к какой-либо библиотеке, бросьте ее и двигайтесь дальше. Я хочу извлечь уроки из этого опыта и получить больше опыта в программировании Rx / сети.

RobIII

Конечно, но так как библиотека с открытым исходным кодом, вы можете увидеть, как она там реализована

Flagbug

Конечно, но взгляд на исходный код не объясняет, почему некоторые дизайнерские решения были приняты. И поскольку я относительно новичок в Rx в сочетании с сетевым программированием, у меня недостаточно опыта, чтобы сказать, хороша ли эта библиотека, имеет ли дизайн смысл, были ли приняты правильные решения и даже если это подходит мне.

RobIII

Я думаю, что было бы хорошо переосмыслить сервер как активный член рукопожатия, поэтому сервер инициирует соединение, а не прослушивает соединения. Например, здесь: codeproject.com/Articles/20250/Reverse-Connection-Shell

Асинхронное сетевое программирование с использованием Reactive Extensions

Ответы: