May 13, 2021

async-streams, httpclient, ndjson

0 comments

My first post on working with NDJSON stream in .NET was a result of a use case I needed to satisfy in one of the projects I'm involved with (and it worked great). As a result, I've received some questions on this subject. I've already answered the most common one on this blog, but there were a couple more that were really frequent. So I've decided I'm going to add a few more posts to this series:

Streaming NDJSON With HttpClient

When I was writing my first post about NDJSON, I didn't know there were services out there using it as an input format. The fact is that they do, which creates a need for a way to stream NDJSON with HttpClient. The natural expectation is to be able to POST an async stream, preferably with simple and elegant code.

async IAsyncEnumerable<WeatherForecast> streamWeatherForecastsAsync()
{
    for (int daysFromToday = 1; daysFromToday <= 10; daysFromToday++)
    {
        WeatherForecast weatherForecast = await GetWeatherForecastAsync(daysFromToday).ConfigureAwait(false);

        yield return weatherForecast;
    };
};

using HttpClient httpClient = new();

using HttpResponseMessage response = await httpClient.PostAsync("...",
    new NdjsonAsyncEnumerableContent<WeatherForecast>(streamWeatherForecastsAsync())
    ).ConfigureAwait(false);

response.EnsureSuccessStatusCode();

The way to achieve the elegancy in the above snippet is through the custom implementation of HttpContent. The first part of it is the boilerplate - setting up media-type and charset, initializing serializer. I'm not going to dive into transcoding and limit the implementation to UTF-8.

public class NdjsonEnumerableContent<T> : HttpContent
{
    private static readonly JsonSerializerOptions _defaultJsonSerializerOptions = new(JsonSerializerDefaults.Web);

    private readonly IAsyncEnumerable<T> _values;
    private readonly JsonSerializerOptions _jsonSerializerOptions;

    public NdjsonAsyncEnumerableContent(IAsyncEnumerable<T> values, JsonSerializerOptions? jsonSerializerOptions = null)
    {
        _values = values ?? throw new ArgumentNullException(nameof(values));
        _jsonSerializerOptions = jsonSerializerOptions ?? _defaultJsonSerializerOptions;

        Headers.ContentType = new MediaTypeHeaderValue("application/x-ndjson")
        {
            CharSet = Encoding.UTF8.WebName
        };
    }

    ...

    protected override bool TryComputeLength(out long length)
    {
        length = -1;

        return false;
    }
}

The actual work happens in SerializeToStreamAsync. Here the goal is to enumerate the async stream, immediately serialize every incoming object and flush it over the wire together with the delimiter.

public class NdjsonEnumerableContent : HttpContent
{
    private static readonly byte[] _newlineDelimiter = Encoding.UTF8.GetBytes("\n");
    ...

    protected override async Task SerializeToStreamAsync(Stream stream, TransportContext? context)
    {
        await foreach (T value in _values.ConfigureAwait(false))
        {
            await JsonSerializer.SerializeAsync<T>(stream, value, _jsonSerializerOptions).ConfigureAwait(false);
            await stream.WriteAsync(_newlineDelimiter).ConfigureAwait(false);
            await stream.FlushAsync().ConfigureAwait(false);
        }
    }

    ...
}

This way one can fully utilize async streams when working with services that can accept NDJSON. If you are looking for something ready to use, I've made a more polished version available here with some extensions to make it even more elegant to use.