Thread by @itrvd

As promised - a thread about cool ZStream features in the upcoming version of #ZIO!

Adapting callback-based interfaces (like RabbitMQ) to streams is always awkward and requires running effects unsafely. @dariusrobson generously contributed the implementation of Stream.effectAsync, so you can now create streams elegantly like so:

There are also additional variants, like Stream.effectAsyncInterrupt, which allows the caller to specify how to stop the emission of elements when the stream is cancelled; or Stream.effectAsyncMaybe, if sometimes we can emit the stream’s values synchronously.

Following that, Vasil Vasilev contributed the ZStream#throttleEnforce and ZStream#throttleShape combinators, so you could handily limit the throughput of a stream. This stream would emit messages at 4kb/minute:

The throttling functionality is actually implemented as a Sink, allowing for rich composition; for example, the throttling sink could be combined with a folding sink to limit the rate of element folding.

Lastly, we added two operators inspired by Akka Streams’ batch and groupedWithin operators: ZStream#aggregate and ZStream#aggregateWithin. These are real workhorses for production systems!

aggregate lets you apply a sink to elements for as long as the next transformation is busy. So this stream would aggregate messages into lists of up to 16kB for as long as persistToDb is running:

aggregateWithin aggregates elements until a timeout imposed by a Schedule expires or the Sink signals completion. So this stream would aggregate elements in lists of up to 16kB, for 5 seconds - whichever comes first:

What’s interesting about aggregateWithin is that its Schedule is fed the last aggregated element to determine the delay. So this version would decrease the timeout if the amount of messages is over 500:

You could also do fancier things: switch to another schedule entirely for a predefined period, and so forth. The Schedule’s output is also emitted to the stream if you’d like to process it.

This is a prime example of how ZIO's abstractions combine to create powerful constructs!

All of these great additions are coming in the next version of #ZIO - stay tuned!

Oops! I tagged the wrong Darius; @DariusRRobson contributed the Stream.effectAsync work, thank you!