@Tailable(spring-data-reactive-mongodb) svarende til spring-data-r2dbc

Jeg var på det samme problem, ikke sikker på, om du fandt en løsning eller ej, men jeg var i stand til at opnå noget lignende ved at gøre følgende. Først tilføjede jeg trigger til min tabel

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

Dette vil sætte en trigger på bordet, hver gang en række opdateres, slettes eller indsættes. Så kalder den triggerfunktionen jeg har sat op, som så nogenlunde sådan her ud:

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$;

Dette vil give mig mulighed for at 'lytte' til enhver af disse opdateringer fra mit spring boot-projekt, og det vil sende hele rækken som en nyttelast. Dernæst konfigurerede jeg i mit spring boot-projekt en forbindelse til min db.

@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());
    }
}

Med det autowirer jeg (afhængighedsinjektion) den ind i konstruktøren i min serviceklasse og caster den til en r2dbc PostgressqlConnection-klasse som sådan:

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

Nu vil vi 'lytte' til vores bord og få besked, når vi udfører en opdatering til vores bord. For at gøre det opsætter vi en initialiseringsmetode, der udføres efter afhængighedsinjektion ved at bruge @PostContruct-annotationen

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

Bemærk, at vi lytter til det navn, vi lægger i pg_notify-metoden. Vi ønsker også at opsætte en metode til at lukke forbindelsen, når bønnen er ved at blive smidt væk, sådan:

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

Nu opretter jeg simpelthen en metode, der returnerer en Flux af hvad der er i min tabel i øjeblikket, og jeg flette den også med mine notifikationer, som jeg sagde før notifikationerne kommer ind som en json, så jeg var nødt til at deserialisere den, og jeg besluttede at bruge ObjectMapper. Så det vil se nogenlunde sådan ud:

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());
}

Håber dette hjælper. Skål!