War der IObserver <T> von .NET für das Abonnieren mehrerer IObservables vorgesehen?

In .NET gibt es IObservable- und IObserver- Schnittstellen (auch hier und hier ). Interessanterweise enthält die konkrete Implementierung des IObservers keinen direkten Verweis auf das IObservable. Es weiß nicht, wen es abonniert hat. Es kann nur den Abmelder aufrufen. "Bitte ziehen Sie den Stift, um sich abzumelden."

edit: Der Abmelder implementiert das IDisposable. Ich denke, dieses Schema wurde angewendet, um das Problem des erloschenen Hörers zu verhindern .

Zwei Dinge sind mir jedoch nicht ganz klar.

1. Bietet die innere Unsubscriber-Klasse das Subscribe-and-Forget-Verhalten? Wer (und wann genau) ruft IDisposable.Dispose()den Abmelder an? Garbage Collector (GC) ist nicht deterministisch.
   [Haftungsausschluss: Insgesamt habe ich mehr Zeit mit C und C ++ verbracht als mit C #.]

2. Was soll passieren, wenn ich einen Beobachter K einem beobachtbaren L1 abonnieren möchte und der Beobachter bereits einen anderen beobachtbaren L2 abonniert hat?

   K.Subscribe(L1);
   K.Subscribe(L2);
   K.Unsubscribe();
   L1.PublishObservation(1003);
   L2.PublishObservation(1004);
   

   Als ich diesen Testcode anhand des Beispiels von MSDN ausführte, blieb der Beobachter bei L1 abonniert. Dies wäre in der realen Entwicklung eigenartig. Möglicherweise gibt es drei Möglichkeiten, dies zu verbessern:

   • Wenn der Beobachter bereits eine Instanz für das Abbestellen hat (dh sie ist bereits abonniert), wird er sich stillschweigend vom ursprünglichen Anbieter abmelden, bevor er eine neue abonniert. Dieser Ansatz verbirgt die Tatsache, dass er nicht mehr beim ursprünglichen Anbieter abonniert ist, was später zu einer Überraschung werden kann.
   • Wenn der Beobachter bereits eine Abmeldeinstanz hat, wird eine Ausnahme ausgelöst. Ein gut erzogener Aufrufcode muss den Beobachter explizit abbestellen.
   • Observer abonniert mehrere Anbieter. Dies ist die faszinierendste Option. Kann dies jedoch mit IObservable und IObserver implementiert werden? Mal schauen. Der Beobachter kann eine Liste der nicht abonnierten Objekte führen: eines für jede Quelle. Liefert leider IObserver.OnComplete()keinen Verweis auf den Anbieter, der ihn gesendet hat. Die IObserver-Implementierung mit mehreren Anbietern kann daher nicht bestimmen, von welchem ​​Anbieter abgemeldet werden soll.

3. War der IObserver von .NET für das Abonnieren mehrerer IObservables vorgesehen?
   Erfordert die Lehrbuchdefinition des Beobachtermusters, dass ein Beobachter mehrere Anbieter abonnieren kann? Oder ist es optional und implementierungsabhängig?

Die beiden Schnittstellen sind tatsächlich Teil von Reactive Extensions (kurz Rx). Sie sollten diese Bibliothek so gut wie immer verwenden, wenn Sie sie verwenden möchten.

Die Schnittstellen befinden sich technisch in mscrolib, nicht in einer der Rx-Assemblys. Ich denke, dies soll die Interoperabilität vereinfachen: Auf diese Weise können Bibliotheken wie TPL Dataflow Mitglieder bereitstellen , die mit diesen Schnittstellen arbeiten , ohne tatsächlich auf Rx zu verweisen.

Wenn Sie Rx Subjectals Implementierung von verwenden IObservable, Subscribewird ein zurückgegeben IDisposable, der zum Abbestellen verwendet werden kann:

var observable = new Subject<int>();

var unsubscriber =
    observable.Subscribe(Observer.Create<int>(i => Console.WriteLine("1: {0}", i)));
observable.Subscribe(Observer.Create<int>(i => Console.WriteLine("2: {0}", i)));

unsubscriber.Dispose();

observable.OnNext(1003);
observable.OnNext(1004);

Nur um ein paar Dinge zu klären, die in den offiziellen Rx Design Guidelines und ausführlich auf meiner Website IntroToRx.com gut dokumentiert sind :

• Sie verlassen sich nicht auf den GC, um Ihre Abonnements zu bereinigen. Hier ausführlich behandelt
• Es gibt keine UnsubscribeMethode. Sie abonnieren eine beobachtbare Sequenz und erhalten ein Abonnement . Sie können dieses Abonnement dann entsorgen, um anzuzeigen, dass Ihre Rückrufe nicht mehr aufgerufen werden sollen.
• Eine beobachtbare Sequenz kann nicht mehr als einmal abgeschlossen werden (siehe Abschnitt 4 der Rx-Entwurfsrichtlinien).
• Es gibt zahlreiche Möglichkeiten, mehrere beobachtbare Sequenzen zu konsumieren. Es gibt auch eine Fülle von Informationen dazu auf Reactivex.io und erneut bei IntroToRx.

Um genau zu sein und die ursprüngliche Frage direkt zu beantworten, ist Ihre Verwendung von hinten nach vorne. Sie schieben nicht viele beobachtbare Sequenzen in einen einzelnen Beobachter. Sie setzen beobachtbare Sequenzen zu einer einzigen beobachtbaren Sequenz zusammen. Sie abonnieren dann diese einzelne Sequenz.

Anstatt von

K.Subscribe(L1);
K.Subscribe(L2);
K.Unsubscribe();
L1.PublishObservation(1003);
L2.PublishObservation(1004);

Was nur Pseudocode ist und in der .NET-Implementierung von Rx nicht funktionieren würde, sollten Sie Folgendes tun:

var source1 = new Subject<int>(); //was L1
var source2 = new Subject<int>(); //was L2

var subscription = source1
    .Merge(source2)
    .Subscribe(value=>Console.WriteLine("OnNext({0})", value));


source1.OnNext(1003);
source2.OnNext(1004);

subscription.Dispose();

Das passt nicht genau zur ursprünglichen Frage, aber ich weiß nicht, was ich tun K.Unsubscribe()sollte (von allen abbestellen, vom letzten oder vom ersten Abonnement?!)

Du hast recht. Das Beispiel funktioniert schlecht für mehrere IObservables.

Ich denke, OnComplete () liefert keine Referenz zurück, weil sie nicht wollen, dass das IObservable es behalten muss. Wenn ich schreiben würde, dass ich wahrscheinlich mehrere Abonnements unterstützen würde, indem Subscribe einen Bezeichner als zweiten Parameter verwendet, der an den OnComplete () -Aufruf zurückgegeben wird. Man könnte also sagen

K.Subscribe(L1,"L1")
K.Subscribe(L2,"L2")
K.Unsubscribe("L1")

Derzeit scheint der .NET IObserver nicht für mehrere Beobachter geeignet zu sein. Aber ich nehme an, Ihr Hauptobjekt (LocationReporter im Beispiel) könnte haben

public Dictionary<String,IObserver> Observers;

und das würde es Ihnen ermöglichen, zu unterstützen

K.Subscribe(L1,"L1")
K.Subscribe(L2,"L2")
K.Unsubscribe("L1")

auch.

Ich nehme an, Microsoft könnte argumentieren, dass sie daher nicht mehrere IObservables direkt in den Schnittstellen unterstützen müssen.

Ich weiß, das ist viel zu spät zur Party, aber ...

Die Schnittstellen I Observable<T>und IObserver<T>sind nicht Teil von Rx ... sie sind Kerntypen ... aber Rx nutzt sie ausgiebig.

Es steht Ihnen frei, so viele (oder so wenige) Beobachter zu haben, wie Sie möchten. Wenn Sie mehrere Beobachter erwarten, liegt es in der Verantwortung des Beobachtbaren, OnNext()Anrufe für jedes beobachtete Ereignis an die entsprechenden Beobachter weiterzuleiten. Das Observable benötigt möglicherweise eine Liste oder ein Wörterbuch, wie Sie vorschlagen.

Es gibt gute Fälle, in denen nur einer zugelassen wird - und gute Fälle, in denen viele zugelassen werden. In einer CQRS / ES-Implementierung können Sie beispielsweise einen einzelnen Befehlshandler pro Befehlstyp auf einem Befehlsbus erzwingen , während Sie möglicherweise mehrere leseseitige Transformationen für einen bestimmten Ereignistyp im Ereignisspeicher benachrichtigen .

Wie in anderen Antworten angegeben, gibt es keine Unsubscribe. Entsorgen Sie das, was Sie erhalten, wenn Sie im SubscribeAllgemeinen die Drecksarbeit erledigen. Der Beobachter oder ein Vertreter davon ist dafür verantwortlich , an dem Token festzuhalten, bis er keine weiteren Benachrichtigungen mehr erhalten möchte . (Frage 1)

Also, in Ihrem Beispiel:

K.Subscribe(L1);
K.Subscribe(L2);
K.Unsubscribe();
L1.PublishObservation(1003);
L2.PublishObservation(1004);

... es wäre eher wie:

using ( var l1Token = K.Subscribe( L1 ) )
{
  using ( var l2Token = K.Subscribe( L2 );
  {
    L1.PublishObservation( 1003 );
    L2.PublishObservation( 1004 );
  } //--> effectively unsubscribing to L2 here

  L2.PublishObservation( 1005 );
}

... wo K 1003 und 1004 hören würde, aber nicht 1005.

Für mich sieht das immer noch lustig aus, denn Abonnements sind nominell langlebige Dinge ... oft für die Dauer des Programms. Sie unterscheiden sich in dieser Hinsicht nicht von normalen .Net-Ereignissen.

In vielen Beispielen, die ich gesehen habe, entfernt der DisposeToken den Beobachter aus der Beobachterliste des Observablen. Ich bevorzuge, dass das Token nicht so viel Wissen enthält ... und deshalb habe ich meine Abonnement-Token so verallgemeinert, dass sie nur ein übergebenes Lambda aufrufen (mit identifizierenden Informationen, die zum Zeitpunkt des Abonnements erfasst wurden:

public class SubscriptionToken<T>: IDisposable
{
  private readonly Action unsubscribe;

  private SubscriptionToken( ) { }
  public SubscriptionToken( Action unsubscribe )
  {
    this.unsubscribe = unsubscribe;
  }

  public void Dispose( )
  {
    unsubscribe( );
  }
}

... und das Observable kann das Abmeldeverhalten während des Abonnements installieren:

IDisposable Subscribe<T>( IObserver<T> observer )
{
  var subscriberId = Guid.NewGuid( );
  subscribers.Add( subscriberId, observer );

  return new SubscriptionToken<T>
  (
    ( ) =>
    subscribers.Remove( subscriberId );
  );
}

Wenn Ihr Beobachter Ereignisse von mehreren Observablen abfängt, möchten Sie möglicherweise sicherstellen, dass die Ereignisse selbst Korrelationsinformationen enthalten ... wie dies bei .Net-Ereignissen der Fall ist sender. Es liegt an Ihnen, ob das wichtig ist oder nicht. Es ist nicht eingebrannt, wie Sie richtig argumentiert haben. (Frage 3)