Alpakka AMQP：如何检测声明异常？

Question

我有一个带有声明的 AMQP 源和 AMQP 接收器：

List<Declaration> declarations = new ArrayList<Declaration>() {{
      add(QueueDeclaration.create(sourceExchangeName));
      add(BindingDeclaration.create(sourceExchangeName, sourceExchangeName).withRoutingKey(sourceRoutingKey));
    }};

amqpSource = AmqpSource
    .committableSource(
        NamedQueueSourceSettings.create(connectionProvider, sourceExchangeName)
            .withDeclarations(declarations),
        bufferSize);

AmqpWriteSettings amqpWriteSettings = AmqpWriteSettings.create(connectionProvider)
    .withExchange("DEST_XCHANGE")
    .withRoutingKey("ROUTE123")
    .withDeclaration(ExchangeDeclaration.create(destinationExchangeName,
        BuiltinExchangeType.DIRECT.getType()));

amqpSink = AmqpSink.create(amqpWriteSettings);

然后我有一个流程..

amqpSource.map(doSomething).async().map(doSomethingElse).async().to(amqpSink)

现在，在我启动应用程序后，发送到源队列的消息没有被消耗。后来我发现这是由于声明期间发生的错误。 （即，当我在 Source 和 Sink 设置中删除 .withDeclarations(..) 时，它运行良好。

所以我的问题：

1. 如何检测 AMQP Source 和 Sink 是否正常运行？
2. 如何忽略声明异常？
3. 如果出现异常，如何知道并导致系统出现故障？

Answer 1

要回答 1 和 3，AmqpSink 实现了一个 CompletionStage<Done>，您必须保留并处理（注册一些回调函数）以观察流的失败和完成。在文档示例中，我们阻止了在生产代码中不好的完成阶段 (https://doc.akka.io/docs/alpakka/current/amqp.html#with-sink)，这可能是因为该示例包含在 Alpakka 测试之一中。更喜欢通常的CompletionStage 回调/转换方法（例如参见this introduction）。

CompletionStage 将在发生错误时失败，当流被物化/启动时或在元素处理期间，或者在源到达末尾并且每个元素都通过您的流进入接收器时完成。这意味着要启动流，如果它没有很快失败，它就会运行。

对于问题 2，不确定是否可以忽略声明异常，可能是那些总是连接失败。