f # beobachtbare Gabel und Nebeneffekt

Question

Ich habe eine beobachtbare Kette, die anfängliche Observable stammt aus dem Netzwerk und wird jedes Mal ausgelöst, wenn die Nachricht zum Lesen bereit ist. Der nächste Handler liest dann die Nachricht und deserialisiert sie. Jetzt habe ich eine Gabelung des Observablen, der eine ist der Nachrichtenhandler und der andere protokolliert die Nachricht.

Das Problem ist, dass, weil ich Observable benutze ich tatsächlich versuchen werde, die Nachricht zweimal zu lesen.

Ich verstehe, dass die Verwendung von Event anstelle von Observable das Problem lösen wird, jedoch habe ich dann ein Garbage Collection-Problem, das dazu führen kann, dass Sockets nicht erfasst werden.

Eine Lösung, an die ich gedacht habe, ist das Einfügen einer Art Trennzeichen, die eine Kette von Observablen beendet und eine neue erzeugt, existiert eine solche Funktion bereits als Teil von fsharp oder einer anderen Bibliotheksbibliothek.

Gibt es andere Lösungen für das Problem?

Edit:

Codebeispiel, das nicht richtig

let messagesStream = 
    socket.observable |> 
    Observable.map (fun() -> socket.read()) |> 
    Observable.map (fun m -> deserialize m) 

messagesStream |> Observable.add (fun m -> printf m) 
messagesStream |> Observable.add (fun m -> handle m) 

Answer 1

Es klingt funktioniert, wie Sie eine beobachtbare, die das Lesen der Nachricht übernimmt aus dem Netzwerk schaffen könnte und deserialisiert es. Unter der Annahme, dass dies mit Standard-Rx-Operatoren geschieht, sollte dies eine Observable zurückgeben, die neue, deserialisierte Netzwerknachrichten überträgt.

Sie können 2 Abonnenten für diese Observable haben, eine, die auf neue Nachrichten mit der gewünschten Geschäftslogik reagiert und der zweite Abonnent die Nachrichten protokolliert.

Das sollte den Nebeneffekt des mehrfachen Lesens aus dem Netzwerk beseitigen. Das Drücken von 2 Kopien der deserialisierten Nachricht sollte keinen Nebeneffekt haben.

Answer 2

Der einfachste Weg, um einige Protokollierung hinzuzufügen ist Observable.iter wie folgt zu verwenden:

let messagesStream = 
    socket.observable 
    |> Observable.map (fun() -> socket.read()) 
    |> Observable.map (fun m -> deserialize m) 
    |> Observable.iter (printfn "%A") 

messagesStream |> Observable.add (fun m -> handle m)