Async Streams – A Look at New Language Features in C# 8

Matthias Koch

It’s been a while since our last part in the C# 8 series, but here it is! We will continue our journey through the new language features, and dive into async streams.

In this series, we are looking at:

• Indices, Ranges, and Null-coalescing Assignments
• Switch Expressions and Pattern-Based Usings
• Recursive Pattern Matching
• Async Streams
• Nullable Reference Types: Migrating a Codebase
• Nullable Reference Types: Contexts and Attributes

Asynchronous programming is a form of programming that prevents blocking of our application by allowing time-consuming work – like downloading data from a network – to be decoupled from the main thread. For our implementation, in principal, it requires to hand over a callback to the asynchronous method, that in turn will invoke the callback when it has finished its work.

With C# 5, the async and await keywords have been introduced. These fundamentally improved how we can write and consume asynchronous methods. Basically, the structure of our code can remain almost the same compared to a synchronous implementation, but the compiler takes over a whole lot of work to restructure our code into mentioned callbacks to make it stateful and non-blocking. This, though, only worked for void and single-result methods using Task and Task<T> as return type.

Methods producing streams of data – a.k.a. IEnumerable<T> – were left out, and required custom code to be written. However, such requirements are quite popular, thinking of working with cloud applications, collecting data from IoT sensors, or receiving data from databases. Hello, C# 8 and IAsyncEnumerable!

With the release of .NET Standard 2.1, a set of 3 interfaces was introduced: IAsyncDisposable, IAsyncEnumerable<T> and IAsyncEnumerator<T>. These interfaces allow us – in a way – to represent an asynchronous version of IEnumerable<T>, and thus to consume async streams:

public interface IAsyncDisposable
{
    ValueTask DisposeAsync();
}

public interface IAsyncEnumerable<out T>
{
    IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken token = default);
}

public interface IAsyncEnumerator<out T> : IAsyncDisposable
{
    ValueTask<bool> MoveNextAsync();

    T Current { get; }
}

For IAsyncEnumerable<T> to properly work under all circumstances, as with await in finally blocks, we also need an asynchronous version of the IDisposable interface. You may have guessed, it’s called IAsyncDisposable:

public interface IAsyncDisposable
{
    ValueTask DisposeAsync();
}

Since .NET Core SDK 3.0, those types are actually implemented and with C# 8 the foreach and using keywords can be prefixed with await to work with IAsyncEnumerable and IAsyncDisposable instances.

Iterating an IAsyncEnumerable<T> essentially means that fetching the next object becomes an asynchronous operation and the foreach body becomes our callback. Only when we continue our loop, a new item is fetched (as opposed to calling break). That basically marks this approach as pull-based. The await foreach statement also works pattern based, meaning it would work with any type that provides a suitable GetAsyncEnumerator method but doesn’t implement the IAsyncEnumerable<T> interface.

Let’s see how these additions can be used in combination. Curious developers should definitely take the chance and look at its decompiled code:

static async Task Main()
{
    await foreach (var data in ReadAndNormalize())
    {
        Console.WriteLine(data);
    }
}

static async IAsyncEnumerable<Data> ReadAndNormalize(
    [EnumeratorCancellation] CancellationToken token = default)
{
    while (!token.IsCancellationRequested)
    {
        var data = await ReadInputFromDevice();
        var normalizedData = ConvertAndNormalize(data);
        yield return normalizedData;
    }
}

static async Task<byte[]> ReadInputFromDevice() { return default; }

static Data ConvertAndNormalize(byte[] data) { return default; }

struct Data { }

Note that the CancellationToken is marked with an EnumeratorCancellation attribute. Similar to attributes from the JetBrains.Annotations package, the EnumeratorCancellation attribute (contained in System.Runtime.CompilerServices) marks code fragements for additional magic, that cannot (yet) be expressed with the language itself.

As mentioned earlier, using statements can now also be prefixed with the await keyword. Let’s see that in action, too!

async Task WriteToFileAsync(string file)
{
    await using (var fs = new FileStream(file, FileMode.CreateNew))
    await using (var sw = new StreamWriter(fs))
    {
        await sw.WriteAsync("Hello C#8!");
    }
}

Since .NET Standard 2.1, the two involved types FileStream and StreamWriter are both implementing IAsyncDisposable. This allows all the disposal work, like flushing changes to disk, to be done asynchronously.

How do ReSharper and Rider help with async streams?

Besides parsing the new language constructs, ReSharper and Rider bring several other goodies that help to work with async streams.

Starting with the most obvious operation we can do with enumerables, the foreach postfix template has been updated to complete to await foreach when the enumerable in question is of the type IAsyncEnumerable<T> or matches its pattern:

In the last example, we’ve made a mistake. We didn’t pass the cancellation token! Luckily, there’s a new code inspection, and a handy quick-fix to make things right:

Hold on, there’s another tricky example, where ReSharper and Rider will notice that a cancellation token can be passed using the WithCancellation extension method. Remember the EnumeratorCancellation attribute we’ve mentioned? This will allow the compiler to substitute every usage of the cancellation token with the effective cancellation token that is combined with the original token for the IAsyncEnumerator:

Whenever we want to return an IAsyncEnumerable<T> but the method is not marked as async yet, ReSharper and Rider will suggest to do so from either the method name region or from await keyword occurrences:

Equally, when we’re working with methods that have already been marked as async, ReSharper and Rider will allow changing from mistakenly defined IEnumerable<T> return types to IAsyncEnumerable<T>:

On the consuming side, another quick fix allows changing synchronous foreach statements to asynchronous await foreach statements. This can be of great use when converting existing code to make use of the new language feature:

Last but not least, let’s get back to IAsyncDisposable and await using. If a method is async, and the object in question implements an asynchronous counterpart DisposeAsync, Rider and ReSharper will show an inspection, that await using can be used instead. Also, we can turn the await using statements into declarations as described in pattern-based usings. The corresponding quick-fixes can be applied in scope:

Download ReSharper Ultimate 2019.2 or check out Rider 2019.2 to start taking advantage of ReSharper’s language support updates and C# 8. We’d love to hear your thoughts!

Subscribe to a monthly digest curated from the .NET Tools blog:

Subscribe form

By submitting this form, I agree to the JetBrains Privacy Policy Notification icon

By submitting this form, I agree that JetBrains s.r.o. ("JetBrains") may use my name, email address, and location data to send me newsletters, including commercial communications, and to process my personal data for this purpose. I agree that JetBrains may process said data using third-party services for this purpose in accordance with the JetBrains Privacy Policy. I understand that I can revoke this consent at any time in my profile. In addition, an unsubscribe link is included in each email.

Submit

Thanks, we've got you!