Anfrage/Antwort mit Streams

Question

Ich habe einen Input-Stream von einem RS-232-Port und eine Warteschlange von Befehlen an den seriellen Port mit Rx-Streams. Ich habe meinen Code vereinfacht wie folgt:

void Init() 
{ 
    SerialPort srl; 

    ... // open serial port 

    IObservable<string> obInput = 
     Observable.FromEventPattern< 
     SerialDataReceivedEventHandler, 
     SerialDataReceivedEventArgs> 
     (
      handler => srl.DataReceived += handler, 
      handler => srl.DataReceived -= handler 
     ).SelectMany(_ => 
     { 
     List<string> ret; 
     ... //extract messages 
     return ret; 
     }).Publish().Refcount(); 

    obCommandOk = 
     obInput 
     .Where(msg => msg == "OK" || msg == "KO"); 

    var sAction = new Subject<string>(); 
    var sCommandOk = new Subject<Tuple<string,bool>>(); 

    sAction 
     .Do(srl.WriteLine) 
     .Zip(obCommandOk, (cmd, result) => 
     { 
     if (result == "OK") 
      sCommandOk.OnNext(Tuple.Create(cmd, true)) 
     else 
      sCommandOk.OnNext(Tuple.Create(cmd, false)) 
     }); 
} 

async bool Command(string cmd) 
{ 
    sAction.OnNext(cmd); 

    return 
     await sCommandOk 
     .Where(t => t.Item1 == cmd) 
     .Select(t => t.Item2) 
     .FirstAsync(); 
} 

kommt vor, dass nach OnNext das Ergebnis bereits nach sCommandOk gedrückt wurde und so habe ich es zu verlieren.

Können Sie mir einen besseren Ansatz vorschlagen, um den Verlust von Antworten zu vermeiden?

Answer 1

Sie haben eine Wettlaufbedingung in Ihrer Command Methode. Sie drücken die Aktion, dann abonnieren Sie das Ergebnis. Wenn das Ergebnis schnell ist, dann verlieren Sie es.

eine kleine Änderung hier kann man erkennen, mildern:

async Task<bool> Command(string cmd) 
{ 
    var result = sCommandOk 
     .Where(t => t.Item1 == cmd) 
     .Select(t => t.Item2) 
     .FirstAsync(); 

    sAction.OnNext(cmd); 

    return await result; 
} 

Ich glaube, Sie weiter durch das Entfernen des sCommandOk Thema optimieren können alle zusammen

async Task<bool> Command(string cmd) 
{ 
    var result = obInput 
     .Where(t => t.Item1 == cmd) 
     .Select(t => t.Item2) 
     .FirstAsync(); 

    sAction.OnNext(cmd); 

    return await result; 
}