SpringData MongoDB DefaultMessageListenerContainer Sınıfı

constructor
Şöyle yaparız.

@Configuration
class Config {

  @Bean
  MessageListenerContainer messageListenerContainer(MongoTemplate template) {
    return new DefaultMessageListenerContainer(template);
  }
}

register metodu
Şöyle yaparız.

MessageListener<ChangeStreamDocument<Document>, Person> messageListener = (message) -> {
    System.out.println("Hello " + message.getBody().getFirstname());
};

ChangeStreamRequest<Person> request = ChangeStreamRequest.builder()
    .collection("person")
    .filter(newAggregation(match(where("operationType").is("insert"))))
    .publishTo(messageListener)
    .build();

Subscription subscription = messageListenerContainer.register(request, Person.class);

// ...

Aynı şeyi Flux ile şöyle yaparız.

Flux changeStream = reactiveTemplate
  .changeStream(newAggregation(match(where("operationType").is("insert"))),
    Person.class, ChangeStreamOptions.empty(), "person");

changeStream.doOnNext(event -> System.out.println("Hello " +
                               event.getBody().getFirstname()))
  .subscribe();