Use the new Transport API for UDP transport

reactor · Apr 3, 2020 · ccd0442 · ccd0442
1 parent 279eff4
commit ccd0442
Show file tree

Hide file tree

Showing 20 changed files with 647 additions and 1,403 deletions.
diff --git a/src/main/java/reactor/netty/resources/NewConnectionProvider.java b/src/main/java/reactor/netty/resources/NewConnectionProvider.java
@@ -23,8 +23,12 @@
 import java.util.function.Supplier;
 
 import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
+import io.netty.resolver.AddressResolverGroup;
+import org.reactivestreams.Subscription;
+import reactor.core.CoreSubscriber;
 import reactor.core.Disposable;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoSink;
@@ -33,10 +37,14 @@
 import reactor.netty.ConnectionObserver;
 import reactor.netty.channel.BootstrapHandlers;
 import reactor.netty.channel.ChannelOperations;
+import reactor.netty.transport.TransportConfig;
+import reactor.netty.transport.TransportConnector;
 import reactor.util.Logger;
 import reactor.util.Loggers;
 import reactor.util.context.Context;
 
+import javax.annotation.Nullable;
+
 import static reactor.netty.ReactorNetty.format;
 
 /**
@@ -75,8 +83,33 @@ public Mono<? extends Connection> acquire(Bootstrap b) {
 			else {
 				f = bootstrap.bind();
 			}
-			DisposableConnect disposableConnect = new DisposableConnect(sink, f, bootstrap);
-			f.addListener(disposableConnect);
+			DisposableConnectChannelFuture disposableConnectChannelFuture = new DisposableConnectChannelFuture(sink, f, bootstrap);
+			f.addListener(disposableConnectChannelFuture);
+			sink.onCancel(disposableConnectChannelFuture);
+		});
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public <CONN> Mono<CONN> acquire(TransportConfig config, @Nullable Supplier<? extends SocketAddress> remoteAddress,
+			@Nullable AddressResolverGroup<?> resolverGroup) {
+		return Mono.create(sink -> {
+			SocketAddress remote = null;
+			if (remoteAddress != null) {
+				remote = Objects.requireNonNull(remoteAddress.get(), "Remote Address supplier returned null");
+			}
+
+			ConnectionObserver connectionObserver =
+					new NewConnectionObserver((MonoSink<Connection>) sink, config.connectionObserver());
+			DisposableConnect disposableConnect = new DisposableConnect((MonoSink<Connection>) sink, config.localAddress());
+			if (remote != null) {
+				TransportConnector.connect(config, remote, resolverGroup, connectionObserver)
+				                  .subscribe(disposableConnect);
+			}
+			else {
+				TransportConnector.bind(config, connectionObserver)
+				                  .subscribe(disposableConnect);
+			}
 			sink.onCancel(disposableConnect);
 		});
 	}
@@ -103,15 +136,74 @@ static void convertLazyRemoteAddress(Bootstrap b) {
 	}
 
 
-	static final class DisposableConnect
-			implements Disposable, ChannelFutureListener {
+	static final class DisposableConnect implements CoreSubscriber<Channel>, Disposable {
+		final MonoSink<Connection> sink;
+		final Supplier<? extends SocketAddress> localAddress;
+
+		Subscription s;
+
+		DisposableConnect(MonoSink<Connection> sink, @Nullable Supplier<? extends SocketAddress> localAddress) {
+			this.sink = sink;
+			this.localAddress = localAddress;
+		}
+
+		@Override
+		public Context currentContext() {
+			return sink.currentContext();
+		}
+
+		@Override
+		public void dispose() {
+			// TODO
+		}
+
+		@Override
+		public boolean isDisposed() {
+			// TODO
+			return false;
+		}
+
+		@Override
+		public void onComplete() {
+		}
+
+		@Override
+		public void onError(Throwable t) {
+			if (localAddress != null && (t instanceof BindException ||
+					// With epoll/kqueue transport it is
+					// io.netty.channel.unix.Errors$NativeIoException: bind(..) failed: Address already in use
+					(t instanceof IOException && t.getMessage() != null &&
+							t.getMessage().contains("Address already in use")))) {
+				sink.error(ChannelBindException.fail(localAddress.get(), null));
+			}
+			else {
+				sink.error(t);
+			}
+		}
+
+		@Override
+		public void onNext(Channel channel) {
+			if (log.isDebugEnabled()) {
+				log.debug(format(channel, "Connected new channel"));
+			}
+		}
+
+		@Override
+		public void onSubscribe(Subscription s) {
+			this.s = s;
+			s.request(Long.MAX_VALUE);
+		}
+	}
+
+
+	static final class DisposableConnectChannelFuture implements Disposable, ChannelFutureListener {
 
 		final MonoSink<Connection> sink;
 		final ChannelFuture f;
 		final Bootstrap bootstrap;
 
 
-		DisposableConnect(MonoSink<Connection> sink, ChannelFuture f, Bootstrap
+		DisposableConnectChannelFuture(MonoSink<Connection> sink, ChannelFuture f, Bootstrap
 				bootstrap) {
 			this.sink = sink;
 			this.f = f;