Send message to AkkaStream

SendMessageToAkkaStream.scala

import akka.actor.{ActorSystem, Props}
import akka.stream.{ActorMaterializer, OverflowStrategy}
import akka.stream.scaladsl.{Flow, Sink, Source}
import com.typesafe.config.ConfigFactory
import jetbrains.exodus.entitystore.{PersistentEntityStoreImpl, PersistentEntityStores}
import jetbrains.exodus.env.{Environments, Store}
import org.apache.commons.lang3.SerializationUtils

import collection.JavaConverters._
import scala.beans.BeanProperty

object MembersService extends App {
  val config = ConfigFactory.load.getConfig("MembersService")

  implicit val system = ActorSystem("MembersService", config)
  import system.dispatcher

  val worker = system.actorOf(Props[Worker], "remote-worker")

  println(s"Worker actor path is ${worker.path}")

  val weatherSource = Source.actorRef[Weather](Int.MaxValue, OverflowStrategy.fail)

  val sunnySource = weatherSource.filter(!_.raining)
  val temp = Flow[Weather].map{t => t.temp}
  val sink = Sink.foreach[Double](println)

  implicit val flowMaterailiser = ActorMaterializer()

  val ref2 = Flow[Weather]
    .via(temp)
    .to(sink)
    .runWith(sunnySource)

  println(s"ref2 actor path is ${ref2.path}")

  ref2 ! Weather("02139", 32.0, true)
  ref2 ! Weather("02139", 31.0, false)
  ref2 ! Weather("02139", 444.0, false)
}

case class Weather(zip : String, temp : Double, raining : Boolean)

case class Foo(name:String)