Continuous Query With Spring Data Reactive MongoDB

Continuous Query With Spring Data Reactive MongoDB

On this blog post, we will take a look how to implement Continuous Queries in MongoDB. Also, we will have a dash of Spring WebFlux in Action.

Let’s do it, right now!!!

What is Continuous Query???

Is a kind of active query, when the data were arriving in the database, if this piece of data matches accordingly with our query an event will be emitted to our application, with the piece of data. We can think a kind of Event-Driven Programming using the database as an event trigger.

It is a powerful feature and enable us to add interactive behaviors in our application.

Capped Collections and Tailable Cursors

In MongoDB, there is a feature called Capped Collections. This kind of collection has a fixed size and support high-throughput operations and retrieve the documents based on insert order. We will use this kind of collection to store our data to simulate the “continuous query.”

Also, MongoDB has an exciting feature called Tailable Cursors which we are able to use on Capped Collections. This cursor is similar to

1

tail-f

in Unix command system, which means we will retrieve the documents in natural order.

Spring Data Reactive MongoDB

Now we know how Capped Collections and Tailable Cursor works. Then is time to use Spring Data Reactive Mongo DB stuff.

Use Case

We will simulate a kind of data collector for IoT components. The main feature of our project is collect components data and provides an API to show max and min temperatures when the temperature reaches the limits configured for a specific device. Then the Frontend Team can build a fantastic reactive page to display the device information.

Awesome, there is some interesting stuff here. The first one is the @Tailable annotation, which instructs the Spring Frameworks to await the MongoDB Tailable Cursors. When the data arrives in MongoDb with de data matches with a specific query one event will be emitted. In this case, we will watch events from a specific device. Take a look at our return type we always should use the Flux<T> to make application reactive.

Time to create our service class, the implementation should be like this:

On our method init(), we have declared a capped collection. First, we have removed the collection created and then reactively we have created and configured the temperatures collection. Remember this method will call one time. The others methods are so simple the main idea here is to call the Repository Layer to retrieve the documents. The important part here is the return types of methods, always Flux<T> when we expect the events.

In our RestController we will use the Server-Sent Event (SSE), to add more interactive behaviors for our API. The main idea here is to push data to the client using the Half-Duplex HTTP connection.

As we can see take a look at the methods declarations we have used the MediaType.APPLICATION_STREAM_JSON_VALUE. It is much important to add Stream Behavior for our application.

Simple like that we have created a couple of classes and lines, to add a powerful feature like “continuous query” in our application. Spring Data Reactive reactive brings reactive characteristics for our application and make them more resilient and cost-effective.

The Full source code can be found at GitHub. There is a client to add some data and make the test easier, the client will send data to interval pre-configured.

The Spring 5.0 By Example Book

Recently I launched the Spring 5.0 By Example Book. This book contains a lot of Spring Concepts and Examples, using the Microservices Architecture and much codes using the Spring Boot 2.