Quarkus SSE Redis订阅

Question

我喜欢做一个有redis回应的SSE。在quarkus中订阅。

我有一个来自quarkus快速入门的简单SSE示例

 @GET
  @Produces(MediaType.SERVER_SENT_EVENTS)
  @SseElementType(MediaType.TEXT_PLAIN)
  @Path("{name}/streaming")
  public Multi<String> greeting(@org.jboss.resteasy.annotations.jaxrs.PathParam String name) {
    return Multi.createFrom().publisher(vertx.periodicStream(2000).toMulti())
        .map(l -> String.format("Hello %s! (%s)%n", name, new Date()));
  }

这个效果很好，每2秒钟我就会收到Hello。。。。在我的web浏览器中

现在我尝试订阅Redis，所以我应该会收到Redis的消息。

Redis示例：

(cmd window 1)
SUBSCRIBE message-channel
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "message-channel"
3) (integer) 1

(cmd window 2)
PUBLISH  message-channel HelloWorld
(integer) 1

(cmd window 1)
1) "message"
2) "message-channel"
3) "HelloWorld"

现在，我用quarkus SSE尝试以下方法：

  @Inject
  ReactiveRedisClient reactiveRedisClient;

 @GET
  @Produces(MediaType.SERVER_SENT_EVENTS)
  @SseElementType(MediaType.TEXT_PLAIN)
  @Path("sse/redissse")
  public Multi<String> redissse() {
    List<String> subscriberList = new ArrayList();
    subscriberList.add("message-channel");

    return reactiveRedisClient.subscribe(subscriberList)
        .onItem().transformToMulti(keys -> Multi.createFrom().iterable(keys))
        .onItem().castTo(String.class);
  }

我收到的是一个例外：

WARNING [io.ver.red.cli.imp.RedisConnectionImpl] (vert.x-eventloop-thread-0) No handler waiting for message: [subscribe, message-channel, 1]

有人能支持我吗？有一个简单的例子吗？我对此一无所知，我无法接收带有“订阅”发布的Redis消息。

任何建议。。。

Answer 1

现在，我执行以下操作：

  @Inject
  @RedisClientName("second")
  RedisClient redisClient2;

void onStart(@Observes StartupEvent ev) throws IOException {
  this.redisClient2.subscribe(List.of("message-channel"));
}


  @GET
  @Produces(MediaType.SERVER_SENT_EVENTS)
  @SseElementType(MediaType.TEXT_PLAIN)
  @Path("/redis/subscribe")
  public Publisher<String> subscribechannel(){
     return eventBus.<String>consumer("io.vertx.redis.message-channel").toPublisherBuilder()
        .map(Message::body)
        .buildRs();
  }

现在它可以工作了，但是如果我从多个浏览器进行SSE，它们会共享事件。所以每个消费者（浏览器）之后只有一个收到事件。

Answer 2

我没有使用Redis pub sub，但我确实使用了Redis streams，我需要做的是：

`

return Multi.createBy().repeating()
    .supplier(() -> this.reactiveRedisClient.subscribe(subscriberList)
                        .onItem().transformToMulti(keys -> Multi.createFrom().iterable(keys))
                        .onItem().castTo(String.class))
        .indefinitely()
        .onItem().disjoint();

`

我想既然pub-sub是非阻塞的，它只运行一次，就不会等到另一条消息到达。您必须以反应式的方式实现自己的while（true）循环。