2

I am building a client for continuously consuming new records from a data source. The integration is pull-based and my client periodically queries the data source for new records. I'm using IAsyncEnumerable as a return type for this continuous stream of new records. Here is the gist of the method in question:

public async IAsyncEnumerable<Record> StreamRecords(...)
{
    while (!cancellationToken.IsCancellationRequested)
    {
        using var response = await _httpClient.SendAsync(request, cancellationToken);

        var records = _parser.Parse(response.Content);

        foreach (var r in records) yield return r;
        
        await Task.Delay(waitPeriod, cancellationToken);
    }
}

Is this an appropriate use of IAsyncEnumerable? This stream should be "never ending" or "continuous" (at least until cancellationToken or error).

Wesley Thorsen
  • 93
  • 2
  • 8
  • 1
    I would consider this to be acceptable usage, as long as the behavior is clear from the documentation and method name. But it might be opinion-based. – JonasH Oct 07 '21 at 07:46
  • *"my client periodically queries the data source for new records"*: Where do you store the information about what records have already been consumed by the client? Is this info stored client side or server side? – Theodor Zoulias Oct 07 '21 at 11:34
  • Client side. The real implementation of this method handles this. – Wesley Thorsen Oct 07 '21 at 18:10
  • 1
    As a side note, the `while (!cancellationToken.IsCancellationRequested)` combined with the `await Task.Delay(waitPeriod, cancellationToken);` can cause inconsistent cancellation behavior. When the `cancellationToken` is signaled, the asynchronous sequence may terminate either successfully or as canceled, depending on the timing of the cancellation. The problem is the `IsCancellationRequested`. The `ThrowIfCancellationRequested` should be used instead, if needed (in this case it's probably not needed). – Theodor Zoulias Oct 07 '21 at 18:41

2 Answers2

3

As with most things: the key point is that the intent and behaviour is clearly communicated. I see no conceptual issue over a non-ending sequence, whether that is IEnumerable<T> or IAsyncEnumerable<T> - but obviously if the consumer calls ToList[Async](), things are going to end badly. This is not in itself a problem - to give a similar scenario: some I[Async]Enumerable<T> sequences are non-repeatable - either they can only be iterated once (with subsequent attempts failing), or they could yield different results each time (without even requiring things like list changes). This is perfectly legal, yet code exists in the wild that assumes that sequences are repeatable and will yield the same data. The same discussion exists here, and ultimately, it isn't the producer at fault, but the consumer.

So: your sequence (producer) will work perfectly fine with a reasonable consumer that understands that the data should be treated as unbounded. If that's what your application has, then: great!

Marc Gravell
  • 1,026,079
  • 266
  • 2,566
  • 2,900
1

Your StreamRecords method returns essentially a consuming sequence, similar in nature with the sequences returned by the methods BlockingCollection<T>.GetConsumingEnumerable and ChannelReader<T>.ReadAllAsync. Consuming means that when the caller enumerates the sequence, the returned elements are permanently removed from some backing storage. In the case of these two methods the backing storage is an internal ConcurrentQueue<T>. In your case (based on this comment) the backing storage is located server-side, with some client-side code that knows what data to fetch next.

Exposing a consuming sequence comes with some challenges:

  1. Should the method have the word Consuming or Destructive as part of its name?
  2. What happens in case of cancellation? Is the sequence responsive enough when a cancellation signal arrives? Are the caller's expectations met?
  3. What happens if the caller abandons the enumeration prematurely, either by deliberately breaking or returning from the await foreach loop, or unwillingly by a transient exception thrown inside the loop? Are any consumed elements in danger of being lost?

Regarding the 1st challenge you can read this GitHub issue, or even better watch this video, where the various options were debated. Spoiler alert, the Microsoft engineers settled for the seemingly innocuous ReadAllAsync.

Regarding the 2nd challenge you can read this question, showing that the (technically justified) decision made by Microsoft regarding the ChannelReader<T>.ReadAllAsync API, resulted in non-intuitive/unexpected behavior.

Regarding the 3rd challenge you could consider taking advantage of the mechanics of the finally blocks inside iterator methods. Check out this answer for more details.

Because of these nuances, it might be a good idea to give additional consuming options to the callers. Something like public Task<Record[]> TakeAllNewRecords() for example.

Theodor Zoulias
  • 34,835
  • 7
  • 69
  • 104
  • 1
    Very helpful, thank you. For the curious, the real consuming client is here: https://github.com/wesleythorsen1/dotnet-craigslist/blob/master/src/DotnetCraigslist/CraigslistStreamingClient.cs – Wesley Thorsen Oct 07 '21 at 20:15