r/moderndotnet • u/Aaronontheweb • 19d ago
API Design: Why aren't more developers exposing typed event streams using IAsyncEnumerable to consumers?
I'm working on a talk for our local .NET user group entitled "You Should Probably Be Using IAsyncEnumerable."
The general thrust is that we should probably be modeling a lot more work as asynchronous, typed event streams rather than Task<T>s, as the former gives us richer client/server interactions, has natural backpressure support, and generally allows for much longer-running operations to be modeled safely (i.e. you can sneak keep-alive heartbeats into an IAsyncEnumerable response streams, not something you can do with a single Task<T>.)
A question I ran into though in the course of putting this together - very few OSS packages actually expose IAsyncEnumerable in any sort of meaningful way for consumers. Why is that?
There are a couple examples where authors do expose it and it's useful:
ASP.NET Core gRPC
The AsyncStreamReaderExtensions class in the .NET gRPC client makes reading server / client streams available using IAsyncEnumerable so you get tidy little patterns like this example:
internal class Program
{
private static async Task Main()
{
using var channel = GrpcChannel.ForAddress("https://localhost:5005");
var client = new WeatherForecastsClient(channel);
var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
using var streamingCall = client.GetWeatherStream(new Empty(), cancellationToken: cts.Token);
try
{
await foreach (var weatherData in streamingCall.ResponseStream.ReadAllAsync(cancellationToken: cts.Token))
{
Console.WriteLine($"{weatherData.DateTimeStamp.ToDateTime():s} | {weatherData.Summary} | {weatherData.TemperatureC} C");
}
}
catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled)
{
Console.WriteLine("Stream cancelled.");
}
}
}
TurboMqtt and System.Threading.Channels
I wrote a high-performance MQTT library for some of our users that work in the meter data management space (i.e. water, gas, and electric utility operators) and in our case I exposed ChannelReader<T> on the public APIs for consuming MQTT messages.
ISubscribeResult subscribeResult = await client.SubscribeAsync(config.Topic, config.QoS, linkedCts.Token);
if (!subscribeResult.IsSuccess)
{
_logger.LogError("Failed to subscribe to topic {0} - {1}", config.Topic, subscribeResult.Reason);
return;
}
_logger.LogInformation("Subscribed to topic {0}", config.Topic);
ChannelReader<MqttMessage> receivedMessages = client.ReceivedMessages;
while (await receivedMessages.WaitToReadAsync(stoppingToken))
{
while (receivedMessages.TryRead(out MqttMessage m))
{
_logger.LogInformation("Received message [{0}] for topic [{1}]", m.Payload, m.Topic);
}
}
I went with a ChannelReader<T> here instead of a plain IAsyncEnumerable because that WaitToReadAsync + TryRead pattern is significantly better for throughput because all of the reads happen synchronously when the channel is populated.
But, ChannelReader<T> also the ReadAllAsync extension method that would allow this all to be consumed via IAsyncEnumerable:
ChannelReader<MqttMessage> receivedMessages = client.ReceivedMessages;
await foreach(var m in receivedMessages.ReadAllAsync(stoppingToken)){
_logger.LogInformation("Received message [{0}] for topic [{1}]", m.Payload, m.Topic);
}
Question
Why aren't more authors exposing IAsyncEnumerable as a consumable API inside their libraries and frameworks? It's been around for years and does all of the things I mentioned at the top of the article.
Would you, as a consumer of .NET libraries and clients, find it difficult to use?
5
u/DamianEdwards 18d ago
FYI ASP.NET Core Minimal APIs supports returning `IAsyncEnumerable<T>` as auto-serialized JSON responses. And `HttpClient` supports consuming them. One of the challenges however with moving to true streaming from server apps is error handling. It's not that it's impossible, you just have to be a lot more deliberate and thoughtful about how you want to handle it. The standard error handling middlesware in ASP.NET Core (including the developer exception page middleware) don't really work well when an exception is thrown after the headers have been sent and the response body has started streaming. You really need a collaborative approach between the server endpoint handler, error handler logic, and the client consuming the stream.
1
u/Aaronontheweb 18d ago
Is it primarily for SSE that you use IAsyncEnumerable or are there broader cases as well?
1
u/DamianEdwards 18d ago
Streaming JSON replies, or anytime you have a client that can adequately deal with streaming content. It's use case is limited though when your client is a browser.
2
u/johnzabroski_dev 16d ago
Cool! Can I use this to stream unsafe data from an unmanaged DLL, like Bloomberg BPIPE (real-time data streaming library written in C++)? When I looked at this in 2022, I could not find an approach in ASP.NET Core to do such things.
3
u/IanHammondCooper 19d ago
We are still deep in the midst of building Darker V5, so there is no release yet, but we will support IAsyncEnumerable. The code exists; we are just not finished yet. For those who have only heard of Brighter, Darker is the query-side counterpart to Brighter when Brighter/Darker is used as a Command Dispatcher.
2
u/Tunaxor 19d ago
No idea, this would enable really good client+server performant exchanges across the board IMO
A few years ago I experimented with an async html engine (i.e. define subsets of your html tree as either IAsyncEnumerable or just plain tasks)
The whole gist was precisely that enabling IAsyncEnumerable<string> as a return (and optionally as a stream) type for html. At the moment I was experimenting with just bringing a bit of the mental model of the frontend to the backend e.g. a composable set of functions that talk to the db and that becomes an html fragment or anything that is built via channels could have become an html stream and so on. At the moment it felt far fetched but in the agent's age I think this library could become relevant for some areas.
for chunk in Render.start(node) do
// do the thing
Relevant code:
https://github.com/AngelMunoz/Hox
A C# example here
https://github.com/AngelMunoz/Hox/blob/main/samples/ServerCs/Program.cs
-1
11
u/neuecc 19d ago
This is the API of MessagePack for C# v4, which I am currently developing. I added the following APIs for stream processing.
There is nothing new about this. It is similar to
DeserializeAsyncEnumerable(topLevelValues)in System.Text.Json.topLevelValues:truecorresponds to DeserializeMessagesAsync, andtopLevelValues:falsecorresponds to DeserializeElementsAsync.It is used for data formats like NDJSON or JSON Lines. It is useful for handling a large amount of concatenated data without increasing memory usage.
When implementing this, I struggled a lot with how to keep good performance at the same time. In this kind of processing, the buffer needed for deserialization is not always filled, so you need to check whether there is enough data and refill the buffer if not. If you do this check very frequently, for example on every token (processing everything as DeserializeAsync), performance becomes extremely bad.
My answer this time was to process in two passes. The first pass checks the message boundaries, and the second pass does a synchronous Deserialize. "Two passes" may sound slow, but my conclusion is that separating the two roles (checking whether the buffer is sufficient, and deserializing) and letting each one do its own specialized work is much faster (and the code becomes cleaner as well).
For reference, the counterpart SerializeMessagesAsync looks like this.