title: How to batch an IAsyncEnumerable<T>, enforcing a maximum interval policy between consecutive batches?
body: <p>I have an asynchronous sequence (stream) of messages that are arriving sometimes numerously and sometimes sporadically, and I would like to process them in batches of 10 messages per batch. I also want to enforce an upper limit to the latency between receiving a message and processing it, so a batch with fewer than 10 messages should also be processed, if 5 seconds have passed after receiving the first message of the batch. I found that I can solve the first part of the problem by using the <a href="https://github.com/dotnet/reactive/blob/main/Ix.NET/Source/System.Interactive.Async/System/Linq/Operators/Buffer.cs" rel="nofollow noreferrer"><code>Buffer</code></a> operator from the <a href="https://www.nuget.org/packages/System.Interactive.Async/" rel="nofollow noreferrer">System.Interactive.Async</a> package:</p>
  <pre><code>IAsyncEnumerable&lt;Message&gt; source = GetStreamOfMessages();
  IAsyncEnumerable&lt;IList&lt;Message&gt;&gt; batches = source.Buffer(10);
  await foreach (IList&lt;Message&gt; batch in batches)
  {
      // Process batch
  }
  </code></pre>
  <p>The signature of the <code>Buffer</code> operator:</p>
  <pre><code>public static IAsyncEnumerable&lt;IList&lt;TSource&gt;&gt; Buffer&lt;TSource&gt;(
      this IAsyncEnumerable&lt;TSource&gt; source, int count);
  </code></pre>
  <p>Unfortunately the <code>Buffer</code> operator has no overload with a <code>TimeSpan</code> parameter, so I can't solve the second part of the problem so easily. I'll have to implement somehow a batching operator with a timer myself. My question is: how can I implement a variant of the <code>Buffer</code> operator that has the signature below?</p>
  <pre><code>public static IAsyncEnumerable&lt;IList&lt;TSource&gt;&gt; Buffer&lt;TSource&gt;(
      this IAsyncEnumerable&lt;TSource&gt; source, TimeSpan timeSpan, int count);
  </code></pre>
  <p>The <code>timeSpan</code> parameter should affect the behavior of the <code>Buffer</code> operator like so:</p>
  <ol>
  <li>A batch must be emitted when the <code>timeSpan</code> has elapsed after emitting the previous batch (or initially after the invocation of the <code>Buffer</code> method).</li>
  <li>An empty batch must be emitted if the <code>timeSpan</code> has elapsed after emitting the previous batch, and no messages have been received during this time.</li>
  <li>Emitting batches more frequently than every <code>timeSpan</code> implies that the batches are full. Emitting a batch with less than <code>count</code> messages before the <code>timeSpan</code> has elapsed, is not desirable.</li>
  </ol>
  <p>I am OK with adding external dependencies to my project if needed, like the <a href="https://www.nuget.org/packages/System.Interactive.Async/" rel="nofollow noreferrer">System.Interactive.Async</a> or the <a href="https://www.nuget.org/packages/System.Linq.Async" rel="nofollow noreferrer">System.Linq.Async</a> packages.</p>
  <p>P.S. this question was inspired by <a href="https://stackoverflow.com/questions/67573683/channels-with-cancellationtokensource-with-timeout-memory-leak-after-dispose">a recent question</a> related to channels and memory leaks.</p>