UNI等待VERTX EventBus消息

Question

我有两个端点：

    @GET
    @Produces(MediaType.TEXT_PLAIN)
    @Path("/waitForEvent")
    public Uni<Object> waitForEvent() {
        return Uni.createFrom().emitter(em -> {
            //wait for event from eventBus
    //            eventBus.consumer("test", msg -> {
    //                System.out.printf("receive event: %s\n", msg.body());
    //                em.complete(msg);
    //            });
        }).ifNoItem().after(Duration.ofSeconds(5)).failWith(new RuntimeException("timeout"));
    }
    
    @GET
    @Path("/send")
    public void test() {
        System.out.println("send event");
        eventBus.send("test", "send test event");
    }

wareforevent（）仅在从EventBus接收事件的情况下才能完成。如何使用Vertx和Mutiny实现这一目标？

Answer 1

通常，我们避免使用这种模式，并使用事件总线中的请求/回复机制：

@GET
@Path("/send")
public Uni<String> test() {
   return bus.<String>request("test", name)        
        .onItem().transform(Message::body)
        .ifNoItem().after(Duration.ofSeconds(5)).failWith(new RuntimeException("timeout"));
}

使用两个端点实现（如问题）时，它可能会变得更加复杂，好像您对/wateforevent 端点，您需要确保每个“消费者”收到消息。

仍然有可能，但是将需要这样的东西：

@GET
@Produces(MediaType.TEXT_PLAIN)
@Path("/waitForEvent")
public Uni<String> waitForEvent() {
  return Uni.createFrom().emitter(emitter -> {
    MessageConsumer<String> consumer = bus.consumer("test");
      consumer.handler(m -> {
      emitter.complete(m.body());
      consumer.unregisterAndForget();
   })
        .ifNoItem().after(Duration.ofSeconds(5)).failWith(new RuntimeException("timeout"));

}
    
@GET
@Path("/send")
public void test() {
  bus.publish("test", "send test event");
}

请确保使用io.vertx.mutiny.core.eventbus.eventbus.eventbus事件总线的变体。