Merge branch '2.1.x'

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/mongo/MongoReactiveAutoConfiguration.java

@@ -19,11 +19,15 @@ package org.springframework.boot.autoconfigure.mongo;
 import java.util.stream.Collectors;
 
 import com.mongodb.MongoClientSettings;
+import com.mongodb.MongoClientSettings.Builder;
 import com.mongodb.connection.netty.NettyStreamFactoryFactory;
 import com.mongodb.reactivestreams.client.MongoClient;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import reactor.core.publisher.Flux;
 
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
@@ -60,23 +64,51 @@ public class MongoReactiveAutoConfiguration {
 	}
 
 	@Configuration(proxyBeanMethods = false)
-	@ConditionalOnClass(SocketChannel.class)
+	@ConditionalOnClass({ SocketChannel.class, NioEventLoopGroup.class })
 	static class NettyDriverConfiguration {
 
 		@Bean
 		@Order(Ordered.HIGHEST_PRECEDENCE)
-		public MongoClientSettingsBuilderCustomizer nettyDriverCustomizer(
+		public NettyDriverMongoClientSettingsBuilderCustomizer nettyDriverCustomizer(
 				ObjectProvider<MongoClientSettings> settings) {
-			return (builder) -> {
+			return new NettyDriverMongoClientSettingsBuilderCustomizer(settings);
-				if (!isStreamFactoryFactoryDefined(settings.getIfAvailable())) {
-					builder.streamFactoryFactory(
-							NettyStreamFactoryFactory.builder().build());
-				}
-			};
 		}
 
-		private boolean isStreamFactoryFactoryDefined(MongoClientSettings settings) {
+		private static final class NettyDriverMongoClientSettingsBuilderCustomizer
-			return settings != null && settings.getStreamFactoryFactory() != null;
+				implements MongoClientSettingsBuilderCustomizer, DisposableBean {
+
+			private final ObjectProvider<MongoClientSettings> settings;
+
+			private volatile EventLoopGroup eventLoopGroup;
+
+			private NettyDriverMongoClientSettingsBuilderCustomizer(
+					ObjectProvider<MongoClientSettings> settings) {
+				this.settings = settings;
+			}
+
+			@Override
+			public void customize(Builder builder) {
+				if (!isStreamFactoryFactoryDefined(this.settings.getIfAvailable())) {
+					NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
+					this.eventLoopGroup = eventLoopGroup;
+					builder.streamFactoryFactory(NettyStreamFactoryFactory.builder()
+							.eventLoopGroup(eventLoopGroup).build());
+				}
+			}
+
+			@Override
+			public void destroy() {
+				EventLoopGroup eventLoopGroup = this.eventLoopGroup;
+				if (eventLoopGroup != null) {
+					eventLoopGroup.shutdownGracefully().awaitUninterruptibly();
+					this.eventLoopGroup = null;
+				}
+			}
+
+			private boolean isStreamFactoryFactoryDefined(MongoClientSettings settings) {
+				return settings != null && settings.getStreamFactoryFactory() != null;
+			}
+
 		}
 
 	}

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/mongo/MongoReactiveAutoConfigurationTests.java

@@ -17,6 +17,7 @@
 package org.springframework.boot.autoconfigure.mongo;
 
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import com.mongodb.MongoClientSettings;
 import com.mongodb.ReadPreference;
@@ -25,6 +26,7 @@ import com.mongodb.connection.StreamFactory;
 import com.mongodb.connection.StreamFactoryFactory;
 import com.mongodb.connection.netty.NettyStreamFactoryFactory;
 import com.mongodb.reactivestreams.client.MongoClient;
+import io.netty.channel.EventLoopGroup;
 import org.junit.Test;
 
 import org.springframework.boot.autoconfigure.AutoConfigurations;
@@ -89,11 +91,17 @@ public class MongoReactiveAutoConfigurationTests {
 
 	@Test
 	public void nettyStreamFactoryFactoryIsConfiguredAutomatically() {
+		AtomicReference<EventLoopGroup> eventLoopGroupReference = new AtomicReference<>();
 		this.contextRunner.run((context) -> {
 			assertThat(context).hasSingleBean(MongoClient.class);
-			assertThat(getSettings(context).getStreamFactoryFactory())
+			StreamFactoryFactory factory = getSettings(context).getStreamFactoryFactory();
-					.isInstanceOf(NettyStreamFactoryFactory.class);
+			assertThat(factory).isInstanceOf(NettyStreamFactoryFactory.class);
+			EventLoopGroup eventLoopGroup = (EventLoopGroup) ReflectionTestUtils
+					.getField(factory, "eventLoopGroup");
+			assertThat(eventLoopGroup.isShutdown()).isFalse();
+			eventLoopGroupReference.set(eventLoopGroup);
 		});
+		assertThat(eventLoopGroupReference.get().isShutdown()).isTrue();
 	}
 
 	@Test