Observable.Using with async Task

user_1025407 0
Viewed 1359

I have used Observable.Using with methods that return an IDisposable in the way:

Observable.Using(() => new Stream(), s => DoSomething(s));

But how would we proceed when the stream is created asynchronously? Like in this:

Observable.Using(async () => await CreateStream(), s => DoSomething(s));

async Task<Stream> CreateStream() 
{
    ...
}

DoSomething(Stream s)
{
    ...
}

This doesn't compile because it says that s is Task<Stream> instead a Stream.

What's the deal?

observable system.reactive .net c#
2 Answers

So the problem here is that there are two overloads with very awkward type signatures.

When you pass a Task<TResource> returning function as the first argument it seems to infer that the second argument, the "observable factory", is also going to return a Task but inference then fails because, in your example, there is no parameter declared for cancellation token by either function and both require that parameter. Inference fails in a myriad of ways and it gets confusing.

I will try to put it into perspective with an example that is realistic.

Note: writing code like the following is probably a bad idea but it is at least real world type use case.

var connectionString = @"Data Source=.\SQLEXPRESS;Integrated Security=SSPI;app=LINQPad";

Observable.Using(
        resourceFactoryAsync: async ct => await ConnectAsync(connectionString, ct),
        observableFactoryAsync: async (connection, ct) => await QueryAsync(connection, ct)
    )
    .Subscribe(
        onNext: Console.WriteLiner,
        onError: Console.Error.WriteLine
    );

async Task<IObservable<string>> QueryAsync(SqlConnection c, CancellationToken ct = default)
{
    var command = c.CreateCommand();
    command.CommandText = "select * from Categories as c order by c.CategoryId";
    var enumerableQuery = 
        from IDataRecord record in await command.ExecuteReaderAsync(ct)
        select (string)record["CategoryName"];
    return enumerableQuery.ToObservable();
}

async Task<SqlConnection> ConnectAsync(string cs, CancellationToken ct = default)
{
    var connection = new SqlConnection(cs);
    await connection.OpenAsync(ct);
    return connection;
}

Note: to the use of named arguments is just to document which overload is being selected.

That was the best I could come up with but feel free to edit this with better examples.

Somethings that are noteworthy are

  • All of this comes crashing down completely, compilation failing, if a nullary function is provided for either resourceFactoryAsync, or observableFactoryAsync. You must declare the CancellationToken as a formal parameter of both the resource factory and the observable factory functions. They are both named ct in the code above for the sake of brevity. While we are just passing them along, what matters is that they are declared.

  • notice that the connection parameter of the observableFactoryAsync is not an awaitable but the actual resource. The error you received, prompting you to naturally assume that the resource should be awaited, was due to type inference selecting the synchronous overload and thus forwarding TResource as a Task<X> into the next function which was also assumed to be synchronous.

  • The way these abstractions compose feels rather awkward, at least to me. We are mixing multiple asynchronous and synchronous programming styles and we have IDisposables mixed up into it such that we are creating them explicitly, asynchronously, but who knows how they will be disposed.

  • The QueryAsync function actually creates a secondary IDisposable, an SqlCommand, which I would normally wrap in a using block but cannot because I need the lazy evaluation of the query preserved and so I need the command to not have been disposed when the closure returned is called.

  • There is probably a better way to do this, but I wonder if it involves calling Observable.Using multiple times and then merging them. I am basically just rambling right now about how complicated the Observable monad can feel (granted I am a novice Rx user) but it really feels like it wants your entire application to be wrapped inside of it.

user_19158930
Related