根据源的物化值创建接收器

tabdulradi 发表于 Dev

塔布杜拉迪

我想围绕Pubsub主题构建流程

\-----------------------------------------------------\
 \  ------------------  ------------    -------------  \
  > > wrapWithPublish > > toPubsub |    | fromPubsub >  >
 /  ------------------  ------------    -------------  /
/-----------------------------------------------------/

这是我到目前为止编写的代码

def mediatorFlow[In, Out](mediator: ActorRef, topic: String): Flow[In, Out, Unit] = {
  val source =
    Source
      .actorRef[Out](10, OverflowStrategy.dropHead)
      .mapMaterializedValue { ref => mediator ! DistributedPubSubMediator.Subscribe(topic, ref); ref }

  val wrapWithPublish =
    Flow[In].map(DistributedPubSubMediator.Publish(topic, _))

  val unsubscribe = DistributedPubSubMediator.Unsubscribe(topic, ref???)

  val toPubsub =
    Sink.actorRef[DistributedPubSubMediator.Publish](mediator, unsubscribe)

  Flow.fromSinkAndSource(wrapWithPublish to toPubsub, source)
}

问题在于的定义unsubscribe，我想DistributedPubSubMediator.Subscribe在流的末尾发送a ，它指定aref应该是source上面定义的物化值。

我知道，当Pubsub在流的末尾死亡时，它将自动取消订阅Actor。但是我很好奇无论如何如何解决问题。

罗兰·库恩

为了实现这一点，您将需要构建一个比紧密结合的Flow fromSinkAndSource，您将需要使用GraphDSL：

val source = ... // as above
Flow.fromGraph(GraphDSL.create(source) { implicit b =>
  src =>
    import GraphDSL.Implicits._
    val concat = b.add(Concat[Any](2))
    val wrapWithPublish = b.add(Flow[In].map(DistributedPubSubMediator.Publish(topic, _)))
    val toPubSub = b.add(Sink.actorRef[Any](mediator, unsubscribe))

    wrapWithPublish ~> concat ~> toPubSub
    b.materializedValue.map(DistributedPubSubMediator.Unsubscribe(topic, _)) ~> concat

    FlowShape(wrapWithPublish.in, src.out)
})

这样，您可以将其中一部分的物化值注入到流元素的级别中，以使其易于发送给pubsub介体。

本文收集自互联网，转载请注明来源。

如有侵权，请联系[email protected] 删除。