What in the Rx Framework allows me to return an IObservable<T> while awaiting other methods during creation?

.net, .net-4.5, c#-5.0, system.reactive, task-parallel-library

Solution

I wouldn't return a `Task<IObservable<T>>`. Mixing Tasks and Observables in your public API just end up being confusing. Remember that Tasks can be thought of as observables that produce a single value. This also means don't mix CancellationTokens with observables in your public API. You control observables by subscribing and unsubscribing.

That doesn't mean you can't mix the concepts behind the scenes. Here's how to do what you want using `Observable.Using`, `Task.ToObservable` and `CancellationDisposable`

First, modify your method to return a `Task<ISourceBlock<string>>`:

public async Task<ISourceBlock<string>> CreateBlock(string parameter, CancellationToken cancellationToken)
{
     // Make some call that requires await.
     var response = await _httpClient.SendAsync(parameter, cancellationToken).ConfigureAwait(false);

     // Create a BufferBlock which will be the observable.
     var block = new BufferBlock<T>();

     // Start some background task which will use the block and publish to it
     // on a separate task.  This is not something that is awaited for.
     ConsumeResponse(response, block, cancellationToken);

     return block;
}

Now here is your new Create method that uses the method above:

public IObservable<string> Create(string parameter)
{
    // Create a cancellation token that will be canceled when the observable is unsubscribed, use this token in your call to CreateBlock.
    // Use ToObservable() to convert the Task to an observable so we can then
    // use SelectMany to subscribe to the block itself once it is available
    return Observable.Using(() => new CancellationDisposable(),
           cd => CreateBlock(parameter, cd.Token)
               .ToObservable()
               .SelectMany(block => block.AsObservable()));
}

Edit: I've since discovered Rx already has implemented this pattern with `FromAsync`:

public IObservable<string> Create(string parameter)
{
    return Observable.FromAsync(token => CreateBlock(parameter, token))
                     .SelectMany(block => block.AsObservable());
}

And also, `DeferAsync`, which is even more appropriate since your `Task` is actually creating the Observable you really want to observe (e.g. your block):

public IObservable<string> Create(string parameter)
{
    return Observable.DeferAsync(async token => (await CreateBlock(parameter, token)).AsObservable());
}

Problem

I've been working on creating an `IObservable<T>` implementation using the Reactive Extensions for Twitter's streaming APIs. From a high level an HTTP request is sent and the connection is kept open. Length-prefixed items are sent to consume. Basically, it's a call in a loop to `Stream.ReadAsync` using the `await` keyword. An `IObserver<T>` interface implementation (from `Observable.Create` or a block from the Dataflow library, it doesn't matter, it's an implementation detail) is passed to this loop which then calls the metods on the `IObserver<T>` implementation, producing the observable. However, there's a number of things that have to be done before this loop begins processing which require calls to `Task<T>`-returning methods, all of which are more easily called in C# 5.0 using the `await` keyword. Something like this: ``` public async Task<IObservable<string>> Create(string parameter, CancellationToken cancellationToken) { // Make some call that requires await. var response = await _httpClient.SendAsync(parameter, cancellationToken). ConfigureAwait(false); // Create a BufferBlock which will be the observable. var block = new BufferBlock<T>(); // Start some background task which will use the block and publish to it // on a separate task. This is not something that is awaited for. ConsumeResponse(response, block, cancellationToken); // Create the observable. return block.AsObservable(); } ``` That said, I'm currently returning a `Task<IObservable<T>>` from my method, but I feel that I'm missing something in the Reactive Extensions which would allow me to use `await` to facilitate the calls I need to make, but also return an `IObservable<T>` instead of a `Task<IObservable<T>>`. What method in the Reactive Extensions will allow me to create an observable that requires awaiting methods before returning from the creation method? The closest thing I've found is `Observable.DeferAsync`. Assuming that the call to my method and use of the observable is something along the lines of: ``` public async Task Observe() { // NOT the real name of the interface, but explains it's role here. IObservableFactory factory; // Create is really named something else. IObservable<string> observable = factory.Create("parameter"); // Subscribe. observable.Subscribe(o => Console.WriteLine("Observed: {0}", o)); // Wait. await observable; } ``` Using `DeferAsync` won't work here, as the call to `Subscribe` will send the first request, and then read off that, and then the call to `await` on `observable` will create a second subscription, but to a different observable. Or, ultimately, is returning `Task<IObservable<T>>` the appropriate method of doing this in the Reactive Framework? Subsequently, since the method returns a `Task<T>`, it's good practice to pass a `CancellationToken` to cancel the operation. That said, I can understand the `CancellationToken` being used to cancel the creation of the observable, but should it also be used to cancel the actual observable (as it could be passed on down to read the stream, etc). My gut says no, because there's a violation of separation of concerns here as well as the DRY principle with the cancellation: - The cancellation of the creation and the cancellation of the observable are two separate things. - Calling `Subscribe` is going to return an `IDisposable` implementation which will cancel the subscription.

Original source