9

There are IObservable and IObserver interfaces in .NET (also here and here). Interestingly, the concrete implementation of the IObserver does not hold a direct reference to the IObservable. It doesn't know who it's subscribed to. It can only invoke the unsubscriber. "Please pull the pin to unsubscribe."

edit: The unsubscriber implements the IDisposable. I think, this scheme was employed to prevent the lapsed listener problem.

Two things are not entirely clear to me, though.

  1. Does the inner Unsubscriber class provide the subscribe-and-forget behavior? Who (and when exactly) calls IDisposable.Dispose() on the Unsubscriber? Garbage collector (GC) is not deterministic.
    [Disclaimer: overall, I've spent more time with C and C++ than with C# .]
  2. What should happen if I want to subscribe an observer K to an observable L1 and the observer is already subscribed to some other observable L2?

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

    When I ran this test code against MSDN's example, the observer remained subscribed to L1. This would be peculiar in real development. Potentially, there are 3 avenues to improve this:

    • If the observer already has an unsubscriber instance (i.e. it’s already subscribed), then it quietly unsubscribes from the original provider before subscribing to a new one. This approach hides the fact that it’s no longer subscribed to the original provider, which may become a surprise later.
    • If the observer already has an unsubscriber instance, then is throws an exception. A well-behaved calling code has to unsubscribe the observer explicitly.
    • Observer subscribes to multiple providers. This is the most intriguing option, but can this be implemented with IObservable and IObserver? Let’s see. It is possible for the observer to keep a list of unsubscriber objects: one for each source. Unfortunately, IObserver.OnComplete() does not provide a reference back to the provider who have sent it. So, the IObserver implementation with multiple providers would not be able to determine which one to unsubscribe from.
  3. Was .NET's IObserver intended for subscribing to multiple IObservables?
    Does the textbook definition of the observer pattern require that one observer has to be able to subscribe to multiple providers? Or is it optional and implementation-dependent?

5

The two interfaces are actually part of Reactive Extensions (Rx for short), you should use that library pretty much whenever you want to use them.

The interfaces are technically in mscrolib, not in any of the Rx assemblies. I think this is to ease interoperability: this way, libraries like TPL Dataflow can provide members that work with those interfaces, without actually referencing Rx.

If you use Rx's Subject as your implementation of IObservable, Subscribe will return an IDisposable that can be used for unsubscribing:

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);
5

Just to clear up a few things that are well documented in the official Rx Design Guidelines and at length on my web site IntroToRx.com:

  • You dont rely on the GC to clean up your subscriptions. Covered in detail here
  • There is no Unsubscribe method. You subscribe to an observable sequence and are given a subscription. You can then dispose of that subscription indicating that you no longer want to have your callbacks invoked.
  • An observable sequence can not be completed more than once (see section 4 of the Rx Design Guidelines).
  • There are numerous ways to consume multiple observable sequences. There is also a wealth of information regarding that on Reactivex.io and again at IntroToRx.

To be specific and answer the original question directly, your usage is back to front. You don't push many observable sequences into a single observer. You compose observable sequences into a single observable sequence. You then subscribe to that single sequence.

Instead of

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

Which is just pseudo code and would not work in .NET implementation of Rx, you should do the following:

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();

Now this doesn't exactly fit the initial question, but I don't know what K.Unsubscribe() was supposed to do (unsubscribe from all, the last or the first subscription?!)

  • Can I simply encase the subscription object in a "using" block? – Robert Oschler May 19 '15 at 18:09
  • 1
    In this synchronous case you can, however Rx is supposed to be asynchronous. In the asynchronous case, you can not normally use the using block. The cost for a subscription statement should be virtually zero, so you would neter the using block, subscribe, leave the using block (thus unsubscribe) making the code rather pointless – Lee Campbell May 19 '15 at 18:35
3

You're right. The example works poorly for multiple IObservables.

I guess OnComplete() doesn't provide a reference back because they don't want the IObservable to have to keep it around. If I were writing that I would probably support multiple subscriptions by having Subscribe take an identifier as a second parameter, which gets passed back to the OnComplete() call. So you could say

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

As it stands, it appears the .NET IObserver isn't suitable for multiple observers. But I suppose your main object (LocationReporter in the example) could have

public Dictionary<String,IObserver> Observers;

and that would enable you to support

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

as well.

I suppose Microsoft could argue that therefore there is no need for them to directly support multiple IObservables in the interfaces.

  • I was also thinking that the observable implementation can have a list of observers. I too have noticed that IObserver.OnComplete() doesn’t identify who the call comes from. If the observer is subscribed to more than one observable, then it doesn’t know who to unsubscribe from. Anticlimactic. I wonder, does .NET have a better interface for the observer pattern? – Nick Alexeev Jul 4 '14 at 23:35
  • If you want to have a reference to something, you should actually use a reference, not a string. – svick Jul 6 '14 at 10:00
  • This answer helped me with a real-life bug. I was using Observable.Create() to build an observable, and chaining several source observables into it using Subscribe(). I inadvertently passed a completed observable in one code path. This completed my newly created observable, even though the other sources were not complete. Took me ages to work out what I needed to do - switch Observable.Empty() for Observable.Never(). – Olly Sep 23 '17 at 22:42
0

I know this is way late to the party, but...

The interfaces IObservable<T> and IObserver<T> are not part of Rx...they're core types...but Rx makes extensive use of them.

You are free to have as many (or as few) observers as you like. If you anticipate multiple observers, it's the observable's responsibility to route OnNext() calls to the appropriate observers for each observed event. The observable might need a list or a dictionary as you suggest.

There are good cases for allowing only one - and good cases for allowing many. For example, in an CQRS/ES implementation, you might enforce a single command handler per command type on a command bus, while you might notify several read-side transforms for a given event type in the event store.

As stated in other answers, there's no Unsubscribe. Disposing of the thing you're given when you Subscribe generally does the dirty work. The observer, or an agent thereof, is responsible for holding onto the token until it no longer wants to receive further notifications. (qestion 1)

So, in your example:

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

...it would be more like:

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 );
}

...where K would hear 1003 and 1004 but not 1005.

To me, this still looks funny because nominally, subscriptions are long-lived things...often for the duration of the program. They're not dissimilar in this respect to normal .Net events.

In a lot of examples I've seen, the Dispose of the token does work to remove the observer from the observable's list of observers. I prefer that the token not carry so much knowledge around...and so I have generalized my subscription tokens to just call a passed-in lambda (with identifying information captured at subscribe-time:

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

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

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

...and the the observable can install the unsubscribe behavior during subscription:

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

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

If your observer is catching events from multiple observables, you might want to ensure that there's some kind of correlation information in the events themselves...like .Net events do with the sender. It's up to you whether that matters or not. It's not baked in, as you've correctly reasoned. (question 3)

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.