r2dbc postgres driver - how to extract PostgresqlConnection from PooledConnection

Question

Hi, my Spring Boot application has r2dbc connection pool autoconfigured using a property file:

    spring.r2dbc.url=r2dbc:pool:postgres://localhost:5432/ecom
    spring.r2dbc.username=xxx
    spring.r2dbc.password=yyy</code></pre>

Now I need to get a PostgresqlConnection instance and I do that this way:

this.connection = Mono.from(connectionFactory.create()).cast(PostgresqlConnection.class).block();

But because this is a pool configuration I am receiving a ClassCastException and the following PooledConnection object which wraps required PostgresqlConnection:

PooledConnection[PostgresqlConnection{client=io.r2dbc.postgresql.client.ReactorNettyClient@14c93774, codecs=io.r2dbc.postgresql.codec.DefaultCodecs@62a68bcb}]

I'd need to get to PostgresqlConnection and use its native capabilities like notifications:

PostgresqlConnection connection = …;
    Flux<Notification> listen = connection.createStatement("LISTEN mymessage")
    .execute()
    .flatMap(PostgresqlResult::getRowsUpdated)
    .thenMany(connection.getNotifications());

The question is how to properly get PostgresqlConnection instance from connectionFactory? Any help will be appreciated.

Answer 1

1. override the default ConnectionFactory.

    @Bean
    @Primary
    public ConnectionFactory connectionFactory() {
        return new PostgresqlConnectionFactory(
                PostgresqlConnectionConfiguration.builder()
                        .host("localhost")
                        .database("test")
                        .username("user")
                        .password("password")
                        .codecRegistrar(EnumCodec.builder().withEnum("post_status", Post.Status.class).build())
                        .build()
        );
    }
   

2. Create another connectionfactory for listen/notify.

       @Bean
    @Qualifier("pgConnectionFactory")
    public ConnectionFactory pgConnectionFactory() {
        return new PostgresqlConnectionFactory(
                PostgresqlConnectionConfiguration.builder()
                        .host("localhost")
                        .database("test")
                        .username("user")
                        .password("password")
                        //.codecRegistrar(EnumCodec.builder().withEnum("post_status", Post.Status.class).build())
                        .build()
        );
    }
   

I have created an example for the second method, check here.

Start the application, send hello from curl:

curl http://localhost:8080/hello

In the console, you will see there are some messages like the following:

2020-09-15 16:49:20.657  INFO 20216 --- [ctor-http-nio-4] sending notification::                   : onSubscribe(FluxFlatMap.FlatMapMain)
2020-09-15 16:49:20.658  INFO 20216 --- [ctor-http-nio-4] sending notification::                   : request(unbounded)
2020-09-15 16:49:20.666  INFO 20216 --- [actor-tcp-nio-2] reactor.Flux.ConcatMap.2                 : onNext(NotificationResponseWrapper{name=mymessageprocessId=753parameter=Hello world at 2020-09-15T16:49:20.656715600})
2020-09-15 16:49:20.667  INFO 20216 --- [actor-tcp-nio-2] com.example.demo.Listener                : notifications: NotificationResponseWrapper{name=mymessageprocessId=753parameter=Hello world at 2020-09-15T16:49:20.656715600}