Spring WebFlux with ReactiveMongoRepository: not getting database updates through the stream

The issue I am facing:

Whatever I try through various tutorials on using Spring Reactive (WebFlux) REST API’s, I am unable to get it to work. When I initially call my endpoint, I am able to get the results from the MongoDB collection. However, whenever I make an update to a document entry, or add a new document, it is not updated in through the text-event-stream. Each time I have to call the endpoint again to get new results.

The setup:

Currently I have the following setup:

• Spring Cloud Gateway (behind which I run various )
• Spring Boot Service containing RestControllers (calling this the main service)

I am using Spring Webflux, Spring Cloud Gateway and Spring ReactiveMongoRepository.

The dependencies included pom.xml for the Spring Boot main service:

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-security</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.security</groupId>
            <artifactId>spring-security-oauth2-jose</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
        </dependency>

The code for the ReactiveMongoRepository:

@Repository
public interface TestRepository extends ReactiveMongoRepository<TestIntegration, String> {

    @Query(("{'userId': ?0}"))
    Flux<TestIntegration> findbyUserId(String userId);
}

The code for the rest Controller:

@GetMapping(value = "main/integrations", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<TestIntegration> retrieveIntegrations(ServerWebExchange exchange) {
        return testRepository.findAll();
}

According to every tutorial/guideline I have read, this should be working according to this approach. Has anyone experienced this as well or might be able to assist in the matter? Currently stuck on this for days…

Answer

This works as intended. When you stream items from a Flux<T> it will feed as long as there are items in the stream. Then the stream will close. So in your case, it fetched all data from the database, streamed it to you and then closed.

If you wish to keep the stream open you need to keep sending data. One way to do that is to send :keep alive messages (messages that start with a comma) using ServerSentsEvents. You can read more about ServerSentEvents and the colon operator and these types of messages in the official mozilla documentation.

When you are able to actually keep the stream open, and wish to send data, your service will not know when new data is written to the database. So either you poll your database, or trigger an event when something is written, to fetch the newly written data and place this in the stream.

How to place data in a continuous open stream is way too big of a topic to explain here. But i suggest you read the following sections in the official reactor documentation:

Programmatically creating a sequence

Processors and Sinks