akkaQueueExample.scala

akkaQueueExample.scala

 
import akka.actor._
import akka.actor.ActorLogging
import akka.actor.ActorSystem
import akka.stream._
import akka.stream.actor.ActorSubscriberMessage._
import akka.stream.actor._
import akka.stream.scaladsl._
import akka.{ NotUsed, Done }
import scala.io.Source
import java.io._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent._

implicit val system = ActorSystem("AkkaStreamTemplate")
implicit val materializer = ActorMaterializer()

val queue = akka.stream.scaladsl.Source.queue[String](100, OverflowStrategy.fail)
val (  queue2 , y) =queue.map( x=> s"yyyyyyyyyyy$x").toMat( akka.stream.scaladsl.Sink.foreach( x=>println(x)))(Keep.both).run()

queue2.offer("10")
queue2.offer("20")
queue2.offer("30")

queue2.complete