How to create a Source that can receive elements later via a method call?

by

There are three ways this can be achieved:

1. Post Materialization with SourceQueue

You can use Source.queue that materializes the Flow into a SourceQueue:

case class Weather(zipCode : String, temperature : Double, raining : Boolean)

val bufferSize = 100

//if the buffer fills up then this strategy drops the oldest elements
//upon the arrival of a new element.
val overflowStrategy = akka.stream.OverflowStrategy.dropHead

val queue = Source.queue(bufferSize, overflowStrategy)
                  .filter(!_.raining)
                  .to(Sink foreach println)
                  .run() // in order to "keep" the queue Materialized value instead of the Sink's

queue offer Weather("02139", 32.0, true)

2. Post Materialization with Actor

There is a similar question and answer here, the gist being that you materialize the stream as an ActorRef and send messages to that ref:

val ref = Source.actorRef[Weather](Int.MaxValue, fail)
                .filter(!_.raining)
                .to(Sink foreach println )
                .run() // in order to "keep" the ref Materialized value instead of the Sink's

ref ! Weather("02139", 32.0, true)

3. Pre Materialization with Actor

Similarly, you could explicitly create an Actor that contains a message buffer, use that Actor to create a Source, and then send that Actor messages as described in the answer here:

object WeatherForwarder {
  def props : Props = Props[WeatherForwarder]
}

//see provided link for example definition
class WeatherForwarder extends Actor {...}

val actorRef = actorSystem actorOf WeatherForwarder.props 

//note the stream has not been instatiated yet
actorRef ! Weather("02139", 32.0, true) 

//stream already has 1 Weather value to process which is sitting in the 
//ActorRef's internal buffer
val stream = Source(ActorPublisher[Weather](actorRef)).runWith{...}

More Related Contents:

  1. Accessing the underlying ActorRef of an akka stream Source created by Source.actorRef
  2. sender inside a future
  3. No configuration setting found for key ‘akka.version’
  4. Akka Kill vs. Stop vs. Poison Pill?
  5. How to restrict actor messages to specific types?
  6. Akka-Stream implementation slower than single threaded implementation
  7. What is the purpose of type ascriptions in Scala?
  8. What is the advantage of using abstract classes instead of traits?
  9. Provide schema while reading csv file as a dataframe
  10. How to flatmap a nested Dataframe in Spark
  11. Scala Map implementation keeping entries in insertion order?
  12. Modify collection inside a Spark RDD foreach
  13. Scala return statements in anonymous functions
  14. Is there a reason not to use SparkContext.getOrCreate when writing a spark job?
  15. What are the various join types in Spark?
  16. What is the rule for parenthesis in Scala method invocation?
  17. How to calculate the size of dataframe in bytes in Spark?
  18. Why does partition parameter of SparkContext.textFile not take effect?
  19. Can I zip more than two lists together in Scala?
  20. Cross product in Scala
  21. Static return type of Scala macros
  22. Spark Scala list folders in directory
  23. Exception in thread “main” java.lang.NoSuchMethodError: scala.Product.$init$(Lscala/Product;)
  24. How to remove parentheses around records when saveAsTextFile on RDD[(String, Int)]?
  25. Reading in multiple files compressed in tar.gz archive into Spark [duplicate]
  26. Why is this implicit ambiguity behaviour happening?
  27. Difference between case object and object
  28. How to read a file as a byte array in Scala
  29. How to sort an RDD in Scala Spark?
  30. Insert if not exists in Slick 3.0.0

Leave a Comment