Reactive: Trying to understand how Subject work

Question

Trying to understand how the Subject<T>, ReplaySubject<T> and other work. Here is example:

(Subject is Observable and observer)

public IObservable<int> CreateObservable()
{
     Subject<int> subj = new Subject<int>();                // case 1
     ReplaySubject<int> subj = new ReplaySubject<int>();    // case 2

     Random rnd = new Random();
     int maxValue = rnd.Next(20);
     Trace.TraceInformation("Max value is: " + maxValue.ToString());

     subj.OnNext(-1);           // specific value

     for(int iCounter = 0; iCounter < maxValue; iCounter++)
     {
          Trace.TraceInformation("Value: " + iCounter.ToString() + " is about to publish");
          subj.OnNext(iCounter);
      }

      Trace.TraceInformation("Publish complete");
      subj.OnComplete();

      return subj;
 }

 public void Main()
 {
     //
     // First subscription
     CreateObservable()
         .Subscribe(
               onNext: (x)=>{  
                  Trace.TraceInformation("X is: " + x.ToString()); 
      });

     //
     // Second subscribe
     CreateObservable()
         .Subscribe(
               onNext: (x2)=>{  
                  Trace.TraceInformation("X2 is: " + x.ToString());
     });

Case 1: The strange situation is - when I use Subject<T> no subscription is made (???) - I never see the "X is: " text - I only see the "Value is: " and "Max value is"... Why does Subject<T> does not push values to subscription ?

Case 2: If I use ReplaySubject<T> - I do see the values in Subscription but I could not apply Defer option to anything. Not to Subject and not to Observable.... So every subscription will receive different values because CreateObservable function is cold observable. Where is Defer ?

Answer 1

Whenever you need to create an observable out of thin air, Observable.Create should be the first thing to think of. Subjects enter the picture in two cases:

• You need some kind of "addressable endpoint" to feed data to in order for all subscribers to receive it. Compare this to a .NET event which has both an invocation side (through delegate invocation) and a subscription side (through delegate combine with +- and -= syntax). You'll find in a lot of cases, you can achieve the same effect using Observable.Create.

• You need multicasting of messages in a query pipeline, effectively sharing an observable sequence by many forks in your query logic, without triggering multiple subscriptions. (Think of subscribing to your favorite magazine once for your dorm and putting a photo copier right behind the letter box. You still pay one subscription, though all of your friends can read the magazine delivered through OnNext on the letter box.)

Also, in a lot of cases, there's already a built-in primitive in Rx that does exactly what you need. For example, there's From* factory methods to bridge with existing concepts (such as events, tasks, asynchronous methods, enumerable sequence), some of which using a subject under the covers. For the second case of multicasting logic, there's the Publish, Replay, etc. family of operators.

Answer 2

You need to be mindful of when code is executed.

In "Case 1", when you use a Subject<T>, you'll notice that the all of the calls to OnNext & OnCompleted finish before the observable is returned by the CreateObservable method. Since you are using a Subject<T> this means that any subsequent subscription will have missed all of the values so you should expect to get what you got - nothing.

You have to delay the operations on the subject until you have the observer subscribed. To do that using the Create method. Here's how:

public IObservable<int> CreateObservable()
{
    return Observable.Create<int>(o =>
    {
        var subj = new Subject<int>();
        var disposable = subj.Subscribe(o);

        var rnd = new Random();
        var maxValue = rnd.Next(20);
        subj.OnNext(-1);
        for(int iCounter = 0; iCounter < maxValue; iCounter++)
        {
            subj.OnNext(iCounter);
        }
        subj.OnCompleted();

        return disposable;
    });
}

I've removed all the trace code for succinctness.

So now, for every subscriber, you get a new execution of the code inside the Create method and you would now get the values from the internal Subject<T>.

The use of the Create method is generally the correct way to create observables that you return from methods.

Alternatively you could use a ReplaySubject<T> and avoid the use of the Create method. However this is unattractive for a number of reasons. It forces the computation of the entire sequence at creation time. This give you a cold observable that you could have produced more efficiently without using a replay subject.

Now, as an aside, you should try to avoid using subjects at all. The general rule is that if you're using a subject then you're doing something wrong. The CreateObservable method would be better written as this:

public IObservable<int> CreateObservable()
{
    return Observable.Create<int>(o =>
    {
        var rnd = new Random();
        var maxValue = rnd.Next(20);
        return Observable.Range(-1, maxValue + 1).Subscribe(o);
    });
}

No need for a subject at all.

Let me know if this helps.