fizzbuzz-cloudflow-printer.scala

package fizzbuzz

import cloudflow.akkastream._
import cloudflow.akkastream.scaladsl._
import cloudflow.streamlets._
import cloudflow.streamlets.avro._

class FizzBuzzPrinter extends AkkaStreamlet {
  val inlet = AvroInlet[FizzBuzzResult]( "in" )
  val shape = StreamletShape.withInlets( inlet )

  override def createLogic = new RunnableGraphStreamletLogic() {

    def flow = {
      FlowWithOffsetContext[FizzBuzzResult]
        .map { result ⇒
          system.log.info( result.toString )
          result
        }
    }

    def runnableGraph = {
      sourceWithOffsetContext( inlet ).via( flow ).to( sinkWithOffsetContext )
    }
  }
}