spring-projectsGH-2788: Add MongoDbChangeStreamMessageProducer

Fixes spring-projects#2788

* Introduce a `MessageProducerSupport.subscribeToPublisher(Publisher<Message<?>>)`
for components which produces `Flux` for data from their source
* Such a component is auto-stopped when subscription to that `Publisher` is canceled
* Implement a `MongoDbChangeStreamMessageProducer` based on the reactive support for
in Spring Data MongoDb
* Implement a Java DSL for `MongoDbChangeStreamMessageProducer`
* Disable a test for change stream since it requires server of version 4.x started with 'replSet' option
* Add `MongoHeaders` for change stream events

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
 package org.springframework.integration.endpoint;
 
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscription;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.SmartInitializingSingleton;
@@ -39,8 +40,7 @@
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
-import reactor.core.Disposable;
-import reactor.core.Disposables;
+import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.Flux;
 
 /**
@@ -54,8 +54,6 @@
 public abstract class MessageProducerSupport extends AbstractEndpoint implements MessageProducer, TrackableComponent,
 		SmartInitializingSingleton, IntegrationPattern {
 
-	private final Disposable.Composite reactiveSubscriptions = Disposables.composite();
-
 	private final MessagingTemplate messagingTemplate = new MessagingTemplate();
 
 	private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy();
@@ -70,6 +68,8 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
 
 	private boolean shouldTrack = false;
 
+	private volatile Subscription subscription;
+
 	protected MessageProducerSupport() {
 		this.setPhase(Integer.MAX_VALUE / 2);
 	}
@@ -193,11 +193,17 @@ protected void doStart() {
 	}
 
 	/**
-	 * Takes no action by default. Subclasses may override this if they
+	 * Cancels {@link #subscription} if any.
+	 * Subclasses may override this if they
 	 * need lifecycle-managed behavior.
 	 */
 	@Override
 	protected void doStop() {
+		Subscription subs = this.subscription;
+		if (subs != null) {
+			subs.cancel();
+			this.subscription = null;
+		}
 	}
 
 	protected void sendMessage(Message<?> messageArg) {
@@ -219,24 +225,28 @@ protected void sendMessage(Message<?> messageArg) {
 
 	protected void subscribeToPublisher(Publisher<? extends Message<?>> publisher) {
 		MessageChannel outputChannel = getRequiredOutputChannel();
-		Flux<Message<?>> messageFlux =
+
+		Flux<? extends Message<?>> messageFlux =
 				Flux.from(publisher)
-						.map(this::trackMessageIfAny);
+						.map(this::trackMessageIfAny)
+						.doOnSubscribe(subscription -> this.subscription = subscription)
+						.doOnComplete(this::stop)
+						.doOnCancel(this::stop);
+
 		if (outputChannel instanceof ReactiveStreamsSubscribableChannel) {
 			((ReactiveStreamsSubscribableChannel) outputChannel).subscribeTo(messageFlux);
 		}
 		else {
-			this.reactiveSubscriptions.add(
-					messageFlux
-							.doOnNext((message) -> {
-								try {
-									sendMessage(message);
-								}
-								catch (Exception ex) {
-									logger.error("Error sending a message: " + message, ex);
-								}
-							})
-							.subscribe());
+			messageFlux
+					.doOnNext((message) -> {
+						try {
+							sendMessage(message);
+						}
+						catch (Exception ex) {
+							logger.error("Error sending a message: " + message, ex);
+						}
+					})
+					.subscribe();
 		}
 	}
 
@@ -297,7 +307,11 @@ private Message<?> trackMessageIfAny(Message<?> message) {
 
 	@Override
 	public void destroy() {
-		this.reactiveSubscriptions.dispose();
+		Subscription subs = this.subscription;
+		if (subs != null) {
+			subs.cancel();
+			this.subscription = null;
+		}
 		super.destroy();
 	}
 

spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveMessageProducerTests.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.endpoint;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -43,15 +45,22 @@ public class ReactiveMessageProducerTests {
 	@Autowired
 	public FluxMessageChannel fluxMessageChannel;
 
+	@Autowired
+	public MessageProducerSupport producer;
+
 	@Test
 	public void test() {
+		assertThat(this.producer.isRunning()).isTrue();
+		
 		StepVerifier.create(
 				Flux.from(this.fluxMessageChannel)
 						.map(Message::getPayload)
 						.cast(String.class))
 				.expectNext("test1", "test2")
 				.thenCancel()
 				.verify();
+
+		assertThat(this.producer.isRunning()).isFalse();
 	}
 
 	@Configuration

spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDb.java

@@ -24,6 +24,7 @@
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.expression.ValueExpression;
+import org.springframework.integration.mongodb.inbound.MongoDbChangeStreamMessageProducer;
 
 /**
  * Factory class for building MongoDb components
@@ -140,6 +141,18 @@ public static ReactiveMongoDbMessageSourceSpec reactiveInboundChannelAdapter(Rea
 		return new ReactiveMongoDbMessageSourceSpec(mongoTemplate, new ValueExpression<>(query));
 	}
 
+	/**
+	 * Create a {@link MongoDbChangeStreamMessageProducerSpec} builder instance
+	 * based on the provided {@link ReactiveMongoOperations}.
+	 * @param mongoOperations the {@link ReactiveMongoOperations} to use.
+	 * @return the {@link MongoDbChangeStreamMessageProducerSpec} instance
+	 * @since 5.3
+	 */
+	public static MongoDbChangeStreamMessageProducerSpec changeStreamInboundChannelAdapter(
+			ReactiveMongoOperations mongoOperations) {
+
+		return new MongoDbChangeStreamMessageProducerSpec(new MongoDbChangeStreamMessageProducer(mongoOperations));
+	}
 
 	private MongoDb() {
 	}

spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDbChangeStreamMessageProducerSpec.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.mongodb.dsl;
+
+import org.springframework.data.mongodb.core.ChangeStreamOptions;
+import org.springframework.integration.dsl.MessageProducerSpec;
+import org.springframework.integration.mongodb.inbound.MongoDbChangeStreamMessageProducer;
+
+/**
+ * A {@link MessageProducerSpec} for tne {@link MongoDbChangeStreamMessageProducer}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.3
+ */
+public class MongoDbChangeStreamMessageProducerSpec
+		extends MessageProducerSpec<MongoDbChangeStreamMessageProducerSpec, MongoDbChangeStreamMessageProducer> {
+
+	/**
+	 * Construct a builder based on an initial {@link MongoDbChangeStreamMessageProducerSpec}.
+	 * @param producer the {@link MongoDbChangeStreamMessageProducerSpec} to use.
+	 */
+	public MongoDbChangeStreamMessageProducerSpec(MongoDbChangeStreamMessageProducer producer) {
+		super(producer);
+	}
+
+	/**
+	 * Configure a domain type to convert change event body into.
+	 * @param domainType the type to use.
+	 * @return the spec.
+	 */
+	public MongoDbChangeStreamMessageProducerSpec domainType(Class<?> domainType) {
+		this.target.setDomainType(domainType);
+		return this;
+	}
+
+	/**
+	 * Configure a collection to subscribe for change events.
+	 * @param collection the collection to use.
+	 * @return the spec.
+	 */
+	public MongoDbChangeStreamMessageProducerSpec collection(String collection) {
+		this.target.setCollection(collection);
+		return this;
+	}
+
+	/**
+	 * Configure a {@link ChangeStreamOptions}.
+	 * @param options the {@link ChangeStreamOptions} to use.
+	 * @return the spec.
+	 */
+	public MongoDbChangeStreamMessageProducerSpec options(ChangeStreamOptions options) {
+		this.target.setOptions(options);
+		return this;
+	}
+
+	/**
+	 * Configure a flag to extract body from a change event or use event as a payload.
+	 * @param extractBody to extract body or not.
+	 * @return the spec.
+	 */
+	public MongoDbChangeStreamMessageProducerSpec extractBody(boolean extractBody) {
+		this.target.setExtractBody(extractBody);
+		return this;
+	}
+
+}

spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbChangeStreamMessageProducer.java

@@ -0,0 +1,131 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.mongodb.inbound;
+
+import org.bson.Document;
+import org.reactivestreams.Publisher;
+
+import org.springframework.data.mongodb.core.ChangeStreamEvent;
+import org.springframework.data.mongodb.core.ChangeStreamOptions;
+import org.springframework.data.mongodb.core.ReactiveMongoOperations;
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.mongodb.support.MongoHeaders;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * A {@link MessageProducerSupport} for MongoDB Change Stream implementation.
+ * The functionality is based on the
+ * {@link ReactiveMongoOperations#changeStream(String, ChangeStreamOptions, Class)}
+ * and {@link MessageProducerSupport#subscribeToPublisher(Publisher)} consumption.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.3
+ */
+public class MongoDbChangeStreamMessageProducer extends MessageProducerSupport {
+
+	private final ReactiveMongoOperations mongoOperations;
+
+	private Class<?> domainType = Document.class;
+
+	@Nullable
+	private String collection;
+
+	private ChangeStreamOptions options = ChangeStreamOptions.empty();
+
+	private boolean extractBody = true;
+
+	/**
+	 * Create an instance based on the provided {@link ReactiveMongoOperations}.
+	 * @param mongoOperations the {@link ReactiveMongoOperations} to use.
+	 * @see ReactiveMongoOperations#changeStream(String, ChangeStreamOptions, Class)
+	 */
+	public MongoDbChangeStreamMessageProducer(ReactiveMongoOperations mongoOperations) {
+		Assert.notNull(mongoOperations, "'mongoOperations' must not be null");
+		this.mongoOperations = mongoOperations;
+	}
+
+	/**
+	 * Specify an object type to convert an event body to.
+	 * Defaults to {@link Document} class.
+	 * @param domainType the class for event body conversion.
+	 * @see ReactiveMongoOperations#changeStream(String, ChangeStreamOptions, Class)
+	 */
+	public void setDomainType(Class<?> domainType) {
+		Assert.notNull(domainType, "'domainType' must not be null");
+		this.domainType = domainType;
+	}
+
+	/**
+	 * Specify a collection name to track change events from.
+	 * By default tracks all the collection in the {@link #mongoOperations} configured database.
+	 * @param collection a collection to use.
+	 * @see ReactiveMongoOperations#changeStream(String, ChangeStreamOptions, Class)
+	 */
+	public void setCollection(String collection) {
+		this.collection = collection;
+	}
+
+	/**
+	 * Specify a {@link ChangeStreamOptions}.
+	 * @param options the {@link ChangeStreamOptions} to use.
+	 * @see ReactiveMongoOperations#changeStream(String, ChangeStreamOptions, Class)
+	 */
+	public void setOptions(ChangeStreamOptions options) {
+		Assert.notNull(options, "'options' must not be null");
+		this.options = options;
+	}
+
+	/**
+	 * Configure this channel adapter to build a {@link Message} to produce
+	 * with a payload based on a {@link ChangeStreamEvent#getBody()} (by default)
+	 * or use a whole {@link ChangeStreamEvent} as a payload.
+	 * @param extractBody to extract {@link ChangeStreamEvent#getBody()} or not.
+	 */
+	public void setExtractBody(boolean extractBody) {
+		this.extractBody = extractBody;
+	}
+
+	@Override
+	public String getComponentType() {
+		return "mongo:change-stream-inbound-channel-adapter";
+	}
+
+	@Override
+	protected void doStart() {
+		Flux<Message<?>> changeStreamFlux =
+				this.mongoOperations.changeStream(this.collection, this.options, this.domainType)
+						.map(event ->
+								MessageBuilder
+										.withPayload(
+												!this.extractBody || event.getBody() == null
+														? event
+														: event.getBody())
+										.setHeader(MongoHeaders.COLLECTION_NAME, event.getCollectionName())
+										.setHeader(MongoHeaders.CHANGE_STREAM_OPERATION_TYPE, event.getOperationType())
+										.setHeader(MongoHeaders.CHANGE_STREAM_TIMESTAMP, event.getTimestamp())
+										.setHeader(MongoHeaders.CHANGE_STREAM_RESUME_TOKEN, event.getResumeToken())
+										.build());
+		subscribeToPublisher(changeStreamFlux);
+	}
+
+}