Comment créer une Source que l'on peut recevoir des éléments plus tard par l'intermédiaire d'un appel de méthode?

Il y a trois façons cela peut être réalisé:

1. Post Matérialisation avec SourceQueue

Vous pouvez utiliser Source.queue qui matérialise l'Écoulement dans un 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 Matérialisation avec l'Acteur

Il y a une question similaire, et de répondre à ici, l'essentiel étant que vous matérialiser le flux comme un ActorRef et envoyer des messages pour que 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. Pré Matérialisation avec l'Acteur

De la même manière, vous pouvez créer explicitement un Acteur qui contient un tampon de message, utilisez qu'un Acteur, pour créer une Source, puis de l'envoyer que des messages d'acteurs comme indiqué dans la réponse ici:

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{...}