Embrace Rx Zen Nature

[anaisbetts]

This PR cleans up a lot of the interface that Rx users would use, to something more idiomatic Rx'y. Note that I haven't actually tested any of this code and it may be completely broken, but it's On The Right Track™ :)

[itay]

@paulcbetts you should change your user name to "Paul Betts™" :)

[dahlbyk]

In the previous commit, once you called AsObservable(), we kept trucking on forever even if the last subscriber dropped off early, until the stream was exhausted.

Instead, have at most one underlying person doing the reads and the rest recieving that result, and if everybody unsubscribes early, we don't Dispose and we stop reading the stream altogether.

So for SearchResults.AsObservable(), ReadResultAsync will be called as soon as someone subscribes, and continue to be called until either:

1. Everyone has unsubscribed, in which case the observable becomes idle until a new subscriber comes along; or
2. The stream completes via read returning null (or an error), in which case the SearchResults will be Disposed and all subscribers will receive a completed event (and maybe an error).

[anaisbetts]

@dahlbyk Nailed it

[glennblock]

Hi @paulcbetts thanks for sending this. @David-Noble-at-work and I are reviewing the suggested changes.

[glennblock]

Hi @paulcbetts

Thanks for this PR and for the effort! After consulting with @headinthebox, we've decided not to make this change. The reasoning is that we don't want to require customers to pull in Rx packages in order to use our SDK. With this current model you can use Rx if you need it.

@headinthebox can you weigh in?

Thanks! Glenn

[headinthebox]

It is perfectly valid for producers of IObservable to provide their own implementation with pulling in the whole of Rx, as long as they respect the Rx contract.

The real solution would be to have the source code for a conformant standard implementation of IObservable available for people to re-use (instead of binary reuse of the library). This is the second time I ran into this issue.

Once you have an observable in hand, you can pull in Rx yourself.

[As to wether to be or be convertable to is independent of Observable or Enumerable, i.e. if you are one you should also be the other and vice versa].

[glennblock]

@headinthebox thanks! I agree having this code readily available makes sense. Sounds like a perfect job for a Nuget source package!

[anaisbetts]

Sounds like a perfect job for a Nuget source package!

I don't see how you can do that as a source package, I had to use the class internals to do it. If you're not going to take this change, please remove the existing Rx code and leave IObservable out completely, it's pretty broken and incorrect

[glennblock]

@paulcbetts we had @headinthebox just review the code and he found no issues. What specifically are you seeing as a problem?

The source package would be a way to distribute a default implementation of IObservable.

[anaisbetts]

The ones that stand out are:

• Returning Task<Observable<T>> is Weird™
• Subscribing doesn't start reading the content, and unsubscribing doesn't stop it from being read
• The Observable is Hot, so anyone who subscribes late misses items (combined with #1 above, I don't know how you can possibly not lose events

Maybe I didn't read the existing code carefully enough though, I may be missing something. I understand the desire to not bolt on all of Rx, but why not just pull in the code for those operators, or use ILMerge?

[headinthebox]

What you point out @paulcbetts are bugs that should be fixed; but the main thing is that we should make it easy for folks to expose IObservable without needing to pull in the whole shebang.