Awaiting server responses in an async way to different calls at the same time

Question

Rx would probably make your life a lot easier. I would also imagine that it would reduce some of the potential race-conditions in your code and also just end up with a lot less plumbing style code (ie the code to maintain the cache).

If I start off by taking what I think are the key elements of the code (from https://github.com/schjan/RailNet/blob/master/src/RailNet.Clients.Ecos/Basic/NachrichtenDispo.cs) and pull them out into methods I see this.

private bool HasBeginAndEnd(string[] message)
{
    bool isValid = true;

    if (!message[0].StartsWith("<") || !message[0].EndsWith(">"))
        isValid = false;

    if (!message.Last().StartsWith("<END"))
        isValid = false;

    return isValid;
}
private bool IsReplyMessage(string[] message)
{
    return message.Length>0 && message[0].StartsWith("<REPLY ");
}
private BasicAntwort ParseResponse(string[] message)
{
    string header = message[0].Substring(7, message[0].Length - 8);
    return new BasicAntwort(message, header);
}

Using these nice little descriptive methods I can use Rx to create an observable sequence of your responses.

var incomingMessages = Observable.FromEventPattern<MessageReceivedEventArgs>(
    h => _networkClient.MessageReceivedEvent += h,
    h => _networkClient.MessageReceivedEvent -= h)
.Select(x => x.EventArgs.Content)
.Where(HasBeginAndEnd)
.Where(IsReplyMessage)
.Select(ParseResponse);

Cool, now we have an incoming stream/sequence.

Next we want to be able to Issue commands and have the appropriate response returned for it. Rx can do this too.

incomingMessages.Where(reply=>reply.Header == befehl)

To continue we also want to add a timeout, so that if we dont get a response from out command for a given time (8000ms?) we should throw. We can convert this to a task, if we only want the single value, then we can do this with Rx too.

incomingMessages.Where(reply=>reply.Header == befehl)
                .Timeout(TimeSpan.FromSeconds(2))
                .Take(1)
                .ToTask();

Nearly done now. We just want a way to send the command and return the task with the response (or the timeout). No problem. Just subscribe first to our incoming message sequence, to avoid race conditions, then issue the command.

public Task<BasicAntwort> SendCommand(NetworkClient networkClient, string befehl)
{
    //Subscribe first to avoid race condition.
    var result = incomingMessages
                        .Where(reply=>reply.Header == befehl)
                        .Timeout(TimeSpan.FromSeconds(2))
                        .Take(1)
                        .ToTask();

    //Send command
    networkClient.SendMessage(befehl);

    return result;
}

Here is the entire code as a LinqPad script

    void Main()
    {
        var _networkClient = new NetworkClient();
        var sendCommandTask = SendCommand(_networkClient, "MyCommand");
        BasicAntwort reply = sendCommandTask.Result;
        reply.Dump();
    }


    private static bool HasBeginAndEnd(string[] message)
    {
        bool isValid = true;

        if (!message[0].StartsWith("<") || !message[0].EndsWith(">"))
            isValid = false;

        if (!message.Last().StartsWith("<END"))
            isValid = false;

        return isValid;
    }
    private static bool IsReplyMessage(string[] message)
    {
        return message.Length>0 && message[0].StartsWith("<REPLY ");
    }
    private static BasicAntwort ParseResponse(string[] message)
    {
        string header = message[0].Substring(7, message[0].Length - 8);
        return new BasicAntwort(message, header);
    }

    public IObservable<BasicAntwort> Responses(NetworkClient networkClient)
    {
        return Observable.FromEventPattern<MessageReceivedEventArgs>(
                h => networkClient.MessageReceivedEvent += h,
                h => networkClient.MessageReceivedEvent -= h)
            .Select(x => x.EventArgs.Content)
            .Where(HasBeginAndEnd)
            .Where(IsReplyMessage)
            .Select(ParseResponse);
    }   

    public Task<BasicAntwort> SendCommand(NetworkClient networkClient, string befehl)
    {
        //Subscribe first to avoid race condition.
        var result = Responses(networkClient)
                            .Where(reply=>reply.Header == befehl)
                            .Timeout(TimeSpan.FromSeconds(2))
                            .Take(1)
                            .ToTask();

        //Send command
        networkClient.SendMessage(befehl);

        return result;
    }   


public class NetworkClient
{
public event EventHandler<MessageReceivedEventArgs> MessageReceivedEvent;
public bool Connected { get; set; }
public void SendMessage(string befehl)
{
    var handle = MessageReceivedEvent;
    if(handle!=null){

        var message = new string[3]{"<REPLY " + befehl +">", "Some content", "<END>"};

        handle(this, new UserQuery.MessageReceivedEventArgs(){Content=message});
    }
}
}
public class MessageReceivedEventArgs : EventArgs
{
public string[] Content { get; set; }
}

public class BasicAntwort
{
    public BasicAntwort(string[] message, string header)
    {
        Header = header;
        Message = message;
    }

    public string Header { get; set; }
    public string[] Message { get; set; }
}