Observable.Publish mit Retrow Verhalten

Question

Ich habe eine kalte Observable, die OnError, wenn es abonniert ist. Wie kann ich ein kaltes Observable erstellen, das ein einzelnes Element zurückgibt (ein Objekt, das die Observable Quelle als Abhängigkeit empfängt) oder das OnError der Quelle propagiert. Mit dem Publish Operator wird der onError Handler nicht aufgerufen.

Private Shared Sub Test() 
    Dim source = Observable.Throw(Of Integer)(New Exception) 
    ' Dim source = Observable.Range(0, 9) 

    Dim obs = source.Publish(Function(published) 
           Return Observable. 
             Return(New ObjectThatConsumesXs(published)) 
          End Function) 

    obs.Subscribe(Sub(a) 

        End Sub, 
        Sub(ex) 

        End Sub, 
        Sub() 

        End Sub) 
End Sub 


Private Class ObjectThatConsumesXs 
    Private _subscription As IDisposable 

    Public Sub New(source As IObservable(Of Integer)) 
     _subscription = source.Subscribe(Sub(x) 

             End Sub, 
             Sub(ex) 

             End Sub, 
             Sub() 

             End Sub) 
    End Sub 
End Class 

EDIT:

Das wird irgendwie eine lange Beschreibung sein.

Ich habe ein Gerät, das im Wesentlichen ein CAN Bus Scanner ist. Dieses Gerät verfügt über eine serielle Schnittstelle und beginnt nach Empfang eines Start-Befehls mit der Spiegelung aller Nachrichten, die es auf dem CAN-Bus erfasst, bis es einen Stoppbefehl erhält. Die beantworteten Nachrichten sind in einer Variante von PPP Protocol verpackt, um Fehler zu mindern, da die Baudrate des seriellen Ports etwa 1 MBaud beträgt.

Ich möchte eine Desktop-Anwendung entwerfen, die eine Verbindung zum Scanner herstellt, Befehle an sie sendet und die erfassten CAN-Nachrichten empfängt. Es sollte die empfangenen Nachrichten in einer ListBox/ListView anzeigen, mit der Möglichkeit zu filtern, was durch einige Kriterien angezeigt wird. Es sollte auch nach IDs gruppiert werden, die in jeder Nachricht eingebettet sind, und eine Liste mit aufgetretenen IDs und deren Gesamtvorkommen anzeigen. Es sollte auch eine Gesamtzahl eindeutiger IDs und eine Gesamtanzahl von Nachrichten anzeigen.

Zwischen einem Startbefehl und einem Stoppbefehl wird eine Sammlung von Nachrichten empfangen, die einen Datensatz darstellen. Die Anwendung sollte in der Lage sein, mehrere Male in einer Sitzung aufzuzeichnen, und sie muss eine Möglichkeit bieten, Datensätze auf der Festplatte zusammen mit angewendeten Filtern, einem benutzerdefinierten Namen usw. zu archivieren. Die gleiche Anwendung sollte diese Datensätze für die Offline-Analyse importieren können.

Die oben erwähnte ObjectThatConsumesXs ist meine Aufzeichnung, die ihre enthaltenen Nachrichten als Observable exponiert, die ihre Abonnenten antworten (Reply-Operator).

Ich benutze ReactiveUI - MVVM/WinForms/Reaktive Erweiterungen, und unter anderem habe ich es geschafft, einen Dienst zu entwerfen, der eine GetRecordUntil Funktion verfügbar macht, die eine IObservable(Of Record) zurückgibt. Beim Abonnieren gibt das Observable einen einzelnen Datensatz aus, der mit den vom Scanner empfangenen Nachrichten aktualisiert wird.

Ich bin offen für Vorschläge in Bezug auf das Design der Anwendung. Aber ich habe Angst, dass meine Frage zumindest neu getaggt werden sollte, wenn nicht sogar umbenannt.

Answer 1

Im Allgemeinen würde ich vorschlagen, ein Design, das Sie nicht IObservable(Of T) in Parameter übergeben. Die Observable-Schnittstellen bieten eine Möglichkeit für etwas, auf das Sie angewiesen sind (d. H. Kein Wissen von Ihnen hat), Sie zurückzurufen. Wenn Sie jedoch eine beobachtbare Sequenz zu etwas weiterleiten, dann wissen Sie deutlich davon, und es erwartet eindeutig, dass es Reiz genannt wird (oder darauf reagiert). Warum rufen Sie nicht direkt Methoden zu dieser Abhängigkeit auf, wenn die Ereignisse passieren?

Unabhängig davon, das aktuelle Design, das Sie haben, wird die OnError nicht weitergeben. Die äußere Sequenz garantieren Sie immer nur OnNext eine einzelne ObjectThatConsumesXs und dann vervollständigen Sie die Sequenz. Intern wird ObjectThatConsumesXs die veröffentlichte Sequenz subskribieren, den Fehler erhalten, aber keine Möglichkeit haben, das an den anderen Codepfad zurückzugeben.

Als eine weitere Notiz haben Sie einen Typ mit dem Namen ObjectThatConsumesXs, aber Sie gehen dann zum Verbraucher die äußere Sequenz direkt innerhalb der anderen Methode. Warum das Doppelhandling?

Wenn Sie erklären können, was Sie versuchen zu tun (nicht wie Sie es zu lösen versuchen), dann bin ich sicher, dass die Community Sie auf ein passenderes Design hinweisen kann.