我试图从postgresql和r2dbc接收异步通知，我使用reactivestream和postgres驱动程序，但我不能声明我是java和reactivestream的新手，我不使用像spring这样的框架。
谢谢你们

package com.poc.r2dbc_listen;
import io.r2dbc.postgresql.*;
import io.r2dbc.spi.*;
import org.reactivestreams.*;
import reactor.core.publisher.Flux;

public class R2dbc_listen {

    public static void main(String[] args) {
         PostgresqlConnectionFactory connectionFactory = new PostgresqlConnectionFactory(PostgresqlConnectionConfiguration.builder()
    .host("127.0.0.1")
    .port(5442)  
    .username("postgres")
    .password("psswd")
    .database("database") 
    .build());
    Publisher<? extends Connection> connectionPublisher = connectionFactory.create();
    Flux<Notification> listen = connectionPublisher.createStatement("LISTEN mymessage")
                                .execute()
                                .flatMap(PostgresqlResult::getRowsUpdated)
        .thenMany(receiver.getNotifications());
    }
}