Merge branch '2.4.x'
Closes gh-24544
This commit is contained in:
@@ -69,11 +69,17 @@ public class NettyReactiveWebServerFactory extends AbstractReactiveWebServerFact
|
|||||||
public WebServer getWebServer(HttpHandler httpHandler) {
|
public WebServer getWebServer(HttpHandler httpHandler) {
|
||||||
HttpServer httpServer = createHttpServer();
|
HttpServer httpServer = createHttpServer();
|
||||||
ReactorHttpHandlerAdapter handlerAdapter = new ReactorHttpHandlerAdapter(httpHandler);
|
ReactorHttpHandlerAdapter handlerAdapter = new ReactorHttpHandlerAdapter(httpHandler);
|
||||||
NettyWebServer webServer = new NettyWebServer(httpServer, handlerAdapter, this.lifecycleTimeout, getShutdown());
|
NettyWebServer webServer = createNettyWebServer(httpServer, handlerAdapter, this.lifecycleTimeout,
|
||||||
|
getShutdown());
|
||||||
webServer.setRouteProviders(this.routeProviders);
|
webServer.setRouteProviders(this.routeProviders);
|
||||||
return webServer;
|
return webServer;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
NettyWebServer createNettyWebServer(HttpServer httpServer, ReactorHttpHandlerAdapter handlerAdapter,
|
||||||
|
Duration lifecycleTimeout, Shutdown shutdown) {
|
||||||
|
return new NettyWebServer(httpServer, handlerAdapter, lifecycleTimeout, shutdown);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Returns a mutable collection of the {@link NettyServerCustomizer}s that will be
|
* Returns a mutable collection of the {@link NettyServerCustomizer}s that will be
|
||||||
* applied to the Netty server builder.
|
* applied to the Netty server builder.
|
||||||
|
|||||||
@@ -21,6 +21,7 @@ import java.util.Collections;
|
|||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.function.BiFunction;
|
import java.util.function.BiFunction;
|
||||||
import java.util.function.Predicate;
|
import java.util.function.Predicate;
|
||||||
|
import java.util.function.Supplier;
|
||||||
|
|
||||||
import io.netty.channel.group.DefaultChannelGroup;
|
import io.netty.channel.group.DefaultChannelGroup;
|
||||||
import io.netty.channel.unix.Errors.NativeIoException;
|
import io.netty.channel.unix.Errors.NativeIoException;
|
||||||
@@ -106,11 +107,44 @@ public class NettyWebServer implements WebServer {
|
|||||||
});
|
});
|
||||||
throw new WebServerException("Unable to start Netty", ex);
|
throw new WebServerException("Unable to start Netty", ex);
|
||||||
}
|
}
|
||||||
logger.info("Netty started on port(s): " + getPort());
|
if (this.disposableServer != null) {
|
||||||
|
logger.info("Netty started" + getStartedOnMessage(this.disposableServer));
|
||||||
|
}
|
||||||
startDaemonAwaitThread(this.disposableServer);
|
startDaemonAwaitThread(this.disposableServer);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private String getStartedOnMessage(DisposableServer server) {
|
||||||
|
StringBuilder message = new StringBuilder();
|
||||||
|
tryAppend(message, "port %s", () -> server.port());
|
||||||
|
tryAppend(message, "path %s", () -> server.path());
|
||||||
|
return (message.length() > 0) ? " on " + message : "";
|
||||||
|
}
|
||||||
|
|
||||||
|
private void tryAppend(StringBuilder message, String format, Supplier<Object> supplier) {
|
||||||
|
try {
|
||||||
|
Object value = supplier.get();
|
||||||
|
message.append((message.length() != 0) ? " " : "");
|
||||||
|
message.append(String.format(format, value));
|
||||||
|
}
|
||||||
|
catch (UnsupportedOperationException ex) {
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
DisposableServer startHttpServer() {
|
||||||
|
HttpServer server = this.httpServer;
|
||||||
|
if (this.routeProviders.isEmpty()) {
|
||||||
|
server = server.handle(this.handler);
|
||||||
|
}
|
||||||
|
else {
|
||||||
|
server = server.route(this::applyRouteProviders);
|
||||||
|
}
|
||||||
|
if (this.lifecycleTimeout != null) {
|
||||||
|
return server.bindNow(this.lifecycleTimeout);
|
||||||
|
}
|
||||||
|
return server.bindNow();
|
||||||
|
}
|
||||||
|
|
||||||
private boolean isPermissionDenied(Throwable bindExceptionCause) {
|
private boolean isPermissionDenied(Throwable bindExceptionCause) {
|
||||||
try {
|
try {
|
||||||
if (bindExceptionCause instanceof NativeIoException) {
|
if (bindExceptionCause instanceof NativeIoException) {
|
||||||
@@ -131,20 +165,6 @@ public class NettyWebServer implements WebServer {
|
|||||||
this.gracefulShutdown.shutDownGracefully(callback);
|
this.gracefulShutdown.shutDownGracefully(callback);
|
||||||
}
|
}
|
||||||
|
|
||||||
private DisposableServer startHttpServer() {
|
|
||||||
HttpServer server = this.httpServer;
|
|
||||||
if (this.routeProviders.isEmpty()) {
|
|
||||||
server = server.handle(this.handler);
|
|
||||||
}
|
|
||||||
else {
|
|
||||||
server = server.route(this::applyRouteProviders);
|
|
||||||
}
|
|
||||||
if (this.lifecycleTimeout != null) {
|
|
||||||
return server.bindNow(this.lifecycleTimeout);
|
|
||||||
}
|
|
||||||
return server.bindNow();
|
|
||||||
}
|
|
||||||
|
|
||||||
private void applyRouteProviders(HttpServerRoutes routes) {
|
private void applyRouteProviders(HttpServerRoutes routes) {
|
||||||
for (NettyRouteProvider provider : this.routeProviders) {
|
for (NettyRouteProvider provider : this.routeProviders) {
|
||||||
routes = provider.apply(routes);
|
routes = provider.apply(routes);
|
||||||
@@ -190,7 +210,12 @@ public class NettyWebServer implements WebServer {
|
|||||||
@Override
|
@Override
|
||||||
public int getPort() {
|
public int getPort() {
|
||||||
if (this.disposableServer != null) {
|
if (this.disposableServer != null) {
|
||||||
return this.disposableServer.port();
|
try {
|
||||||
|
return this.disposableServer.port();
|
||||||
|
}
|
||||||
|
catch (UnsupportedOperationException ex) {
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -17,13 +17,19 @@
|
|||||||
package org.springframework.boot.web.embedded.netty;
|
package org.springframework.boot.web.embedded.netty;
|
||||||
|
|
||||||
import java.net.ConnectException;
|
import java.net.ConnectException;
|
||||||
|
import java.net.SocketAddress;
|
||||||
import java.time.Duration;
|
import java.time.Duration;
|
||||||
import java.util.Arrays;
|
import java.util.Arrays;
|
||||||
|
|
||||||
|
import io.netty.channel.Channel;
|
||||||
import org.awaitility.Awaitility;
|
import org.awaitility.Awaitility;
|
||||||
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.Test;
|
||||||
import org.mockito.InOrder;
|
import org.mockito.InOrder;
|
||||||
|
import reactor.core.CoreSubscriber;
|
||||||
|
import reactor.core.Disposable;
|
||||||
import reactor.core.publisher.Mono;
|
import reactor.core.publisher.Mono;
|
||||||
|
import reactor.netty.DisposableChannel;
|
||||||
|
import reactor.netty.DisposableServer;
|
||||||
import reactor.netty.http.server.HttpServer;
|
import reactor.netty.http.server.HttpServer;
|
||||||
import reactor.test.StepVerifier;
|
import reactor.test.StepVerifier;
|
||||||
|
|
||||||
@@ -34,6 +40,7 @@ import org.springframework.boot.web.server.Shutdown;
|
|||||||
import org.springframework.boot.web.server.Ssl;
|
import org.springframework.boot.web.server.Ssl;
|
||||||
import org.springframework.http.MediaType;
|
import org.springframework.http.MediaType;
|
||||||
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
|
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
|
||||||
|
import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
|
||||||
import org.springframework.web.reactive.function.BodyInserters;
|
import org.springframework.web.reactive.function.BodyInserters;
|
||||||
import org.springframework.web.reactive.function.client.WebClient;
|
import org.springframework.web.reactive.function.client.WebClient;
|
||||||
|
|
||||||
@@ -52,11 +59,6 @@ import static org.mockito.Mockito.mock;
|
|||||||
*/
|
*/
|
||||||
class NettyReactiveWebServerFactoryTests extends AbstractReactiveWebServerFactoryTests {
|
class NettyReactiveWebServerFactoryTests extends AbstractReactiveWebServerFactoryTests {
|
||||||
|
|
||||||
@Override
|
|
||||||
protected NettyReactiveWebServerFactory getFactory() {
|
|
||||||
return new NettyReactiveWebServerFactory(0);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
void exceptionIsThrownWhenPortIsAlreadyInUse() {
|
void exceptionIsThrownWhenPortIsAlreadyInUse() {
|
||||||
AbstractReactiveWebServerFactory factory = getFactory();
|
AbstractReactiveWebServerFactory factory = getFactory();
|
||||||
@@ -68,6 +70,14 @@ class NettyReactiveWebServerFactoryTests extends AbstractReactiveWebServerFactor
|
|||||||
.satisfies(this::portMatchesRequirement).withCauseInstanceOf(Throwable.class);
|
.satisfies(this::portMatchesRequirement).withCauseInstanceOf(Throwable.class);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void getPortWhenDisposableServerPortOperationIsUnsupportedReturnsMinusOne() {
|
||||||
|
NettyReactiveWebServerFactory factory = new NoPortNettyReactiveWebServerFactory(0);
|
||||||
|
this.webServer = factory.getWebServer(new EchoHandler());
|
||||||
|
this.webServer.start();
|
||||||
|
assertThat(this.webServer.getPort()).isEqualTo(-1);
|
||||||
|
}
|
||||||
|
|
||||||
private void portMatchesRequirement(PortInUseException exception) {
|
private void portMatchesRequirement(PortInUseException exception) {
|
||||||
assertThat(exception.getPort()).isEqualTo(this.webServer.getPort());
|
assertThat(exception.getPort()).isEqualTo(this.webServer.getPort());
|
||||||
}
|
}
|
||||||
@@ -143,4 +153,102 @@ class NettyReactiveWebServerFactoryTests extends AbstractReactiveWebServerFactor
|
|||||||
.retrieve().bodyToMono(String.class);
|
.retrieve().bodyToMono(String.class);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
protected NettyReactiveWebServerFactory getFactory() {
|
||||||
|
return new NettyReactiveWebServerFactory(0);
|
||||||
|
}
|
||||||
|
|
||||||
|
static class NoPortNettyReactiveWebServerFactory extends NettyReactiveWebServerFactory {
|
||||||
|
|
||||||
|
NoPortNettyReactiveWebServerFactory(int port) {
|
||||||
|
super(port);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
NettyWebServer createNettyWebServer(HttpServer httpServer, ReactorHttpHandlerAdapter handlerAdapter,
|
||||||
|
Duration lifecycleTimeout, Shutdown shutdown) {
|
||||||
|
return new NoPortNettyWebServer(httpServer, handlerAdapter, lifecycleTimeout, shutdown);
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
static class NoPortNettyWebServer extends NettyWebServer {
|
||||||
|
|
||||||
|
NoPortNettyWebServer(HttpServer httpServer, ReactorHttpHandlerAdapter handlerAdapter, Duration lifecycleTimeout,
|
||||||
|
Shutdown shutdown) {
|
||||||
|
super(httpServer, handlerAdapter, lifecycleTimeout, shutdown);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
DisposableServer startHttpServer() {
|
||||||
|
return new NoPortDisposableServer(super.startHttpServer());
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
static class NoPortDisposableServer implements DisposableServer {
|
||||||
|
|
||||||
|
private final DisposableServer delegate;
|
||||||
|
|
||||||
|
NoPortDisposableServer(DisposableServer delegate) {
|
||||||
|
this.delegate = delegate;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public SocketAddress address() {
|
||||||
|
return this.delegate.address();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String host() {
|
||||||
|
return this.delegate.host();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String path() {
|
||||||
|
return this.delegate.path();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Channel channel() {
|
||||||
|
return this.delegate.channel();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void dispose() {
|
||||||
|
this.delegate.dispose();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void disposeNow() {
|
||||||
|
this.delegate.disposeNow();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void disposeNow(Duration timeout) {
|
||||||
|
this.delegate.disposeNow(timeout);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CoreSubscriber<Void> disposeSubscriber() {
|
||||||
|
return this.delegate.disposeSubscriber();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public boolean isDisposed() {
|
||||||
|
return this.delegate.isDisposed();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Mono<Void> onDispose() {
|
||||||
|
return this.delegate.onDispose();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public DisposableChannel onDispose(Disposable onDispose) {
|
||||||
|
return this.delegate.onDispose(onDispose);
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user