Upgrade dependencies; prepare for release

* Fix MongoDB tests for changed return type from Spring Data MongoDB

Author: artembilan

build.gradle

@@ -1,5 +1,5 @@
 buildscript {
-	ext.kotlinVersion = '1.3.50'
+	ext.kotlinVersion = '1.3.60'
 	repositories {
 		maven { url 'https://repo.spring.io/plugins-release' }
 	}
@@ -42,15 +42,15 @@ ext {
 	modifiedFiles =
 			files(grgit.status().unstaged.modified).filter{ f -> f.name.endsWith('.java') || f.name.endsWith('.kt') }
 
-	activeMqVersion = '5.15.10'
+	activeMqVersion = '5.15.11'
 	apacheSshdVersion = '2.3.0'
 	avroVersion = '1.9.1'
-	aspectjVersion = '1.9.4'
+	aspectjVersion = '1.9.5'
 	assertjVersion = '3.14.0'
 	assertkVersion = '0.20'
 	awaitilityVersion = '4.0.1'
 	boonVersion = '0.34'
-	commonsDbcp2Version = '2.6.0'
+	commonsDbcp2Version = '2.7.0'
 	commonsIoVersion = '2.6'
 	commonsNetVersion = '3.6'
 	curatorVersion = '4.2.0'
@@ -59,17 +59,17 @@ ext {
 	googleJsr305Version = '3.0.2'
 	groovyVersion = '2.5.8'
 	hamcrestVersion = '2.2'
-	hazelcastVersion = '3.12.3'
-	hibernateVersion = '5.4.7.Final'
+	hazelcastVersion = '3.12.4'
+	hibernateVersion = '5.4.9.Final'
 	hsqldbVersion = '2.5.0'
 	h2Version = '1.4.200'
-	jacksonVersion = '2.10.0'
+	jacksonVersion = '2.10.1'
 	javaxActivationVersion = '1.2.0'
 	javaxMailVersion = '1.6.2'
 	jmsApiVersion = '2.0.1'
 	jpa21ApiVersion = '1.0.2.Final'
 	jpaApiVersion = '2.2.1'
-	jrubyVersion = '9.2.8.0'
+	jrubyVersion = '9.2.9.0'
 	jschVersion = '0.1.55'
 	jsonpathVersion = '2.4.0'
 	junit4Version = '4.12'
@@ -78,8 +78,8 @@ ext {
 	kryoShadedVersion = '4.0.2'
 	lettuceVersion = '5.2.1.RELEASE'
 	log4jVersion = '2.12.1'
-	micrometerVersion = '1.3.1'
-	mockitoVersion = '3.1.0'
+	micrometerVersion = '1.3.2'
+	mockitoVersion = '3.2.0'
 	mysqlVersion = '8.0.18'
 	pahoMqttClientVersion = '1.2.0'
 	postgresVersion = '42.2.8'
@@ -95,7 +95,7 @@ ext {
 	springRetryVersion = '1.2.4.RELEASE'
 	springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.2.2.RELEASE'
 	springWsVersion = '3.0.8.RELEASE'
-	tomcatVersion = "9.0.27"
+	tomcatVersion = "9.0.29"
 	xstreamVersion = '1.4.11.1'
 }
 
@@ -330,7 +330,7 @@ subprojects { subproject ->
 
 	checkstyle {
 		configFile = file("$rootDir/src/checkstyle/checkstyle.xml")
-		toolVersion = '8.25'
+		toolVersion = '8.26'
 	}
 
 	artifacts {

spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java

@@ -22,6 +22,8 @@
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
+import reactor.core.Disposable;
+import reactor.core.Disposables;
 import reactor.core.publisher.EmitterProcessor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
@@ -47,6 +49,8 @@ public class FluxMessageChannel extends AbstractMessageChannel
 
 	private final ReplayProcessor<Boolean> subscribedSignal = ReplayProcessor.create(1);
 
+	private final Disposable.Composite upstreamSubscriptions = Disposables.composite();
+
 	public FluxMessageChannel() {
 		this.processor = EmitterProcessor.create(1, false);
 		this.sink = this.processor.sink(FluxSink.OverflowStrategy.BUFFER);
@@ -70,23 +74,25 @@ public void subscribe(Subscriber<? super Message<?>> subscriber) {
 
 	@Override
 	public void subscribeTo(Publisher<? extends Message<?>> publisher) {
-		Flux.from(publisher)
-				.delaySubscription(this.subscribedSignal.filter(Boolean::booleanValue).next())
-				.publishOn(Schedulers.boundedElastic())
-				.doOnNext((message) -> {
-					try {
-						send(message);
-					}
-					catch (Exception e) {
-						logger.warn("Error during processing event: " + message, e);
-					}
-				})
-				.subscribe();
+		this.upstreamSubscriptions.add(
+				Flux.from(publisher)
+						.delaySubscription(this.subscribedSignal.filter(Boolean::booleanValue).next())
+						.publishOn(Schedulers.boundedElastic())
+						.doOnNext((message) -> {
+							try {
+								send(message);
+							}
+							catch (Exception e) {
+								logger.warn("Error during processing event: " + message, e);
+							}
+						})
+						.subscribe());
 	}
 
 	@Override
 	public void destroy() {
 		this.subscribedSignal.onNext(false);
+		this.upstreamSubscriptions.dispose();
 		this.processor.onComplete();
 		super.destroy();
 	}

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests.java

@@ -20,7 +20,6 @@
 
 import java.util.List;
 
-import org.bson.Document;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -40,8 +39,9 @@
 import org.springframework.integration.mongodb.rules.MongoDbAvailableTests;
 import org.springframework.messaging.Message;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import com.mongodb.BasicDBObject;
 
 /**
  * @author Oleg Zhurakousky
@@ -50,8 +50,7 @@
  *
  * @since 2.2
  */
-@ContextConfiguration
-@RunWith(SpringJUnit4ClassRunner.class)
+@RunWith(SpringRunner.class)
 @DirtiesContext
 public class MongoDbInboundChannelAdapterIntegrationTests extends MongoDbAvailableTests {
 
@@ -102,7 +101,7 @@ public class MongoDbInboundChannelAdapterIntegrationTests extends MongoDbAvailab
 
 	@Test
 	@MongoDbAvailable
-	public void testWithDefaultMongoFactory() throws Exception {
+	public void testWithDefaultMongoFactory() {
 		this.mongoTemplate.save(createPerson("Bob"), "data");
 
 		this.mongoInboundAdapter.start();
@@ -119,13 +118,13 @@ public void testWithDefaultMongoFactory() throws Exception {
 
 	@Test
 	@MongoDbAvailable
-	public void testWithNamedMongoFactory() throws Exception {
+	public void testWithNamedMongoFactory() {
 		this.mongoTemplate.save(this.createPerson("Bob"), "data");
 
 		this.mongoInboundAdapterNamedFactory.start();
 
 		@SuppressWarnings("unchecked")
-		Message<List<Document>> message = (Message<List<Document>>) replyChannel.receive(10000);
+		Message<List<BasicDBObject>> message = (Message<List<BasicDBObject>>) replyChannel.receive(10000);
 		assertThat(message).isNotNull();
 		assertThat(message.getPayload().get(0).get("name")).isEqualTo("Bob");
 
@@ -135,7 +134,7 @@ public void testWithNamedMongoFactory() throws Exception {
 
 	@Test
 	@MongoDbAvailable
-	public void testWithMongoTemplate() throws Exception {
+	public void testWithMongoTemplate() {
 		this.mongoTemplate.save(this.createPerson("Bob"), "data");
 
 		this.mongoInboundAdapterWithTemplate.start();
@@ -151,7 +150,7 @@ public void testWithMongoTemplate() throws Exception {
 
 	@Test
 	@MongoDbAvailable
-	public void testWithNamedCollection() throws Exception {
+	public void testWithNamedCollection() {
 		this.mongoTemplate.save(this.createPerson("Bob"), "foo");
 
 		this.mongoInboundAdapterWithNamedCollection.start();
@@ -167,7 +166,7 @@ public void testWithNamedCollection() throws Exception {
 
 	@Test
 	@MongoDbAvailable
-	public void testWithQueryExpression() throws Exception {
+	public void testWithQueryExpression() {
 		this.mongoTemplate.save(this.createPerson("Bob"), "foo");
 		this.mongoTemplate.save(this.createPerson("Bob"), "foo");
 		this.mongoInboundAdapterWithQueryExpression.start();
@@ -181,7 +180,7 @@ public void testWithQueryExpression() throws Exception {
 
 	@Test
 	@MongoDbAvailable
-	public void testWithStringQueryExpression() throws Exception {
+	public void testWithStringQueryExpression() {
 		this.mongoTemplate.save(this.createPerson("Bob"), "foo");
 		this.mongoInboundAdapterWithStringQueryExpression.start();
 		@SuppressWarnings("unchecked")
@@ -193,7 +192,7 @@ public void testWithStringQueryExpression() throws Exception {
 
 	@Test
 	@MongoDbAvailable
-	public void testWithNamedCollectionExpression() throws Exception {
+	public void testWithNamedCollectionExpression() {
 		this.mongoTemplate.save(this.createPerson("Bob"), "foo");
 
 		this.mongoInboundAdapterWithNamedCollectionExpression.start();
@@ -209,7 +208,7 @@ public void testWithNamedCollectionExpression() throws Exception {
 
 	@Test
 	@MongoDbAvailable
-	public void testWithOnSuccessDisposition() throws Exception {
+	public void testWithOnSuccessDisposition() {
 		this.mongoTemplate.save(createPerson("Bob"), "data");
 
 		this.inboundAdapterWithOnSuccessDisposition.start();
@@ -227,7 +226,7 @@ public void testWithOnSuccessDisposition() throws Exception {
 
 	@Test
 	@MongoDbAvailable
-	public void testWithMongoConverter() throws Exception {
+	public void testWithMongoConverter() {
 		this.mongoTemplate.save(this.createPerson("Bob"), "data");
 
 		this.mongoInboundAdapterWithConverter.start();
@@ -244,25 +243,25 @@ public void testWithMongoConverter() throws Exception {
 
 	@Test(expected = BeanDefinitionParsingException.class)
 	@MongoDbAvailable
-	public void testFailureWithQueryAndQueryExpression() throws Exception {
+	public void testFailureWithQueryAndQueryExpression() {
 		new ClassPathXmlApplicationContext("inbound-fail-q-qex.xml", this.getClass()).close();
 	}
 
 	@Test(expected = BeanDefinitionParsingException.class)
 	@MongoDbAvailable
-	public void testFailureWithFactoryAndTemplate() throws Exception {
+	public void testFailureWithFactoryAndTemplate() {
 		new ClassPathXmlApplicationContext("inbound-fail-factory-template.xml", this.getClass()).close();
 	}
 
 	@Test(expected = BeanDefinitionParsingException.class)
 	@MongoDbAvailable
-	public void testFailureWithCollectionAndCollectionExpression() throws Exception {
+	public void testFailureWithCollectionAndCollectionExpression() {
 		new ClassPathXmlApplicationContext("inbound-fail-c-cex.xml", this.getClass()).close();
 	}
 
 	@Test(expected = BeanDefinitionParsingException.class)
 	@MongoDbAvailable
-	public void testFailureWithTemplateAndConverter() throws Exception {
+	public void testFailureWithTemplateAndConverter() {
 		new ClassPathXmlApplicationContext("inbound-fail-converter-template.xml", this.getClass()).close();
 	}