在akka流中保持物化值类型divertTo或alsoTo答案

【问题标题】：Keep materialized value type in akka stream divertTo or alsoTo在akka流中保持物化值类型divertTo或alsoTo
【发布时间】：2021-08-30 15:55:35
【问题描述】：

我尝试创建一个具有 Source 类型的 kafka 消费者流 Source<ConsumerMEssage.CommittableMessage<String,String>, Consumer.Control.

从这个来源我想使用基于某些谓词的不同路径。因此我想使用divertTo 方法，有时也使用alsoTo 方法。

这两种方法都接受 Sink。我的问题是，当我构建这个接收器时，我希望接收器具有源提供的物化值类型Consumer.Control。我现在正在做的是像这样建造水槽

private Sink<SomeType, NotUsed> sinkForPathA(){
  Flow.of(SomeType.class)
    .to(Committer.sink(committerSettings));
}

您可能会注意到，物化值类型现在是 NotUsed，这是不需要的。我想要的是这样的：

private Sink<SomeType, Consumer.DrainingControl> sinkForPathA(){
  Flow.of(SomeType.class)
    .toMat(Committer.sink(committerSettings),Consumer::createDrainingControl);
}

是否有可能创建一个具有预定义物化值类型的流，而不仅仅是NotUsed？

【问题讨论】：

您是指diverTo，还是divertTo？
感谢您指出类型。我编辑了我的问题。

标签： java akka akka-stream

【解决方案1】：

您当然可以使用内部Sink 的物化值创建Flow。

以Sink.seq()具有物化值类型：CompletionStage<List<T>>为例。

如果你说像这样创建Flow：

Flow.of(Integer.class)
    .to(Sink.seq());

然后正如您指出的那样，您将在返回的 Sink 中获得 NotUsed。诀窍是使用toMat 和keepRight 来保留Sink 的物化值类型。它会是这样的：

Flow.of(Integer.class)
    .toMat(Sink.seq(), Keep.right());

现在你会得到Sink<Integer, CompletionStage<List<Integer>>> 作为结果。

【讨论】：