Wie schiebe ich eine Entität auf eine Rx Observable?

Question

Ich habe eine Klasse, die für die Generierung von Ereignissen in einem häufigen, aber unregelmäßigen Intervall verantwortlich ist, das andere Klassen verbrauchen und bearbeiten müssen. Ich möchte Reactive Extensions für diese Aufgabe verwenden.

Die Verbraucherseite ist sehr einfach; Ich habe meine Verbraucherklasse IObserver<Payload> und alles scheint gut. Das Problem kommt auf die Erzeugerklasse.

Implementierung IObservable<Payload> direkt (das heißt, für IDisposable Subscribe(IObserver<Payload>) meine eigene Implementierung setzen wird, entsprechend die Dokumentation, nicht zu empfehlen. Es schlägt vor, statt mit dem Observable.Create() Satz von Funktionen zu komponieren. Da meine Klasse für eine lange Zeit laufen, ich habe versucht, mit var myObservable = Observable.Never() einen beobachtbaren erstellen, und dann, wenn ich neue Nutzlasten zur Verfügung habe, myObservable.Publish(payloadData) anrufen. wenn ich das tue, aber ich scheine nicht die OnNext Umsetzung in meinem Verbraucher zu treffen.

ich denke, als ein Workaround, ich kann ein Ereignis in meiner Klasse erstellen und dann erstellen Sie die Observable mit der FromEvent Funktion, aber das scheint wie ein Overly Co mplizierten Ansatz (d. h. es erscheint seltsam, dass die neue Schärfe von Observablen dazu führt, dass Ereignisse funktionieren). Gibt es einen einfachen Ansatz, den ich hier übersehen habe? Wie können Sie Ihre eigenen Observable-Quellen erstellen?

Answer 1

ein new Subject<Payload> erstellen und nennt es OnNext Methode ist ein Ereignis zu senden. Sie können Subscribe zum Thema mit Ihrem Beobachter.

Die Verwendung von Themen wird oft diskutiert. Für eine gründliche Diskussion über die Verwendung von Subjekten (die mit einem Primer verknüpft sind), siehe here - aber zusammenfassend klingt dieser Anwendungsfall gültig (d.h. er erfüllt wahrscheinlich die lokalen + heißen Kriterien).

Nebenbei, Überladungen von Subscribe akzeptieren Delegierten möglicherweise die Notwendigkeit für Sie, eine Implementierung von IObserver<T> bereitzustellen.

Answer 2

Observable.Never() keine Benachrichtigungen auf „Senden“, Sie Observable.Return(yourValue)

verwenden sollten, wenn Sie einen Leitfaden mit konkreten Beispielen muss ich lesen empfehlen Intro to Rx

Answer 3

Es sei denn, ich auf einen besseren Weg kommen, es zu tun, was ich haben sich für jetzt festgelegt ist die Verwendung eines BlockingCollection.

var _itemsToSend = new BlockingCollection<Payload>(); 
IObservable<MessageQueuePayload> _deliverer = 
_itemsToSend.GetConsumingEnumerable().ToObservable(Scheduler.Default);