Quarkus AMQP在请求后向队列发送消息业务逻辑

Question

一旦我收到HTTP Get/Post，我必须持久化并反对，然后将消息发送到其他服务正在监听的队列以开始执行其他复杂工作

我目前的问题是，我不能只调用带有@Outgoing（“channel”）注释的方法，我试过了，只是继续执行该方法而不调用它

有没有办法使用Quarkus框架调用方法将JSON有效负载发送到队列？

PS：我也尝试使用rabbitMQ并切换回ActiveMQ

我遵循Quarkus tuturial的反应式消息传递，并试图在已实现的资源中注册一些内容，但没有成功

@Path("/part")
class PartService : PanacheRepository<PartDao>, Logging {

    @GET
    @Produces(MediaType.APPLICATION_JSON)
    @Transactional
    fun fetchParts(): List<PartDao> {

        val partDao = PartDao(label = "Test", status = PartStatus.INBANK, creatorId = "ghost-007")

        partDao.persist()

        if (partDao.isPersistent) {
            // Send a message to a queue -> PoC
            send(partDao)
        }

        return findAll().list()
    }

    @Outgoing("part-persisted")
    @Transactional
    fun send(partDao: PartDao): CompletionStage<AmqpMessage<*>> {
        val future = CompletableFuture<AmqpMessage<*>>()
        val message = "hello from sender"

        // Debug proposes
        println("Sending (data): $message")
        logger.debug(partDao.toString())

        future.complete(AmqpMessage(message))
        return future
    }

}

预期：

执行以下操作后，在队列中注册消息“hello from sender”：

curl http://localhost/part

实际结果：

发送方法只是继续执行

Answer 1

如果我理解正确，您想调用一个将某些内容放入流中的方法。

据我所知，您必须使用Emitter来执行此操作，参见例如https://github.com/michalszynkiewicz/devoxxpl-demo/blob/master/search/src/main/java/com/example/search/SearchEndpoint.java#L23

看见https://smallrye.io/smallrye-reactive-messaging/#_stream文档