Akka StreamsscalaDSL和Op-Rabbit

Question

我已经开始使用Akka Streams和Op Rabbit，我有点困惑。

我需要根据谓词拆分流，然后将它们组合起来，就像我在创建图形和使用分区和合并时所做的那样。

我已经能够使用GraphDSL. Builder做这样的事情，但似乎无法让它与AckedSource/Flow/Sink一起工作

图表如下所示：

                        | --> flow1 --> |
source--> partition --> |               | --> flow3 --> sink
                        | --> flow2 --> |

我不确定是否拆分什么时候是我应该使用的，因为我总是需要正好2个流。

这是一个不进行分区且不使用GraphDSL. Builder的示例：

def splitExample(source: AckedSource[String, SubscriptionRef],
                 queueName: String)
                (implicit actorSystem: ActorSystem): RunnableGraph[SubscriptionRef] = {
  val toStringFlow: Flow[AckTup[Message], AckTup[String], NotUsed] = Flow[AckTup[Message]]
    .map[AckTup[String]](tup => {
      val (p,m) = tup
      (p, new String(m.data))
    })

  val printFlow1: Flow[AckTup[String], AckTup[String], NotUsed] = Flow[AckTup[String]]
    .map[AckTup[String]](tup => {
      val (p, s) = tup
      println(s"flow1 processing $s")
      tup
     })

  val printFlow2: Flow[AckTup[String], AckTup[String], NotUsed] = Flow[AckTup[String]]
    .map[AckTup[String]](tup => {
      val (p, s) = tup
      println(s"flow2 processing $s")
      tup
    })

  source
    .map(Message.queue(_, queueName))
    .via(AckedFlow(toStringFlow))
    // partition if string.length < 10
    .via(AckedFlow(printFlow1))
    .via(AckedFlow(printFlow2))
    .to(AckedSink.ack)
}

这是我似乎无法使用的代码：

import GraphDSL.Implicits._
def buildModelAcked(source: AckedSource[String, SubscriptionRef] , queueName: String)(implicit actorSystem: ActorSystem):  Graph[ClosedShape, Future[Done]] = {
    import GraphDSL.Implicits._
    GraphDSL.create(Sink.ignore) { implicit builder: GraphDSL.Builder[Future[Done]] => s =>
    import GraphDSL.Implicits._
    source.map(Message.queue(_, queueName)) ~> AckedFlow(toStringFlow) ~> AckedSink.ack
//      source.map(Message.queue(_, queueName)).via(AckedFlow(toStringFlow)).to(AckedSink.ack)
    ClosedShape

}}

编译器无法解析~

所以我的问题是：

>

是否有一个使用scala dsl构建Acked/Source/Flow/Sink图表的示例项目？

是否有一个分区和合并的示例项目与我在这里尝试做的类似？




Answer 1

根据Stefano Bonetti的出色指导，这里有一个可能的解决方案：

graph:    
                        |--> short --|
  rabbitMq --> before --|            |--> after
                        |--> long  --|

解决方案：

val before: Flow[AckTup[Message], AckTup[String], NotUsed] = Flow[AckTup[Message]].map[AckTup[String]](tup => {
  val (p,m) = tup
  (p, new String(m.data))
})

val short: Flow[AckTup[String], AckTup[String], NotUsed] = Flow[AckTup[String]].map[AckTup[String]](tup => {
  val (p, s) = tup
  println(s"short: $s")
  tup
})
val long: Flow[AckTup[String], AckTup[String], NotUsed] = Flow[AckTup[String]].map[AckTup[String]](tup => {
  val (p, s) = tup
  println(s"long: $s")
  tup
})
val after: Flow[AckTup[String], AckTup[String], NotUsed] = Flow[AckTup[String]].map[AckTup[String]](tup => {
  val (p, s) = tup
  println(s"all $s")
  tup
})

def buildSplitGraph(source: AckedSource[String, SubscriptionRef]
                    , queueName: String
                    , splitLength: Int)(implicit actorSystem: ActorSystem):  Graph[ClosedShape, Future[Done]] = {
 GraphDSL.create(Sink.ignore) { implicit builder: GraphDSL.Builder[Future[Done]] => s =>
   val toShort = 0
   val toLong = 1

   // junctions
   val split = builder.add(Partition[AckTup[String]](2, (tup: AckTup[String]) => {
                                                           val (p, s) = tup
                                                           if (s.length < splitLength) toShort else toLong
                                                         }
   ))
   val merge = builder.add(Merge[AckTup[String]](2))

   //graph
   val beforeSplit = source.map(Message.queue(_, queueName)).wrappedRepr ~> AckedFlow(before).wrappedRepr
   beforeSplit ~> split
   // must do short, then long since the split goes in that order
   split ~> AckedFlow(short).wrappedRepr ~> merge
   split ~> AckedFlow(long).wrappedRepr ~> merge
   // after the last AckedFlow, be sure to '.acked' so that the message will be removed from the queue
   merge ~> AckedFlow(after).acked ~> s

  ClosedShape
}}

正如Stefano Bonetti所说，关键是使用与AckedFlow关联的. wrappeRepr，然后使用. ack组合器作为最后一步。

Answer 2

在处理确认流时，请记住以下定义。

1. AckedSource[Out， Mat]是Source[AckTup[Out]， Mat]]
的包装器
2. AckedFlow[In， Out， Mat]是Flow[AckTup[In]， AckTup[Out]， Mat]
的包装器
3. AckedSink[In， Mat]是Sink[AckTup[In]， Mat]
的包装器
4. AckTup[T]是（Promise[Unit]， T）
的别名
5. 经典流组合器将对AckTup
的 T部分进行操作
6. . ack组合器将完成AckedFlow
的 Promise[Unit]

GraphDSL边缘运算符（~

你有两条出路：

1. 您可以定义自己的~