Although Subject<T> is a good way for consumers of Rx to get an IObservable<T> implementation, we tend not to use it internally for efficiency reasons.

If you look at pretty much any operator, you'll see that the IObservable<T> implementation they provide is usually based on one of the IProducer<T> implementations in Producer.cs

The reason for this is that when chaining together a sequence of operators (and most applications that use Rx rarely use just a single operator) we make a distinction between ones that are implemented by Rx, and IObservable<T> implementations where we don't know where they come from.

To ensure reliability, Rx is cautious when dealing with sources of unknown origin. It does not presume they will necessarily comply fully with the laws of Rx. We create wrappers around external observable sources to detect and handle anomalous behaviour, and we also do the same with external observers.

But with something like this:

// the details of this are nonsense - the point is we have a load of operators
// chained together
IObservable<double> r = src
    .Where(x => x.Name[0] == 'I')
    .OrderBy(x => x.Age)
    .Skip(1)
    .Select(x => x.Height)
    .Average();

the thing we don't want to do is have these safeguarding wrappers between every stage. We don't want this:

src -> guard -> Where -> guard -> OrderBy -> Skip -> guard -> Select -> guard -> Average -> guard

We want this:

src -> guard -> Where -> OrderBy -> Skip -> Select -> Average -> guard

We need those safeguards on the input side (assuming src here is something outside of Rx's control) and also on the output side (because who knows what exceptions might emerge from an application-supplied IObserver<T>? we need to ensure Rx's state remains coherent if the app does something odd). But we really don't need them between each and every operator, because we know that for every operator we've implemented, as long as its inputs are well behaved, its outputs will be well behaved. Similarly, as long as it is talking to a well-behaved IObserver<T>, the IObserver<T> that each of our operators presents to its predecessor will also be well behaved.

So the basic strategy is that Rx operators all recognize when they've been connected to another Rx operator, and they can bypass the relatively expensive safeguards that we need when interfacing with external code.

And although Subject<T> is an Rx library feature, it is designed very much to be used at the boundaries between Rx and application code, so we don't treat it as one of our recognized trusted operators.

So, taking that background into account, I think I would want to see this type implemented more like one of the other built-in Rx operators.

I say "I think" because one thing about the way we normally do things is that we mostly avoid making the implementing type of an IObservable<T> visible. Although when you use, say, Any, you're getting a type that derives from Producer<bool, ...>, that implementing type is hidden from you. We just expose an IObservable<T>. (And in fact Any doesn't always return the same type. Like a lot of our operators, we select between a number of different implementation strategies depending on inputs. And we want to retain the flexibility to change the set of strategies available, which means we can't have the concrete type be visible at the API. Also, our IProducer<T> base classes are all internal which prevents anything deriving from them from being public types.)

But the usage model here is quite simply:

new ObservableDisposable()

This is consistent with many of our other Disposables. But it is not consistent with most of our IObservable<T>s. You've created a conundrum here by coming up with a type that is both disposable and observable. We have different common practices for each category, so what is the correct choice for something that is in both categories?

I'm inclined to say this should work more like Disposable.Empty: that gives you an IDisposable whose concrete type is a private class. So with your observable disposable, we might have a factory method like Disposable.Observable(). For that to work, we'd need also to introduce an ICancelableObservable<T> interface that inherits both interfaces, and use ICancelableObservable<Unit> as the return type of that factory method.

One alternative would be just to accept that this ObservableDisposable type has suboptimal performance when used as the source for other Rx operators.

I understand the conflict the class being an obserable AND a disposable.

• If it will be a class or a Disposable.Observable() factory method, it will be be the first creation method of an observable outside of the Observable class (and QueryLanguage, if i see it right).
• If it will be a factory method in the Observable class, it will be the first disposable outside of the Dispoable namespace.

Would it be an option to go both ways to create such object?

I also used Subject here to not have to worry about the list of observers. I had a go at implementing ObservableDisposable as Producer and while this is great, I now have a list of sinks to notify and worry about. But maybe there is a better solution?

Since this Observable would only send one OnNext and one OnComplete notification, I think the suboptiomal performance is acceptable.

I'm not sure if it's your intention for this code to work in the face of concurrent calls to Dispose. The use of Volatile.Read does suggest that this is the goal, although maybe this is just intending to handle correctly strictly sequential calls to Dispose that happen on different threads.

If this is meant to handle concurrent calls, I think there's a bug here. If two threads call Dispose simultaneously, it's possible for each of them to complete this Volatile.Read before the other reaches the Volatile.Write. So both threads will end up calling OnNext and OnCompleted. We will have violated the IObserver<T> contract:

• A call to OnCompleted followed by further calls (at least one more OnCompleted, and possibly a call to OnNext after a call to OnCompleted)
• Potentiall starting a new calls into an IObserver<T> when another call was already in progress and had not returned

There are two ways to deal with this:

1. declare that this is the caller's problem
2. implement something more robust
3. rely on the fact that Rx will treat the Subject<T> as a suspect external source, and it will actually block illegal call sequences

We can rule out 3. Rx helps out applications that make such mistakes, up to a point, but we certainly shouldn't be relying on these safety guards in Rx's own code. In any case, while it would detect and block the IObservable<T> calls that occur after the first OnCompleted I don't think it would help at all with concurrent calls.

Option 1 actually has its charms (and may be what you intended, in which case the XML doc comments should definitely clarify that instances of this class are not safe for concurrent use; someone looking at the source code might see Volatile and assume that it is designed for multithreaded use). It's basically our official position for any IObserver<T> we hand out: concurrent calls are not supported, and you get undefined behaviour if you make them. In that case, this method could be simplified: you'd remove the use of Volatile entirely and make no attempt to handle concurrent use.

However, although that's reasonable for IObservable<T> (because there are well defined rules restricting its use) arguably it isn't for IDisposable. It's not unreasonable to want an IDisposable that tolerates concurrent calls to Dispose.

So that would leave 2. But although this could probably be fixed by using some sort of interlocked exchange instead of simply a volatile read, but in practice, I think if this were all changed to fit the IProducer<T> pattern, and if you used one of the Sink<...> types normally used as part of such an implementation, we'd need to look at how concurrent calls to Dispose interact with the existing disposal handling in there. This already uses Interlocked.Exchange to prevent double disposal, so it's possible you could take advantage of that.

Yes, Interlocked.Exchange seems to be the better solution.

Since this is a sealed class without a finalizer, I don't see how instances of this type can ever end up on the finalization queue, so this won't do anything will it?

Well I shouls discard this call then... 🤦

As far as I can tell, you don't really need this, because when the object is in a disposed state, the _disposedNotification field is null. Removing this field, and just relying on the null/not-null state of _disposedNotification would simplify things because it removes an intermediate state where Dispose has changed this flag but hasn't yet got around to nulling out the _disposedNotification field.

We use a similar technique in the Sink<...> base class that almost all of our operators use in their implementation: we use a single field to a) hold the downstream observer and b) know whether we've been disposed. (We use a sentinel observer rather than null to signify disposal, but the basic principle is the same.)

However, as noted elsewhere, perhaps this should be using Sink and Producer, at which point you might be able to defer entirely to its disposal handling.

Shouldn't this also dispose the observableDiposable?

You've checked that the act of unsubscribing doesn't cause our subscriber to be notified. But I think we also want to check that unsubscribing means we no longer get notified when the thing that normally causes notification happens.

My assumption was that this source would simply complete upon disposal. I wasn't expecting it to produce a value too.

Did you consider both options? I can see either design choice as being valid, and it's not obvious to me which is 'best'. If you thought of both, what made you choose this one.

In any case, the XML doc comments should say what the behaviour is.

Since I use it mostly with the TakeUntil Operator, I want to produce a value. I added OnComplete, because I think it is good behavior to complete a created stream.

I also like to support both scenarios.

This should test that we also get the completion notification, as well as the OnNext.

For completeness, we could also verify that we don't get an OnError.