时间戳

首页 > 解决方案 > Quarkus AMQP 在请求业务逻辑后将消息发送到队列

kotlin - Quarkus AMQP 在请求业务逻辑后将消息发送到队列

问题描述

一旦我收到 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
    }

}

预期的：

完成后在队列中注册消息“来自发件人的你好”：

curl http://localhost/part

实际结果：

发送方法只是继续执行

标签： kotlinquarkus

解决方案

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

据我所知，您必须使用 anEmitter来执行此操作，请参见例如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文档。

推荐阅读