Last active
March 1, 2020 21:47
-
-
Save stevebakh/39c47e55e04f0c09ad01849685416f6d to your computer and use it in GitHub Desktop.
Example usage of zio-streams showing lazy evaluation and transformation.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import zio.console.{putStrLn, Console} | |
| import zio.stream.{Stream, ZSink, ZStream} | |
| import zio.{App, IO, ZIO} | |
| object Main extends App { | |
| val stream: ZStream[Console, Nothing, Int] = ZStream.unfoldM(1) { count => | |
| for { | |
| _ <- putStrLn(s"create stream item: $count") | |
| } yield if (count <= 10) Some(count -> (count + 1)) else None | |
| } | |
| val expandedStream: ZStream[Console, Nothing, Char] = stream | |
| .tap(count => putStrLn(s"expanding stream item $count to additional items of different type")) | |
| .flatMap(_ => Stream.fromIterable('a' to 'c')) | |
| val program: ZIO[Console, Nothing, Unit] = for { | |
| _ <- putStrLn("Running sink on stream...") | |
| _ <- expandedStream.run(ZSink.foldLeftM(1) { (count: Int, streamItem: Char) => | |
| putStrLn(s"Consuming stream, iteration: $count, with item: $streamItem") *> IO.succeed(count + 1) | |
| }) | |
| _ <- putStrLn("Finished reading from stream.") | |
| } yield () | |
| override def run(args: List[String]): ZIO[zio.ZEnv, Nothing, Int] = program.as(0) | |
| } |
Author
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Showcases the lazy nature of streams using zio-streams, by lazily transforming and expanding a stream.
I put this example together to help me understand how I would use zio-streams to expose a ZStream from a method responsible for loading data from a DynamoDB table. The stream represents the initial wrapping of the query result in the stream, where each iteration of the unfoldM is invoked for each page of results left to load. The expandedStream represents a transformation of the query results. In concrete terms, this would be where the items from each page are transformed into domain models in the application.