akkaストリームを使用する場合のイベントの順序

Question

akka-streamsのドキュメントを読むと、メッセージの順番や実行可能かどうかはわかりません。私がチャットサーバーのために書いた小さなコードで私の質問の文脈を設定しましょう。

def flowShape(user: User) = GraphDSL 
    .create(Source.actorRef[ChatMessage](bufferSize = 5, OverflowStrategy.fail)) { 
    implicit builder => 
     implicit chatSource => 

     import GraphDSL.Implicits._ 

     val messageFromOutside = builder.add(Flow[String].map { 
     case msg: String => UserTextMessage(user, msg) 
     case _ => InvalidMessage 
     }) 

     val merge = builder.add(Merge[ChatMessage](2)) 
     // UPDATE --> this is where the change comes 
     // val merge = builder.add(Concat[ChatMessage](2)) 

     // val channelActorSink = Sink.actorRefWithAck(channelActor, ActorInitMessage, AckMessage, UserLeft(user)) 
     val channelActorSink = Sink.actorRef(channelActor, UserLeft(user)) 

     val actorAsSource = builder.materializedValue.map { actor => UserJoined(user, actor) } 

     actorAsSource ~> merge.in(0) 
     messageFromOutside.out ~> merge.in(1) 
     merge ~> channelActorSink 

     FlowShape(messageFromOutside.in, chatSource.out) 
} 

自分のために問題を簡単にするために、私は非常に単純なソースとシンクと、このフロー形状を使用します。このようなもの -

val source = Source(List[String]("hi", "hello", "what are you upto", "this is nice")) 
val sink = Sink.foreach[ChatMessage] { 
    case tm: UserTextMessage => println(s"${tm.user.username} :: ${tm.content}") 
    case ul: UserLeft => println(s"${ul.user.username} just left the channel") 
    case uj: UserJoined => println(s"${uj.user.username} just joined the channel") 
    case _ => println(s"do not know what I just received") 
} 

val mychatchannel = new Channel(420, myactorsystem) 

source.via(mychatchannel.chatFlow(User("sushruta"))).runWith(sink) 

ここで私の心配があります。端末に印刷されるイベントの順序はまったく問題ありません。そして私はそれを修正する方法がわかりません。最初のメッセージhiが出力から欠落している

[INFO] [11/10/2017 17:42:20.431] [akka-streams-akka.actor.default-dispatcher-5] [akka://akka-streams/user/channel-actor-420] sushruta sent a message 
[INFO] [11/10/2017 17:42:20.441] [akka-streams-akka.actor.default-dispatcher-5] [akka://akka-streams/user/channel-actor-420] received a user joined message 
[INFO] [11/10/2017 17:42:20.443] [akka-streams-akka.actor.default-dispatcher-5] [akka://akka-streams/user/channel-actor-420] sushruta sent a message 
[INFO] [11/10/2017 17:42:20.444] [akka-streams-akka.actor.default-dispatcher-5] [akka://akka-streams/user/channel-actor-420] sushruta sent a message 

- ここに私が手出力です。 hiメッセージは、UserJoin messageが印刷される前に送信されたようです。

IはactorRefWithAck使用してそれを固定する（およびメッセージングの周りにいくつかの安全性を追加）しようとした（私は上記のコードでコメントアウト。）これは、同様の出力を与えます。

[INFO] [11/11/2017 06:33:03.731] [akka-streams-akka.actor.default-dispatcher-3] [akka://akka-streams/user/channel-actor-420] channel initialized and ready to take events 
[INFO] [11/11/2017 06:33:03.735] [akka-streams-akka.actor.default-dispatcher-3] [akka://akka-streams/user/channel-actor-420] sushruta sent a message 
[INFO] [11/11/2017 06:33:03.736] [akka-streams-akka.actor.default-dispatcher-3] [akka://akka-streams/user/channel-actor-420] received a user joined message 
[INFO] [11/11/2017 06:33:03.737] [akka-streams-akka.actor.default-dispatcher-4] [akka://akka-streams/user/channel-actor-420] sushruta sent a message 
[INFO] [11/11/2017 06:33:03.737] [akka-streams-akka.actor.default-dispatcher-3] [akka://akka-streams/user/channel-actor-420] sushruta sent a message 
[INFO] [11/11/2017 06:33:03.738] [akka-streams-akka.actor.default-dispatcher-3] [akka://akka-streams/user/channel-actor-420] sushruta sent a message 
[INFO] [11/11/2017 06:33:03.738] [akka-streams-akka.actor.default-dispatcher-3] [akka://akka-streams/user/channel-actor-420] received a UserLeft message 

は明らかに起こっているように見えるものをUserJoinメッセージが送信される前にソースがメッセージを送信しているということです。これをどうすれば解決できますか？概念的には、UserJoin messageをソースが実現するとすぐに、最初のメッセージを実際に送信する前に送信したいと考えています。それは可能ですか？

おかげで水道管などのストリームの

Answer 1

思う：水がある場合には、それが流れます。マージ演算子は、どちらの側の要素が来るかを気にしません。これらの入力を注文する場合は、代わりにConcatを使用してAkkaにそれを伝える必要があります。