Quarkus-基于Kafka写入确认和nack的状态响应

Question

我有一个将数据推送到kafka的endpoint。现在，我想分别在kafka写入成功或失败的情况下使用近似的状态代码2xx或5xx响应调用。代码片段是

@Path("/prices")
public class PriceResource {

    @Inject @Channel("price-create") Emitter<Double> priceEmitter;

    @POST
    @Consumes(MediaType.TEXT_PLAIN)
    public void addPrice(Double price) {
        priceEmitter.send(Message.of(price)
            .withAck(() -> {
                // Called when the message is acked
                return CompletableFuture.completedFuture(null);
            })
            .withNack(throwable -> {
                // Called when the message is nacked
                return CompletableFuture.completedFuture(null);
            }));
         // return appropriate response
          return Response
    }
}

现在的问题是，在执行ack或nack回调之前，endpoint正在使用状态代码进行响应。还尝试了MutinyEmitter的sendAndAwait方法，但该方法返回void。因此，无法知道该消息是已确认还是未确认。

Answer 1

这里最好的方法是链接异步操作，如下所示：

@POST
@Consumes(MediaType.TEXT_PLAIN)
public Uni<Response> addPrice(Double price) {
    return Uni.createFrom().completionStage(priceEmitter.send(price))
            .onItem().transform(ignored -> Response.ok().entity("foo").build())
            .onFailure().recoverWithItem(Response.serverError().build());
}

如果您想使用同步代码（我不推荐）：

@Blocking
@POST
@Consumes(MediaType.TEXT_PLAIN)
public Response addPrice(Double price) {
    try {
        Uni.createFrom().completionStage(priceEmitter.send(price))
                .await().indefinitely();

        return Response.ok().entity("foo").build();
    } catch (Exception e) {
        return Response.serverError().build();
    }
}

这个等待（）。如果Uni发出故障，infinally（）将抛出异常。

您还可以选择直接使用发射器返回的CompletionStage，而无需将其转换为Uni，但请记住Quarkus选择Mutiny作为其默认反应框架。