C# – Async network programming using Reactive Extensions

After doing some (more-or-less) "low-level" async socket programming years ago (in an Event-based Asynchronous Pattern ^(EAP) fashion) and recently moving "up" to a TcpListener (Asynchronous Programming Model ^(APM)) and then trying to move to async/await (Task-based Asynchronous Pattern ^(TAP)) I've pretty much had it with keep having to bother with all this 'low level plumbing'. So I was figuring; why not give RX a go (Reactive Extensions) since it might fit more snugly to my problem domain.

A lot of code I write has to to with many clients connecting over Tcp to my application which then start a two-way (async) communication. The client or server may at any point decide a message needs to be sent and do so, so this is not your classic request/response setup but more of a real-time, two-way, "line" open to both parties to send whatever they want, whenever they want. (If anyone has a decent name to describe this I'd be glad to hear it!).

The "protocol" differs per application (and isn't really relevant to my question). I do, however have an initial question:

1. Given that only one "server" is running, but it has to keep track of many (usually thousands) of connections (e.g. clients) each having (for lack of a better description) their own "state machine" to keep track of their internal states etc, which approach would you prefer? EAP/TAP/APM? Would RX even be considered an option? If not, why?

So, I need to work Async since a) it's not a request/response protocol so I cannot have a thread/client in a "waiting for message"-blocking call or "sending message"-blocking call (however, if the send is blocking for that client only I could live with it) and b) I need to handle many concurrent connections. I see no way of doing this (reliably) using blocking calls.

Most of my applications are VoiP related; be it SIP messages from SIP cients or PBX (related) messaging from applications like FreeSwitch / OpenSIPS etc. but you can, in it's simplest form, try to imagine a "chat"server trying to handle many "chat"clients. Most protocols are text based (ASCII).

So, after having implemented many different permutations of aforementioned techniques I would like to simplify my work by creating an object that I can simply instantiate, tell it on which IPEndpoint to listen and have it tell me whenever something of interest is going on (which I usually use events for, so some EAP is usually mixed with the other two techniques). The class should not bother trying to 'understand' the protocol; it should merely handle incoming/outgoing strings. And thus, having my eye on RX hoping that would (in the end) simplify the work, I created a new "fiddle" from scratch:

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?
    }
}

_{^{I am aware that, in this "fiddle", there is a little implementation specific detail; in this case I'm working with FreeSwitch ESL so the "connect\n\n" in the fiddle should, when refactoring to a more generic approach, be removed.}}

_{^{I am also aware that I need to refactor the anonymous methods to private instance methods on the Server class; I'm just not sure what convention (e.g. "OnSomething" for example) to use for their method-names?}}

This is my basis/starting-point/foundation (which needs some "tweaking"). I have some questions about this:

1. See above question "1"
2. Am I on the right track? Or are my "design" decisions unjust?
3. Concurrency-wise: would this cope with thousands of clients (parsing/handling the actual messages aside)
4. On exceptions: I'm not sure how to get exceptions raised within clients "up" to the server ("RX-wise"); what would be a good way?
5. I can now get any connected client from my server class (using it's ClientId), assuming I expose the clients in one way or another, and call methods on them directly. I can also call methods via the Server class (for example, the Send(clientId, rawmessage) method (whereas the latter approach would be a "convenience" method for quickly getting a message to the other side).
6. I am not quite sure where (and how) to go from here:
   • a) I need to handle incoming messages; how would I set this up? I can get the stream ofcourse, but where would I handle retrieving the received bytes? I think I need some kind of "ObservableStream"-something I can subscribe to? Would I put this in the FiddleClient or FiddleServer?
   • b) Assuming I want to avoid using event until these FiddleClient/FiddleServer classes are implemented more specifically to tailor their application specific protocol handling etc. using more specific FooClient/FooServer classes: how would I go from getting the data in the underlying 'Fiddle'-classes to their more specific counterparts?

Articles/links I already read / skimmed / used for reference:

• Consume a Socket using Reactive Extension
• Creating and Subscribing to Simple Observable Sequences
• How to add or associate context to each ObservableTcpListener Connection/Session
• Asynchronous Programming with the Reactive Framework and the Task Parallel Library — Part 1, 2 and 3.
• Using Reactive Extensions (Rx) for socket programming practical?
• A .NET Rx Driven Web Server and .NET Rx Driven Web Server, Take 2
• Some Rx tutorials