Why Does E.Meijer not like Subjects?

Question

In Erik's PDC talk, I came away with 2 big takes.


I Learnt about Subject<T> and its derivates
Erik says that using them would be wrong and to use Observable.Create instead

This must be a pattern related issue, but I don't get it.  Subject<T> seems to handle all subscribtions and notifications in a thread safe manner so looks like a Silver bullet.

just wondering

Marshall

Answer 1

They are the "mutable variables" of the Rx world and in most cases you do not need them.Typically a solution with Create or the other operators allows you to just wire up continuations without adding extra state. Stated slightly differently, it is good practice to minimize the number of objects that hold on to subscribers, you just want to pass them through.

As an exercise, try writing Where with Create and with a Subject<T> and think about the difference in behavior.

Answer 2

They are the "mutable variables" of the Rx world and in most cases you do not need them.Typically a solution with Create or the other operators allows you to just wire up continuations without adding extra state.

So, in this particular case where the Subject is the initial source of the notifications would its use be appropriate (even if it can be done using Create instead)?

Answer 3

Not sure which "particular case" you refer to;  but if you can do it with a Create, you should not use a Subject IMHO.

Answer 4

Not sure which "particular case" you refer to;  but if you can do it with a Create, you should not use a Subject IMHO.

Sorry, I got my threads mixed up. I was referring to the other thread discussing a class which may be exposing events via an IObservable. In that case since the class is the "source" of the events, then perhaps a Subject would be appropriate. That is, unless the events thrown by that class are in turn generated by an internal IObservable, then you wouldn't want to use a Subject. So I guess that case isn't clear either.

Answer 5

Yes, in that case a subject would be OK; that would correspond to a multicast delegate in a normal .NET event.

Answer 6

They are the "mutable variables" of the Rx world and in most cases you do not need them.Typically a solution with Create or the other operators allows you to just wire up continuations without adding extra state. Stated slightly differently, it is good practice to minimize the number of objects that hold on to subscribers, you just want to pass them through.

As an exercise, try writing Where with Create and with a Subject<T> and think about the difference in behavior.

Are these implementations what your saying Erik?  The first one using create works as expected.  The second one using Subject is too eager. It consumes the stream before the next operator can Hookup. Is this a different issue then the state issue?  How would you fix this implementation using subject and make it lazy as an exercise?  tia

public static IObservable<T> Where1<T>(this IObservable<T> source, Func<T, bool> predicate)
{
 var ob2 = Observable.Create<T>(ob =>
 {
  source.Subscribe(t =>
  {
   if (predicate(t))
    ob.OnNext(t);
  },
  ex => ob.OnError(ex),
  () => ob.OnCompleted());
  return () => { };
 });
 return ob2;
}

public static IObservable<T> Where2<T>(this IObservable<T> source, Func<T, bool> predicate)
{
 Subject<T> sub = new Subject<T>();
 source.Subscribe(t =>
 {
  if (predicate(t))
   sub.OnNext(t);
 },
 ex => sub.OnError(ex),
 () => sub.OnCompleted());
 return sub.AsObservable();
}

---

William Stacey [MVP]

Answer 7

Hi William,

Another problem with the implementation that uses Subject is that the ability to cancel the source subscription is lost.  However, your implementation using Create is actually incorrect because it also doesn't allow the subscription to be cancelled.  You should use CreateWithDisposable instead and return the subscription as the disposable.

- Dave

---

http://davesexton.com/blog

Answer 8

Hi William,

> How would you fix this implementation using subject and make it lazy as an exercise?

Without using Create, you could use Defer instead.  But in that case, you might as well just use Create or CreateWithDisposable and get cancellation for free.

public static IObservable<T> Where3<T>(this IObservable<T> source, Func<T, bool> predicate)
{
	return Observable.Defer(() =>
	{
		var subject = new Subject<T>();

		source.Subscribe(t =>
		{
			if (predicate(t))
				subject.OnNext(t);
		},
		subject.OnError,
		subject.OnCompleted);

		return subject.AsObservable();
	});
}

- Dave

---

http://davesexton.com/blog