sample streams application with akka-streams 1.x ..

streams1.scala

scala> import frdomain.ch6.streams._
import frdomain.ch6.streams._

scala> import Main._
import Main._

scala> import akka.stream.scaladsl._
import akka.stream.scaladsl._

scala> import akka.stream._
import akka.stream._

scala> import common._
import common._

scala> import OnlineService._
import OnlineService._

scala> import Transaction._
import Transaction._

scala> transactions
res0: akka.stream.scaladsl.Source[frdomain.ch6.streams.Transaction,Unit] = akka.stream.scaladsl.Source@207e858e

scala> txnSink
res1: akka.stream.scaladsl.Sink[frdomain.ch6.streams.Transaction,scala.concurrent.Future[frdomain.ch6.streams.Transaction]] = akka.stream.scaladsl.Sink@5fae46df

scala> import scala.concurrent.Future
import scala.concurrent.Future


scala> val netTxn: Source[RunnableGraph[Future[Transaction]], Unit] = 
     |     transactions.map(validate).groupBy(_.accountNo).map { case (a, s) => s.toMat(txnSink)(Keep.right) }
netTxn: akka.stream.scaladsl.Source[akka.stream.scaladsl.RunnableGraph[scala.concurrent.Future[frdomain.ch6.streams.Transaction]],Unit] = akka.stream.scaladsl.Source@2e72ed0b

scala> netTxn.map(_.run()).runForeach(_.foreach(println))
res2: scala.concurrent.Future[Unit] = scala.concurrent.impl.Promise$DefaultPromise@f50422c

scala> Transaction(800259541,a-3,Credit,0,Tue Apr 26 20:59:43 IST 2016)
Transaction(2059049155,a-1,Credit,0,Tue Apr 26 20:59:43 IST 2016)
Transaction(2023562889,a-2,Debit,2000,Tue Apr 26 20:59:43 IST 2016)