Mongo reactive client should use Netty transport

Author: franz1981

extensions/mongodb-client/runtime/src/main/java/io/quarkus/mongodb/runtime/MongoClients.java

@@ -35,6 +35,8 @@
 import org.bson.codecs.pojo.Conventions;
 import org.bson.codecs.pojo.PojoCodecProvider;
 import org.bson.codecs.pojo.PropertyCodecProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.mongodb.AuthenticationMechanism;
 import com.mongodb.Block;
@@ -51,18 +53,22 @@
 import com.mongodb.connection.ClusterConnectionMode;
 import com.mongodb.connection.ClusterSettings;
 import com.mongodb.connection.ConnectionPoolSettings;
+import com.mongodb.connection.NettyTransportSettings;
 import com.mongodb.connection.ServerSettings;
 import com.mongodb.connection.SocketSettings;
 import com.mongodb.connection.SslSettings;
 import com.mongodb.event.CommandListener;
 import com.mongodb.event.ConnectionPoolListener;
 import com.mongodb.reactivestreams.client.ReactiveContextProvider;
 
+import io.netty.channel.socket.SocketChannel;
 import io.quarkus.credentials.CredentialsProvider;
 import io.quarkus.credentials.runtime.CredentialsProviderFinder;
 import io.quarkus.mongodb.MongoClientName;
 import io.quarkus.mongodb.impl.ReactiveMongoClientImpl;
 import io.quarkus.mongodb.reactive.ReactiveMongoClient;
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.impl.VertxByteBufAllocator;
 
 /**
  * This class is sort of a producer for {@link MongoClient} and {@link ReactiveMongoClient}.
@@ -86,20 +92,23 @@ public class MongoClients {
     private final Map<String, ReactiveMongoClient> reactiveMongoClients = new HashMap<>();
     private final Instance<ReactiveContextProvider> reactiveContextProviders;
     private final Instance<MongoClientCustomizer> customizers;
+    private final Vertx vertx;
 
     public MongoClients(MongodbConfig mongodbConfig, MongoClientSupport mongoClientSupport,
             Instance<CodecProvider> codecProviders,
             Instance<PropertyCodecProvider> propertyCodecProviders,
             Instance<CommandListener> commandListeners,
             Instance<ReactiveContextProvider> reactiveContextProviders,
-            @Any Instance<MongoClientCustomizer> customizers) {
+            @Any Instance<MongoClientCustomizer> customizers,
+            Vertx vertx) {
         this.mongodbConfig = mongodbConfig;
         this.mongoClientSupport = mongoClientSupport;
         this.codecProviders = codecProviders;
         this.propertyCodecProviders = propertyCodecProviders;
         this.commandListeners = commandListeners;
         this.reactiveContextProviders = reactiveContextProviders;
         this.customizers = customizers;
+        this.vertx = vertx;
 
         try {
             //JDK bug workaround
@@ -255,6 +264,14 @@ private MongoClientSettings createMongoConfiguration(String name, MongoClientCon
         MongoClientSettings.Builder settings = MongoClientSettings.builder();
 
         if (isReactive) {
+            // we supports just NIO for now
+            if (!vertx.isNativeTransportEnabled()) {
+                var nettyStreaming = NettyTransportSettings.nettyBuilder()
+                        .allocator(VertxByteBufAllocator.POOLED_ALLOCATOR)
+                        .eventLoopGroup(vertx.nettyEventLoopGroup())
+                        .socketChannelClass(SocketChannel.class).build();
+                settings.transportSettings(nettyStreaming);
+            }
             reactiveContextProviders.stream().findAny().ifPresent(settings::contextProvider);
         }