@Tailable(spring-data-reactive-mongodb) ekvivalent v spring-data-r2dbc

Byl jsem na stejném problému a nebyl jsem si jistý, zda jste našli řešení nebo ne, ale podařilo se mi dosáhnout něčeho podobného provedením následujícího. Nejprve jsem přidal trigger do své tabulky

CREATE TRIGGER trigger_name
    AFTER INSERT OR DELETE OR UPDATE 
    ON table_name
    FOR EACH ROW
    EXECUTE PROCEDURE trigger_function_name;

To nastaví spouštěč v tabulce vždy, když je řádek aktualizován, odstraněn nebo vložen. Potom zavolá spouštěcí funkci, kterou jsem nastavil a která vypadala asi takto:

CREATE FUNCTION trigger_function_name
RETURNS trigger
LANGUAGE 'plpgsql'
COST 100
VOLATILE NOT LEAKPROOF
AS 
$BODY$
DECLARE
    payload JSON;
BEGIN
    payload = row_to_json(NEW);
    PERFORM pg_notify('notification_name', payload::text);
    RETURN NULL;
END;
$BODY$;

To mi umožní 'naslouchat' jakékoli z těchto aktualizací z mého jarního bootovacího projektu a odešle celý řádek jako užitečné zatížení. Dále jsem v mém jarním bootovacím projektu nakonfiguroval připojení k mé databázi.

@Configuration
@EnableR2dbcRepositories("com.(point to wherever repository is)")
public class R2DBCConfig extends AbstractR2dbcConfiguration {
    @Override
    @Bean
    public ConnectionFactory connectionFactory() {
        return new PostgresqlConnectionFactory(PostgresqlConnectionConfiguration.builder()
                .host("host")
                .database("db")
                .port(port)
                .username("username")
                .password("password")
                .schema("schema")
                .connectTimeout(Duration.ofMinutes(2))
                .build());
    }
}

S tím jsem to Autowire (vstřikování závislosti) do konstruktoru v mé třídě služeb a přeneslo to do třídy r2dbc PostgressqlConnection takto:

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

Nyní chceme 'poslouchat' náš stůl a dostat upozornění, když provedeme nějakou aktualizaci našeho stolu. Za tímto účelem jsme nastavili metodu inicializace, která se provádí po vložení závislosti pomocí anotace @PostContruct

@PostConstruct
private void postConstruct() {
    postgresqlConnection.createStatement("LISTEN notification_name").execute()
            .flatMap(PostgresqlResult::getRowsUpdated).subscribe();
}

Všimněte si, že posloucháme jakékoli jméno, které vložíme do metody pg_notify. Také chceme nastavit metodu, jak ukončit spojení, když má být fazole odhozena, například takto:

@PreDestroy
private void preDestroy() {
    postgresqlConnection.close().subscribe();
}

Nyní jednoduše vytvořím metodu, která vrátí Flux čehokoli, co je aktuálně v mé tabulce, a také ji sloučím s mými oznámeními, jak jsem řekl předtím, než přijdou oznámení jako json, takže jsem to musel deserializovat a rozhodl jsem se použít ObjectMapper. Takže to bude vypadat nějak takto:

private Flux<YourClass> getUpdatedRows() {
    return postgresqlConnection.getNotifications().map(notification -> {
        try {
            //deserialize json
            return objectMapper.readValue(notification.getParameter(), YourClass.class);
        } catch (IOException e) {
            //handle exception
        }
    });
}

public Flux<YourClass> getDocuments() {
    return documentRepository.findAll().share().concatWith(getUpdatedRows());
}

Doufám, že to pomůže. Na zdraví!