Spring Webflux 异步 PostgreSQL 发布器在第一个结果后停止

Question

我正在尝试用响应式异步替换 PostgreSQL 数据库轮询器 postgres-async-driver and stream newly inserted rows to a Spring 5 Webflux Reactive websocket client like Josh Long's awesome example demoed here and based on Sébastien Deleuze's spring-reactive-playground。

我的 Publisher 获得第一行 row，但随后没有 return 后续行。我的 Observable、Publisher 或我使用 postgres-async-driver Db 的方式有问题吗？

public Observable<WebSocketMessage> getObservableWSM(WebSocketSession session){
    return
        // com.github.pgasync.Db
        db.queryRows(sql)
        // ~RowMapper method
        .map(row -> mapRowToDto(row))
        // serialize dto to String for websocket
        .map(dto -> { return objectMapper.writeValueAsString(dto); })
        // finally, write to websocket session 
        .map(str -> { return session.textMessage((String) str);
        });
}

然后，我使用 RxReactiveStream.toPublisher 转换器将 Observable 连接到我的 WebSocketHandler：

@Bean
WebSocketHandler dbWebSocketHandler() {
    return session -> {
        Observable<WebSocketMessage> o = getObservableWSM(session);
        return session.send(Flux.from(RxReactiveStreams.toPublisher(o)));
    };
}

这将从我的 sql 语句中获取第一个 row，但没有其他行。如何继续流式传输其他行？

理想情况下，我想我想要 MongoDB .

的 PostgreSQL 等价物

Answer 1

我创建了一个 Postgres Trigger，它会在 INSERT 时触发我的 table 基于 this example:

CREATE OR REPLACE FUNCTION table_update_notify() RETURNS trigger AS $$
DECLARE
  id bigint;
BEGIN
  IF TG_OP = 'INSERT' THEN
    id = NEW.id;
  ELSE
    id = OLD.id;
  END IF;
  PERFORM pg_notify('my_trigger_name', json_build_object('table', TG_TABLE_NAME, 'id', id, 'type', TG_OP)::text);
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

然后我使用 reactive-pg-client 订阅了那个 Postgres 触发器。这是他们的 Pub/Sub 示例中的代码：

@Bean
PgPool subscribedNotificationHandler() {
    PgPool client = pgPool();
    client.getConnection(asyncResult -> {
        if (asyncResult.succeeded()) {
            PgConnection connection = asyncResult.result();
            connection.notificationHandler(notification -> {
                notification.getPayload();
                // do things with payload
            });
            connection.query("LISTEN my_trigger_name", ar -> {
                log.info("Subscribed to channel");
            });
        }
    });
    return client;
}

Spring Webflux 异步 PostgreSQL 发布器在第一个结果后停止

Spring Webflux Async PostgreSQL Publisher Stops After First Result

postgresql

spring

websocket

rx-java

spring-webflux