akka - Akka-streams：如何将流连接到流？

Question

如果流 A 产生

Pair<Source<ByteString,?>, Object> 

如何将此出口与在源上映射的 Flow B 入口连接。例如 Flow B 的出口是

Pair<InputStream<Long>,Object>.

Answer 1

如果您获得了您提供的类型的来源：

Source< Pair<Source<ByteString, BoxedUnit>, Object>, BoxedUnit> pairSource;

并且您还将获得一个将ByteString值转换为的流程InputStream<Long>：

Flow<ByteString, InputStream<Long>, ?> byteStrToInputStream;

然后可以使用 flatMapConcat 进行转换：

Source< Pair<InputStream<Long>,Object>, BoxedUnit> inputStreamSource = 
  pairSource.flatMapConcat( pair -> {
    pair.getKey()
        .via(byteStrToInputStream)
        .map(inputStream -> ImmutablePair(inputStream, pair.getValue()))
  })

ByteStrings 的每个传入 Sources 都被耗尽，结果被连接起来。每对的原始右侧，即Object添加到每个InputStream。