A Source is the producer stage in an Akka Streams pipeline. It emits elements upstream of a Flow and Sink, and remains lazy until the stream is run. Sources can read collections, files, database queries, topics, queues, or HTTP endpoints. section 10.1.1 section 10.1.3
Build a stream by combining three stages:
- Source produces elements. section 10.1.2
- Flow transforms elements between the producer and consumer. section 10.1.1
- Sink consumes elements and can provide a materialized result such as Future[Done]. section 10.1.7
Filter a finite source
Source(List(1, 2, 3)).filter(_ % 2 == 0).runForeach(storeDB) creates a runnable stream, filters out odd values, and stores 2 in a fake database. The blueprint performs no work until runForeach runs it. section 10.1.3
Listing 10.1 Stream components
object Main extends App {
implicit val system = ActorSystem(Behaviors.empty,"runner") #1
var fakeDB: List[Int] = List()
def storeDB(value: Int) = fakeDB = fakeDB :+ value
val producer: Source[Int, NotUsed] = Source(List(1, 2, 3)) #2
val processor: Flow[Int, Int, NotUsed] = Flow[Int].filter(_ % 2 == 0) #3
val consumer: Sink[Int, Future[Done]] = Sink.foreach(i => storeDB(i)) #4
val blueprint: RunnableGraph[scala.concurrent.Future[akka.Done]] =
producer.via(processor).toMat(consumer)(Keep.right) #5
val future: Future[Done] = blueprint.run() #6
future.onComplete( _ => system.terminate)(system.executionContext) #7
}
Shows the producer, processor, consumer, blueprint, materialization, and completion handling as separate Scala values. section 10.1.7
