Reactive: Trying to understand how Subject<T> work
c#, system.reactive
Solution
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.
Problem
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` ?