如何在使用反应消息传递时传播JTA状态？

Question

@Inject
@Channel("test")
Emitter<String> emitter;

@POST
@Transactional
public Response test() {
    emitter.send("test");
}

@ApplicationScoped
@Connector("test")
public class TestConnector implements OutgoingConnectorFactory {

    @Inject
    TransactionManager tm;

    @Override
    public SubscriberBuilder<? extends Message<?>, Void> getSubscriberBuilder(Config config) {
        return ReactiveStreams.<Message<?>>builder()
            .flatMapCompletionStage(message -> {
                tm.getTransaction(); // = null
                return message.ack();
            })
            .ignore();
    }
}

顺便说一下，在使用@incoming/@outgoing而不是发射器时，我遇到了同样的问题。我已经决定给你们举这个例子，因为它很容易理解和再现。

Answer 1

此刻，您需要在消息元数据中传递当前事务。因此，它将传播到不同的下游组件（以及连接器）。

注意，Transaction倾向于附加到请求范围，这意味着在连接器中，使用它可能已经太晚了。因此，确保您的endpoint是异步的，并且仅在发出的消息被确认时才返回。

上下文传播在这种情况下没有帮助，因为底层流是在启动时构建的（在Quarkus中是在构建时），所以没有捕获上下文。