diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerReadPublisher.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerReadPublisher.java index 28df306096..3a963712c8 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerReadPublisher.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerReadPublisher.java @@ -65,6 +65,30 @@ public abstract class AbstractListenerReadPublisher implements Publisher { @Nullable private volatile Throwable errorBeforeDemand; + private final String logPrefix; + + + public AbstractListenerReadPublisher() { + this(""); + } + + /** + * Create an instance with the given log prefix. + * @since 5.1 + */ + public AbstractListenerReadPublisher(String logPrefix) { + this.logPrefix = logPrefix; + } + + + /** + * Return the configured log message prefix. + * @since 5.1 + */ + public String getLogPrefix() { + return this.logPrefix; + } + // Publisher implementation... @@ -82,7 +106,7 @@ public abstract class AbstractListenerReadPublisher implements Publisher { * container. */ public final void onDataAvailable() { - logger.trace("I/O event onDataAvailable"); + logger.trace(getLogPrefix() + "onDataAvailable"); this.state.get().onDataAvailable(this); } @@ -91,7 +115,7 @@ public abstract class AbstractListenerReadPublisher implements Publisher { * all data has been read. */ public void onAllDataRead() { - logger.trace("I/O event onAllDataRead"); + logger.trace(getLogPrefix() + "onAllDataRead"); this.state.get().onAllDataRead(this); } @@ -100,7 +124,7 @@ public abstract class AbstractListenerReadPublisher implements Publisher { */ public final void onError(Throwable ex) { if (logger.isTraceEnabled()) { - logger.trace("I/O event onError: " + ex); + logger.trace(getLogPrefix() + "Connection error: " + ex); } this.state.get().onError(this, ex); } @@ -151,13 +175,13 @@ public abstract class AbstractListenerReadPublisher implements Publisher { Subscriber subscriber = this.subscriber; Assert.state(subscriber != null, "No subscriber"); if (logger.isTraceEnabled()) { - logger.trace("Data item read, publishing.."); + logger.trace(getLogPrefix() + "Publishing data read"); } subscriber.onNext(data); } else { if (logger.isTraceEnabled()) { - logger.trace("No more data to read"); + logger.trace(getLogPrefix() + "No more data to read"); } return true; } @@ -168,7 +192,7 @@ public abstract class AbstractListenerReadPublisher implements Publisher { private boolean changeState(State oldState, State newState) { boolean result = this.state.compareAndSet(oldState, newState); if (result && logger.isTraceEnabled()) { - logger.trace(oldState + " -> " + newState); + logger.trace(getLogPrefix() + oldState + " -> " + newState); } return result; } @@ -198,7 +222,7 @@ public abstract class AbstractListenerReadPublisher implements Publisher { @Override public final void request(long n) { if (logger.isTraceEnabled()) { - logger.trace("Signal request(" + n + ")"); + logger.trace(getLogPrefix() + n + " requested"); } state.get().request(AbstractListenerReadPublisher.this, n); } @@ -206,7 +230,7 @@ public abstract class AbstractListenerReadPublisher implements Publisher { @Override public final void cancel() { if (logger.isTraceEnabled()) { - logger.trace("Signal cancel()"); + logger.trace(getLogPrefix() + "Cancellation"); } state.get().cancel(AbstractListenerReadPublisher.this); } @@ -244,14 +268,15 @@ public abstract class AbstractListenerReadPublisher implements Publisher { subscriber.onSubscribe(subscription); publisher.changeState(SUBSCRIBING, NO_DEMAND); // Now safe to check "beforeDemand" flags, they won't change once in NO_DEMAND + String logPrefix = publisher.getLogPrefix(); if (publisher.completionBeforeDemand) { - publisher.logger.trace("Completed before demand"); + publisher.logger.trace(logPrefix + "Completed before demand"); publisher.state.get().onAllDataRead(publisher); } Throwable ex = publisher.errorBeforeDemand; if (ex != null) { if (publisher.logger.isTraceEnabled()) { - publisher.logger.trace("Completed with error before demand: " + ex); + publisher.logger.trace(logPrefix + "Completed with error before demand: " + ex); } publisher.state.get().onError(publisher, ex); } diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java index e51169c242..56bf254f6b 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java @@ -51,7 +51,32 @@ public abstract class AbstractListenerWriteFlushProcessor implements Processo private volatile boolean subscriberCompleted; - private final WriteResultPublisher resultPublisher = new WriteResultPublisher(); + private final WriteResultPublisher resultPublisher; + + private final String logPrefix; + + + public AbstractListenerWriteFlushProcessor() { + this(""); + } + + /** + * Create an instance with the given log prefix. + * @since 5.1 + */ + public AbstractListenerWriteFlushProcessor(String logPrefix) { + this.logPrefix = logPrefix; + this.resultPublisher = new WriteResultPublisher(logPrefix); + } + + + /** + * Create an instance with the given log prefix. + * @since 5.1 + */ + public String getLogPrefix() { + return this.logPrefix; + } // Subscriber methods and async I/O notification methods... @@ -63,7 +88,9 @@ public abstract class AbstractListenerWriteFlushProcessor implements Processo @Override public final void onNext(Publisher publisher) { - logger.trace("Received onNext publisher"); + if (logger.isTraceEnabled()) { + logger.trace(getLogPrefix() + "Received onNext publisher"); + } this.state.get().onNext(this, publisher); } @@ -74,7 +101,7 @@ public abstract class AbstractListenerWriteFlushProcessor implements Processo @Override public final void onError(Throwable ex) { if (logger.isTraceEnabled()) { - logger.trace("Received onError: " + ex); + logger.trace(getLogPrefix() + "Received onError: " + ex); } this.state.get().onError(this, ex); } @@ -85,7 +112,9 @@ public abstract class AbstractListenerWriteFlushProcessor implements Processo */ @Override public final void onComplete() { - logger.trace("Received onComplete"); + if (logger.isTraceEnabled()) { + logger.trace(getLogPrefix() + "Received onComplete"); + } this.state.get().onComplete(this); } @@ -103,7 +132,9 @@ public abstract class AbstractListenerWriteFlushProcessor implements Processo * container to cancel the upstream subscription. */ protected void cancel() { - logger.trace("Received request to cancel"); + if (logger.isTraceEnabled()) { + logger.trace(getLogPrefix() + "Received request to cancel"); + } if (this.subscription != null) { this.subscription.cancel(); } @@ -160,7 +191,7 @@ public abstract class AbstractListenerWriteFlushProcessor implements Processo private boolean changeState(State oldState, State newState) { boolean result = this.state.compareAndSet(oldState, newState); if (result && logger.isTraceEnabled()) { - logger.trace(oldState + " -> " + newState); + logger.trace(getLogPrefix() + oldState + " -> " + newState); } return result; } @@ -168,7 +199,7 @@ public abstract class AbstractListenerWriteFlushProcessor implements Processo private void flushIfPossible() { boolean result = isWritePossible(); if (logger.isTraceEnabled()) { - logger.trace("isWritePossible[" + result + "]"); + logger.trace(getLogPrefix() + "isWritePossible[" + result + "]"); } if (result) { onFlushPossible(); @@ -369,10 +400,12 @@ public abstract class AbstractListenerWriteFlushProcessor implements Processo @Override public void onComplete() { - if (this.processor.logger.isTraceEnabled()) { - this.processor.logger.trace(this.processor.state + " writeComplete"); + Log logger = this.processor.logger; + AtomicReference state = this.processor.state; + if (logger.isTraceEnabled()) { + logger.trace(this.processor.getLogPrefix() + state + " writeComplete"); } - this.processor.state.get().writeComplete(this.processor); + state.get().writeComplete(this.processor); } } } diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java index 5a1b5edaeb..020990a263 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java @@ -56,7 +56,32 @@ public abstract class AbstractListenerWriteProcessor implements Processor implements Processor implements Processor implements Processor implements Processor implements Processor implements Processor " + newState); + logger.trace(getLogPrefix() + oldState + " -> " + newState); } return result; } @@ -216,8 +247,8 @@ public abstract class AbstractListenerWriteProcessor implements Processor T getNativeRequest(); /** - * Return an id for the underlying connection, if available. + * Return an id representing the underlying connection, if available, or + * otherwise the identify of the request object. + * @since 5.1 + */ + public String getConnectionId() { + if (this.connectionId == null) { + this.connectionId = initConnectionId(); + if (this.connectionId == null) { + this.connectionId = ObjectUtils.getIdentityHexString(this); + } + } + return this.connectionId; + } + + /** + * Obtain the connection id, if available. * @since 5.1 */ @Nullable - public String getConnectionId() { + protected String initConnectionId() { return null; } + /** + * For internal use in logging at the HTTP adapter layer. + * @since 5.1 + */ + String getLogPrefix() { + if (this.logPrefix == null) { + this.logPrefix = "[" + getConnectionId() + "] "; + } + return this.logPrefix; + } + } diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractServerHttpResponse.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractServerHttpResponse.java index 4900f99ba7..58779ca3e0 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractServerHttpResponse.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractServerHttpResponse.java @@ -22,8 +22,6 @@ import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; import java.util.stream.Collectors; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -58,8 +56,6 @@ public abstract class AbstractServerHttpResponse implements ServerHttpResponse { private enum State {NEW, COMMITTING, COMMITTED} - private final Log logger = LogFactory.getLog(getClass()); - private final DataBufferFactory dataBufferFactory; @Nullable @@ -90,9 +86,6 @@ public abstract class AbstractServerHttpResponse implements ServerHttpResponse { @Override public boolean setStatusCode(@Nullable HttpStatus status) { if (this.state.get() == State.COMMITTED) { - if (logger.isTraceEnabled()) { - logger.trace("Ignoring status " + status + ": response already committed"); - } return false; } else { diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/DefaultServerHttpRequestBuilder.java b/spring-web/src/main/java/org/springframework/http/server/reactive/DefaultServerHttpRequestBuilder.java index 43f8525762..35aa7bf037 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/DefaultServerHttpRequestBuilder.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/DefaultServerHttpRequestBuilder.java @@ -32,6 +32,7 @@ import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; +import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; /** @@ -191,7 +192,6 @@ class DefaultServerHttpRequestBuilder implements ServerHttpRequest.Builder { private final ServerHttpRequest originalRequest; - @Nullable private final String requestId; @@ -207,7 +207,8 @@ class DefaultServerHttpRequestBuilder implements ServerHttpRequest.Builder { this.body = body; this.originalRequest = originalRequest; this.requestId = originalRequest instanceof AbstractServerHttpRequest ? - ((AbstractServerHttpRequest) originalRequest).getConnectionId() : null; + ((AbstractServerHttpRequest) originalRequest).getConnectionId() : + ObjectUtils.getIdentityHexString(originalRequest); } @Override diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/JettyHttpHandlerAdapter.java b/spring-web/src/main/java/org/springframework/http/server/reactive/JettyHttpHandlerAdapter.java index 5b5f7b602e..76bcef8549 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/JettyHttpHandlerAdapter.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/JettyHttpHandlerAdapter.java @@ -43,19 +43,21 @@ public class JettyHttpHandlerAdapter extends ServletHttpHandlerAdapter { @Override - protected ServerHttpResponse createResponse(HttpServletResponse response, - AsyncContext context) throws IOException { + protected ServletServerHttpResponse createResponse(HttpServletResponse response, + AsyncContext context, ServletServerHttpRequest request) throws IOException { - return new JettyServerHttpResponse(response, context, getDataBufferFactory(), getBufferSize()); + return new JettyServerHttpResponse( + response, context, getDataBufferFactory(), getBufferSize(), request); } private static final class JettyServerHttpResponse extends ServletServerHttpResponse { - public JettyServerHttpResponse(HttpServletResponse response, AsyncContext context, - DataBufferFactory factory, int bufferSize) throws IOException { + public JettyServerHttpResponse(HttpServletResponse response, AsyncContext asyncContext, + DataBufferFactory bufferFactory, int bufferSize, ServletServerHttpRequest request) + throws IOException { - super(response, context, factory, bufferSize); + super(response, asyncContext, bufferFactory, bufferSize, request); } @Override diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorHttpHandlerAdapter.java b/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorHttpHandlerAdapter.java index 2ea3214f87..f0e8539745 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorHttpHandlerAdapter.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorHttpHandlerAdapter.java @@ -68,13 +68,15 @@ public class ReactorHttpHandlerAdapter implements BiFunction logger.trace("Failed to complete: " + ex.getMessage())) - .doOnSuccess(aVoid -> logger.trace("Handling completed")); + .doOnError(ex -> logger.trace(logPrefix + "Failed to complete: " + ex.getMessage())) + .doOnSuccess(aVoid -> logger.trace(logPrefix + "Handling completed")); } } diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorServerHttpRequest.java b/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorServerHttpRequest.java index a3d7431cea..13ef0a4cc4 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorServerHttpRequest.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorServerHttpRequest.java @@ -175,7 +175,8 @@ class ReactorServerHttpRequest extends AbstractServerHttpRequest { } @Override - public String getConnectionId() { + @Nullable + protected String initConnectionId() { return this.request instanceof Connection ? ((Connection) this.request).channel().id().asShortText() : null; } diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/ServletHttpHandlerAdapter.java b/spring-web/src/main/java/org/springframework/http/server/reactive/ServletHttpHandlerAdapter.java index 97f5e8e8b4..e88c819cf3 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/ServletHttpHandlerAdapter.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/ServletHttpHandlerAdapter.java @@ -166,7 +166,7 @@ public class ServletHttpHandlerAdapter implements Servlet { AsyncContext asyncContext = request.startAsync(); asyncContext.setTimeout(-1); - ServerHttpRequest httpRequest; + ServletServerHttpRequest httpRequest; try { httpRequest = createRequest(((HttpServletRequest) request), asyncContext); } @@ -178,21 +178,23 @@ public class ServletHttpHandlerAdapter implements Servlet { asyncContext.complete(); return; } - ServerHttpResponse httpResponse = createResponse(((HttpServletResponse) response), asyncContext); + + ServerHttpResponse httpResponse = + createResponse(((HttpServletResponse) response), asyncContext, httpRequest); if (httpRequest.getMethod() == HttpMethod.HEAD) { httpResponse = new HttpHeadResponseDecorator(httpResponse); } AtomicBoolean isCompleted = new AtomicBoolean(); - HandlerResultAsyncListener listener = new HandlerResultAsyncListener(isCompleted); + HandlerResultAsyncListener listener = new HandlerResultAsyncListener(isCompleted, httpRequest); asyncContext.addListener(listener); - HandlerResultSubscriber subscriber = new HandlerResultSubscriber(asyncContext, isCompleted); + HandlerResultSubscriber subscriber = new HandlerResultSubscriber(asyncContext, isCompleted, httpRequest); this.httpHandler.handle(httpRequest, httpResponse).subscribe(subscriber); } - protected ServerHttpRequest createRequest(HttpServletRequest request, AsyncContext context) + protected ServletServerHttpRequest createRequest(HttpServletRequest request, AsyncContext context) throws IOException, URISyntaxException { Assert.notNull(this.servletPath, "Servlet path is not initialized"); @@ -200,8 +202,10 @@ public class ServletHttpHandlerAdapter implements Servlet { request, context, this.servletPath, getDataBufferFactory(), getBufferSize()); } - protected ServerHttpResponse createResponse(HttpServletResponse response, AsyncContext context) throws IOException { - return new ServletServerHttpResponse(response, context, getDataBufferFactory(), getBufferSize()); + protected ServletServerHttpResponse createResponse(HttpServletResponse response, + AsyncContext context, ServletServerHttpRequest request) throws IOException { + + return new ServletServerHttpResponse(response, context, getDataBufferFactory(), getBufferSize(), request); } @Override @@ -241,13 +245,17 @@ public class ServletHttpHandlerAdapter implements Servlet { private final AtomicBoolean isCompleted; - public HandlerResultAsyncListener(AtomicBoolean isCompleted) { + private final String logPrefix; + + + public HandlerResultAsyncListener(AtomicBoolean isCompleted, ServletServerHttpRequest httpRequest) { this.isCompleted = isCompleted; + this.logPrefix = httpRequest.getLogPrefix(); } @Override public void onTimeout(AsyncEvent event) { - logger.debug("Timeout notification"); + logger.debug(this.logPrefix + "Timeout notification"); AsyncContext context = event.getAsyncContext(); runIfAsyncNotComplete(context, this.isCompleted, context::complete); } @@ -255,7 +263,7 @@ public class ServletHttpHandlerAdapter implements Servlet { @Override public void onError(AsyncEvent event) { Throwable ex = event.getThrowable(); - logger.debug("Error notification: " + (ex != null ? ex : "")); + logger.debug(this.logPrefix + "Error notification: " + (ex != null ? ex : "")); AsyncContext context = event.getAsyncContext(); runIfAsyncNotComplete(context, this.isCompleted, context::complete); } @@ -278,9 +286,15 @@ public class ServletHttpHandlerAdapter implements Servlet { private final AtomicBoolean isCompleted; - public HandlerResultSubscriber(AsyncContext asyncContext, AtomicBoolean isCompleted) { + private final String logPrefix; + + + public HandlerResultSubscriber( + AsyncContext asyncContext, AtomicBoolean isCompleted, ServletServerHttpRequest httpRequest) { + this.asyncContext = asyncContext; this.isCompleted = isCompleted; + this.logPrefix = httpRequest.getLogPrefix(); } @Override @@ -295,16 +309,16 @@ public class ServletHttpHandlerAdapter implements Servlet { @Override public void onError(Throwable ex) { - logger.trace("Failed to complete: " + ex.getMessage()); + logger.trace(this.logPrefix + "Failed to complete: " + ex.getMessage()); runIfAsyncNotComplete(this.asyncContext, this.isCompleted, () -> { if (this.asyncContext.getResponse().isCommitted()) { - logger.trace("Dispatch to container, to raise the error on servlet thread"); + logger.trace(this.logPrefix + "Dispatch to container, to raise the error on servlet thread"); this.asyncContext.getRequest().setAttribute(WRITE_ERROR_ATTRIBUTE_NAME, ex); this.asyncContext.dispatch(); } else { try { - logger.trace("Setting ServletResponse status to 500 Server Error"); + logger.trace(this.logPrefix + "Setting ServletResponse status to 500 Server Error"); this.asyncContext.getResponse().resetBuffer(); ((HttpServletResponse) this.asyncContext.getResponse()).setStatus(500); } @@ -317,7 +331,7 @@ public class ServletHttpHandlerAdapter implements Servlet { @Override public void onComplete() { - logger.trace("Handling completed"); + logger.trace(this.logPrefix + "Handling completed"); runIfAsyncNotComplete(this.asyncContext, this.isCompleted, this.asyncContext::complete); } } diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpRequest.java b/spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpRequest.java index 86368d20ed..b386edae36 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpRequest.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpRequest.java @@ -204,7 +204,7 @@ class ServletServerHttpRequest extends AbstractServerHttpRequest { DataBuffer readFromInputStream() throws IOException { int read = this.request.getInputStream().read(this.buffer); if (logger.isTraceEnabled()) { - logger.trace("InputStream.read returned " + read + (read != -1 ? " bytes" : "")); + logger.trace(getLogPrefix() + "InputStream.read returned " + read + (read != -1 ? " bytes" : "")); } if (read > 0) { @@ -256,6 +256,7 @@ class ServletServerHttpRequest extends AbstractServerHttpRequest { private final ServletInputStream inputStream; public RequestBodyPublisher(ServletInputStream inputStream) { + super(ServletServerHttpRequest.this.getLogPrefix()); this.inputStream = inputStream; } diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpResponse.java b/spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpResponse.java index 0b88aabb54..effcecbf3a 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpResponse.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpResponse.java @@ -60,9 +60,11 @@ class ServletServerHttpResponse extends AbstractListenerServerHttpResponse { private volatile boolean flushOnNext; + private final ServletServerHttpRequest request; + public ServletServerHttpResponse(HttpServletResponse response, AsyncContext asyncContext, - DataBufferFactory bufferFactory, int bufferSize) throws IOException { + DataBufferFactory bufferFactory, int bufferSize, ServletServerHttpRequest request) throws IOException { super(bufferFactory); @@ -73,6 +75,7 @@ class ServletServerHttpResponse extends AbstractListenerServerHttpResponse { this.response = response; this.outputStream = response.getOutputStream(); this.bufferSize = bufferSize; + this.request = request; asyncContext.addListener(new ResponseAsyncListener()); @@ -265,6 +268,10 @@ class ServletServerHttpResponse extends AbstractListenerServerHttpResponse { private class ResponseBodyFlushProcessor extends AbstractListenerWriteFlushProcessor { + public ResponseBodyFlushProcessor() { + super(request.getLogPrefix()); + } + @Override protected Processor createWriteProcessor() { ResponseBodyProcessor processor = new ResponseBodyProcessor(); @@ -275,7 +282,7 @@ class ServletServerHttpResponse extends AbstractListenerServerHttpResponse { @Override protected void flush() throws IOException { if (logger.isTraceEnabled()) { - logger.trace("flush"); + logger.trace(getLogPrefix() + "flush"); } ServletServerHttpResponse.this.flush(); } @@ -294,6 +301,11 @@ class ServletServerHttpResponse extends AbstractListenerServerHttpResponse { private class ResponseBodyProcessor extends AbstractListenerWriteProcessor { + + public ResponseBodyProcessor() { + super(request.getLogPrefix()); + } + @Override protected boolean isWritePossible() { return ServletServerHttpResponse.this.isWritePossible(); @@ -308,23 +320,23 @@ class ServletServerHttpResponse extends AbstractListenerServerHttpResponse { protected boolean write(DataBuffer dataBuffer) throws IOException { if (ServletServerHttpResponse.this.flushOnNext) { if (logger.isTraceEnabled()) { - logger.trace("flush"); + logger.trace(getLogPrefix() + "flush"); } flush(); } boolean ready = ServletServerHttpResponse.this.isWritePossible(); if (logger.isTraceEnabled()) { - logger.trace("write: " + dataBuffer + " ready: " + ready); + logger.trace(getLogPrefix() + "write: " + dataBuffer + " ready: " + ready); } int remaining = dataBuffer.readableByteCount(); if (ready && remaining > 0) { int written = writeToOutputStream(dataBuffer); if (logger.isTraceEnabled()) { - logger.trace("written: " + written + " total: " + remaining); + logger.trace(getLogPrefix() + "written: " + written + " total: " + remaining); } if (written == remaining) { if (logger.isTraceEnabled()) { - logger.trace("releaseData: " + dataBuffer); + logger.trace(getLogPrefix() + "releaseData: " + dataBuffer); } DataBufferUtils.release(dataBuffer); return true; diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/TomcatHttpHandlerAdapter.java b/spring-web/src/main/java/org/springframework/http/server/reactive/TomcatHttpHandlerAdapter.java index 792ddfd839..e7e5f4ff82 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/TomcatHttpHandlerAdapter.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/TomcatHttpHandlerAdapter.java @@ -50,19 +50,20 @@ public class TomcatHttpHandlerAdapter extends ServletHttpHandlerAdapter { @Override - protected ServerHttpRequest createRequest(HttpServletRequest request, AsyncContext asyncContext) + protected ServletServerHttpRequest createRequest(HttpServletRequest request, AsyncContext asyncContext) throws IOException, URISyntaxException { Assert.notNull(getServletPath(), "servletPath is not initialized."); - return new TomcatServerHttpRequest(request, asyncContext, getServletPath(), - getDataBufferFactory(), getBufferSize()); + return new TomcatServerHttpRequest( + request, asyncContext, getServletPath(), getDataBufferFactory(), getBufferSize()); } @Override - protected ServerHttpResponse createResponse(HttpServletResponse response, AsyncContext cxt) - throws IOException { + protected ServletServerHttpResponse createResponse(HttpServletResponse response, + AsyncContext asyncContext, ServletServerHttpRequest request) throws IOException { - return new TomcatServerHttpResponse(response, cxt, getDataBufferFactory(), getBufferSize()); + return new TomcatServerHttpResponse( + response, asyncContext, getDataBufferFactory(), getBufferSize(), request); } @@ -86,7 +87,7 @@ public class TomcatHttpHandlerAdapter extends ServletHttpHandlerAdapter { ServletRequest request = getNativeRequest(); int read = ((CoyoteInputStream) request.getInputStream()).read(byteBuffer); if (logger.isTraceEnabled()) { - logger.trace("read:" + read); + logger.trace(getLogPrefix() + "read:" + read); } if (read > 0) { @@ -113,9 +114,9 @@ public class TomcatHttpHandlerAdapter extends ServletHttpHandlerAdapter { private static final class TomcatServerHttpResponse extends ServletServerHttpResponse { public TomcatServerHttpResponse(HttpServletResponse response, AsyncContext context, - DataBufferFactory factory, int bufferSize) throws IOException { + DataBufferFactory factory, int bufferSize, ServletServerHttpRequest request) throws IOException { - super(response, context, factory, bufferSize); + super(response, context, factory, bufferSize, request); } @Override diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowHttpHandlerAdapter.java b/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowHttpHandlerAdapter.java index f7b55aea0c..af04f06169 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowHttpHandlerAdapter.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowHttpHandlerAdapter.java @@ -66,7 +66,7 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle @Override public void handleRequest(HttpServerExchange exchange) { - ServerHttpRequest request = null; + UndertowServerHttpRequest request = null; try { request = new UndertowServerHttpRequest(exchange, getDataBufferFactory()); } @@ -77,13 +77,13 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle exchange.setStatusCode(400); return; } - ServerHttpResponse response = new UndertowServerHttpResponse(exchange, getDataBufferFactory()); + ServerHttpResponse response = new UndertowServerHttpResponse(exchange, getDataBufferFactory(), request); if (request.getMethod() == HttpMethod.HEAD) { response = new HttpHeadResponseDecorator(response); } - HandlerResultSubscriber resultSubscriber = new HandlerResultSubscriber(exchange); + HandlerResultSubscriber resultSubscriber = new HandlerResultSubscriber(exchange, request); this.httpHandler.handle(request, response).subscribe(resultSubscriber); } @@ -92,8 +92,12 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle private final HttpServerExchange exchange; - public HandlerResultSubscriber(HttpServerExchange exchange) { + private final String logPrefix; + + + public HandlerResultSubscriber(HttpServerExchange exchange, UndertowServerHttpRequest request) { this.exchange = exchange; + this.logPrefix = request.getLogPrefix(); } @Override @@ -108,10 +112,10 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle @Override public void onError(Throwable ex) { - logger.trace("Failed to complete: " + ex.getMessage()); + logger.trace(this.logPrefix + "Failed to complete: " + ex.getMessage()); if (this.exchange.isResponseStarted()) { try { - logger.debug("Closing connection"); + logger.debug(this.logPrefix + "Closing connection"); this.exchange.getConnection().close(); } catch (IOException ex2) { @@ -119,7 +123,7 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle } } else { - logger.debug("Setting HttpServerExchange status to 500 Server Error"); + logger.debug(this.logPrefix + "Setting HttpServerExchange status to 500 Server Error"); this.exchange.setStatusCode(500); this.exchange.endExchange(); } @@ -127,7 +131,7 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle @Override public void onComplete() { - logger.trace("Handling completed"); + logger.trace(this.logPrefix + "Handling completed"); this.exchange.endExchange(); } } diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpRequest.java b/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpRequest.java index 093e13638a..481c4b8c97 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpRequest.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpRequest.java @@ -60,17 +60,14 @@ class UndertowServerHttpRequest extends AbstractServerHttpRequest { private final RequestBodyPublisher body; - private final String connectionId; - public UndertowServerHttpRequest(HttpServerExchange exchange, DataBufferFactory bufferFactory) throws URISyntaxException { super(initUri(exchange), "", initHeaders(exchange)); this.exchange = exchange; - this.body = new RequestBodyPublisher(exchange, bufferFactory); + this.body = new RequestBodyPublisher(exchange, bufferFactory, getLogPrefix()); this.body.registerListeners(exchange); - this.connectionId = ObjectUtils.getIdentityHexString(this.exchange.getConnection()); } private static URI initUri(HttpServerExchange exchange) throws URISyntaxException { @@ -132,8 +129,8 @@ class UndertowServerHttpRequest extends AbstractServerHttpRequest { } @Override - public String getConnectionId() { - return this.connectionId; + protected String initConnectionId() { + return ObjectUtils.getIdentityHexString(this.exchange.getConnection()); } @@ -145,7 +142,8 @@ class UndertowServerHttpRequest extends AbstractServerHttpRequest { private final ByteBufferPool byteBufferPool; - public RequestBodyPublisher(HttpServerExchange exchange, DataBufferFactory bufferFactory) { + public RequestBodyPublisher(HttpServerExchange exchange, DataBufferFactory bufferFactory, String logPrefix) { + super(logPrefix); this.channel = exchange.getRequestChannel(); this.bufferFactory = bufferFactory; this.byteBufferPool = exchange.getConnection().getByteBufferPool(); @@ -183,7 +181,7 @@ class UndertowServerHttpRequest extends AbstractServerHttpRequest { int read = this.channel.read(byteBuffer); if (logger.isTraceEnabled()) { - logger.trace("Channel.read returned " + read + (read != -1 ? " bytes" : "")); + logger.trace(getLogPrefix() + "Channel.read returned " + read + (read != -1 ? " bytes" : "")); } if (read > 0) { diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpResponse.java b/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpResponse.java index b7aa467ab8..b60ba90360 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpResponse.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpResponse.java @@ -52,14 +52,19 @@ class UndertowServerHttpResponse extends AbstractListenerServerHttpResponse impl private final HttpServerExchange exchange; + private final UndertowServerHttpRequest request; + @Nullable private StreamSinkChannel responseChannel; - public UndertowServerHttpResponse(HttpServerExchange exchange, DataBufferFactory bufferFactory) { + public UndertowServerHttpResponse( + HttpServerExchange exchange, DataBufferFactory bufferFactory, UndertowServerHttpRequest request) { + super(bufferFactory); Assert.notNull(exchange, "HttpServerExchange must not be null"); this.exchange = exchange; + this.request = request; } @@ -146,6 +151,7 @@ class UndertowServerHttpResponse extends AbstractListenerServerHttpResponse impl public ResponseBodyProcessor(StreamSinkChannel channel) { + super(request.getLogPrefix()); Assert.notNull(channel, "StreamSinkChannel must not be null"); this.channel = channel; this.channel.getWriteSetter().set(c -> { @@ -168,7 +174,7 @@ class UndertowServerHttpResponse extends AbstractListenerServerHttpResponse impl return false; } if (logger.isTraceEnabled()) { - logger.trace("write: " + dataBuffer); + logger.trace(getLogPrefix() + "write (" + dataBuffer.readableByteCount() + " bytes)"); } // Track write listener calls from here on.. @@ -178,7 +184,7 @@ class UndertowServerHttpResponse extends AbstractListenerServerHttpResponse impl int written = writeByteBuffer(buffer); if (logger.isTraceEnabled()) { - logger.trace("written: " + written + " total: " + total); + logger.trace(getLogPrefix() + "written " + written + ", " + total + " remaining"); } if (written != total) { return false; @@ -188,7 +194,7 @@ class UndertowServerHttpResponse extends AbstractListenerServerHttpResponse impl this.writePossible = true; if (logger.isTraceEnabled()) { - logger.trace("releaseData: " + dataBuffer); + logger.trace(getLogPrefix() + "releaseData: " + dataBuffer); } DataBufferUtils.release(dataBuffer); this.byteBuffer = null; @@ -233,6 +239,10 @@ class UndertowServerHttpResponse extends AbstractListenerServerHttpResponse impl private class ResponseBodyFlushProcessor extends AbstractListenerWriteFlushProcessor { + public ResponseBodyFlushProcessor() { + super(request.getLogPrefix()); + } + @Override protected Processor createWriteProcessor() { return UndertowServerHttpResponse.this.createBodyProcessor(); @@ -243,7 +253,7 @@ class UndertowServerHttpResponse extends AbstractListenerServerHttpResponse impl StreamSinkChannel channel = UndertowServerHttpResponse.this.responseChannel; if (channel != null) { if (logger.isTraceEnabled()) { - logger.trace("flush"); + logger.trace(getLogPrefix() + "flush"); } channel.flush(); } diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/WriteResultPublisher.java b/spring-web/src/main/java/org/springframework/http/server/reactive/WriteResultPublisher.java index 6b74466a2c..37bb8ceebc 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/WriteResultPublisher.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/WriteResultPublisher.java @@ -50,11 +50,18 @@ class WriteResultPublisher implements Publisher { @Nullable private volatile Throwable errorBeforeSubscribed; + private final String logPrefix; + + + public WriteResultPublisher(String logPrefix) { + this.logPrefix = logPrefix; + } + @Override public final void subscribe(Subscriber subscriber) { if (logger.isTraceEnabled()) { - logger.trace(this.state + " subscribe: " + subscriber); + logger.trace(this.logPrefix + this.state + " subscribe: " + subscriber); } this.state.get().subscribe(this, subscriber); } @@ -64,7 +71,7 @@ class WriteResultPublisher implements Publisher { */ public void publishComplete() { if (logger.isTraceEnabled()) { - logger.trace(this.state + " publishComplete"); + logger.trace(this.logPrefix + this.state + " publishComplete"); } this.state.get().publishComplete(this); } @@ -74,7 +81,7 @@ class WriteResultPublisher implements Publisher { */ public void publishError(Throwable t) { if (logger.isTraceEnabled()) { - logger.trace(this.state + " publishError: " + t); + logger.trace(this.logPrefix + this.state + " publishError: " + t); } this.state.get().publishError(this, t); } @@ -99,7 +106,7 @@ class WriteResultPublisher implements Publisher { @Override public final void request(long n) { if (logger.isTraceEnabled()) { - logger.trace(state() + " request: " + n); + logger.trace(this.publisher.logPrefix + state() + " request: " + n); } state().request(this.publisher, n); } @@ -107,7 +114,7 @@ class WriteResultPublisher implements Publisher { @Override public final void cancel() { if (logger.isTraceEnabled()) { - logger.trace(state() + " cancel"); + logger.trace(this.publisher.logPrefix + state() + " cancel"); } state().cancel(this.publisher); } diff --git a/spring-web/src/main/java/org/springframework/web/server/adapter/HttpWebHandlerAdapter.java b/spring-web/src/main/java/org/springframework/web/server/adapter/HttpWebHandlerAdapter.java index cfc82e302e..1ab589e4fd 100644 --- a/spring-web/src/main/java/org/springframework/web/server/adapter/HttpWebHandlerAdapter.java +++ b/spring-web/src/main/java/org/springframework/web/server/adapter/HttpWebHandlerAdapter.java @@ -231,14 +231,8 @@ public class HttpWebHandlerAdapter extends WebHandlerDecorator implements HttpHa } private String initLogId(ServerHttpRequest request) { - String logId = null; - if (request instanceof AbstractServerHttpRequest) { - logId = ((AbstractServerHttpRequest) request).getConnectionId(); - } - if (logId == null) { - logId = ObjectUtils.getIdentityHexString(request); - } - return logId; + return request instanceof AbstractServerHttpRequest ? + ((AbstractServerHttpRequest) request).getConnectionId() : ObjectUtils.getIdentityHexString(request); } private void logExchange(ServerWebExchange exchange) { diff --git a/spring-web/src/test/java/org/springframework/http/server/reactive/ListenerReadPublisherTests.java b/spring-web/src/test/java/org/springframework/http/server/reactive/ListenerReadPublisherTests.java index 70dc487a69..7a48c6b311 100644 --- a/spring-web/src/test/java/org/springframework/http/server/reactive/ListenerReadPublisherTests.java +++ b/spring-web/src/test/java/org/springframework/http/server/reactive/ListenerReadPublisherTests.java @@ -56,6 +56,10 @@ public class ListenerReadPublisherTests { private int readCalls = 0; + public TestListenerReadPublisher() { + super(""); + } + @Override protected void checkOnDataAvailable() { // no-op diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/HandshakeInfo.java b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/HandshakeInfo.java index 46bc202080..67a9b8064b 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/HandshakeInfo.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/HandshakeInfo.java @@ -48,6 +48,9 @@ public class HandshakeInfo { private final Map attributes; + @Nullable + private final String logPrefix; + /** * Constructor with information about the handshake. @@ -57,7 +60,7 @@ public class HandshakeInfo { * @param protocol the negotiated sub-protocol (may be {@code null}) */ public HandshakeInfo(URI uri, HttpHeaders headers, Mono principal, @Nullable String protocol) { - this(uri, headers, principal, protocol, Collections.emptyMap()); + this(uri, headers, principal, protocol, Collections.emptyMap(), null); } /** @@ -67,10 +70,12 @@ public class HandshakeInfo { * @param principal the principal for the session * @param protocol the negotiated sub-protocol (may be {@code null}) * @param attributes initial attributes to use for the WebSocket session + * @param logPrefix log prefix used during the handshake for correlating log + * messages, if any. * @since 5.1 */ public HandshakeInfo(URI uri, HttpHeaders headers, Mono principal, - @Nullable String protocol, Map attributes) { + @Nullable String protocol, Map attributes, @Nullable String logPrefix) { Assert.notNull(uri, "URI is required"); Assert.notNull(headers, "HttpHeaders are required"); @@ -82,6 +87,7 @@ public class HandshakeInfo { this.principalMono = principal; this.protocol = protocol; this.attributes = attributes; + this.logPrefix = logPrefix; } @@ -126,6 +132,16 @@ public class HandshakeInfo { return this.attributes; } + /** + * A log prefix used in the handshake to correlate log messages, if any. + * @return a log prefix, or {@code null} + * @since 5.1 + */ + @Nullable + public String getLogPrefix() { + return this.logPrefix; + } + @Override public String toString() { diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/WebSocketMessage.java b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/WebSocketMessage.java index 228c5bab1a..8fefa88278 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/WebSocketMessage.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/WebSocketMessage.java @@ -134,10 +134,31 @@ public class WebSocketMessage { return this.type.hashCode() * 29 + this.payload.hashCode(); } + @Override + public String toString() { + return "WebSocket " + this.type.name() + " message (" + this.payload.readableByteCount() + " bytes)"; + } /** * WebSocket message types. */ - public enum Type { TEXT, BINARY, PING, PONG } + public enum Type { + /** + * Text WebSocket message. + */ + TEXT, + /** + * Binary WebSocket message. + */ + BINARY, + /** + * WebSocket ping. + */ + PING, + /** + * WebSocket pong. + */ + PONG; + } } diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/AbstractListenerWebSocketSession.java b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/AbstractListenerWebSocketSession.java index b0d4752aef..76e28dc917 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/AbstractListenerWebSocketSession.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/AbstractListenerWebSocketSession.java @@ -65,7 +65,7 @@ public abstract class AbstractListenerWebSocketSession extends AbstractWebSoc @Nullable private final MonoProcessor completionMono; - private final WebSocketReceivePublisher receivePublisher = new WebSocketReceivePublisher(); + private final WebSocketReceivePublisher receivePublisher; @Nullable private volatile WebSocketSendProcessor sendProcessor; @@ -90,13 +90,19 @@ public abstract class AbstractListenerWebSocketSession extends AbstractWebSoc * Alternative constructor with completion {@code Mono<Void>} to propagate * the session completion (success or error) (for client-side use). */ - public AbstractListenerWebSocketSession(T delegate, String id, HandshakeInfo handshakeInfo, + public AbstractListenerWebSocketSession(T delegate, String id, HandshakeInfo info, DataBufferFactory bufferFactory, @Nullable MonoProcessor completionMono) { - super(delegate, id, handshakeInfo, bufferFactory); + super(delegate, id, info, bufferFactory); + this.receivePublisher = new WebSocketReceivePublisher(initLogPrefix(info, id)); this.completionMono = completionMono; } + private static String initLogPrefix(HandshakeInfo info, String id) { + return info.getLogPrefix() != null ? info.getLogPrefix() : "[" + id + "] "; + } + + protected WebSocketSendProcessor getSendProcessor() { WebSocketSendProcessor sendProcessor = this.sendProcessor; @@ -223,15 +229,23 @@ public abstract class AbstractListenerWebSocketSession extends AbstractWebSoc private volatile Queue pendingMessages = Queues.unbounded(Queues.SMALL_BUFFER_SIZE).get(); + WebSocketReceivePublisher(String logPrefix) { + super(logPrefix); + if (logger.isDebugEnabled()) { + logger.debug(getLogPrefix() + "Session id '" + getId() + "' for " + getHandshakeInfo().getUri()); + } + } + + @Override protected void checkOnDataAvailable() { resumeReceiving(); - if (!this.pendingMessages.isEmpty()) { - logger.trace("checkOnDataAvailable, " + this.pendingMessages.size() + " pending messages"); - onDataAvailable(); + int size = this.pendingMessages.size(); + if (logger.isTraceEnabled()) { + logger.trace(getLogPrefix() + "checkOnDataAvailable (" + size + " pending)"); } - else { - logger.trace("checkOnDataAvailable, 0 pending messages"); + if (size > 0) { + onDataAvailable(); } } @@ -248,7 +262,7 @@ public abstract class AbstractListenerWebSocketSession extends AbstractWebSoc void handleMessage(WebSocketMessage message) { if (logger.isTraceEnabled()) { - logger.trace("Received " + message); + logger.trace(getLogPrefix() + "Received " + message); } if (!this.pendingMessages.offer(message)) { throw new IllegalStateException( @@ -266,10 +280,16 @@ public abstract class AbstractListenerWebSocketSession extends AbstractWebSoc private volatile boolean isReady = true; + + WebSocketSendProcessor() { + super(receivePublisher.getLogPrefix()); + } + + @Override protected boolean write(WebSocketMessage message) throws IOException { if (logger.isTraceEnabled()) { - logger.trace("Sending " + message); + logger.trace(getLogPrefix() + "Sending " + message); } return sendMessage(message); } @@ -290,8 +310,8 @@ public abstract class AbstractListenerWebSocketSession extends AbstractWebSoc * async completion callback into simple flow control. */ public void setReadyToSend(boolean ready) { - if (ready) { - logger.trace("Ready to send again"); + if (ready && logger.isTraceEnabled()) { + logger.trace(getLogPrefix() + "Ready to send"); } this.isReady = ready; } diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/JettyWebSocketClient.java b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/JettyWebSocketClient.java index b60d8fcb07..fca91e2966 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/JettyWebSocketClient.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/JettyWebSocketClient.java @@ -20,6 +20,7 @@ import java.net.URI; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.eclipse.jetty.websocket.api.Session; import org.eclipse.jetty.websocket.api.UpgradeRequest; import org.eclipse.jetty.websocket.api.UpgradeResponse; import org.eclipse.jetty.websocket.client.ClientUpgradeRequest; @@ -154,17 +155,17 @@ public class JettyWebSocketClient implements WebSocketClient, Lifecycle { } private Object createHandler(URI url, WebSocketHandler handler, MonoProcessor completion) { - return new JettyWebSocketHandlerAdapter(handler, - session -> { - if (logger.isDebugEnabled()) { - logger.debug("Connected to " + url); - } - HttpHeaders responseHeaders = new HttpHeaders(); - session.getUpgradeResponse().getHeaders().forEach(responseHeaders::put); - String protocol = responseHeaders.getFirst("Sec-WebSocket-Protocol"); - HandshakeInfo info = new HandshakeInfo(url, responseHeaders, Mono.empty(), protocol); - return new JettyWebSocketSession(session, info, this.bufferFactory, completion); - }); + return new JettyWebSocketHandlerAdapter(handler, session -> { + HandshakeInfo info = createHandshakeInfo(url, session); + return new JettyWebSocketSession(session, info, this.bufferFactory, completion); + }); + } + + private HandshakeInfo createHandshakeInfo(URI url, Session jettySession) { + HttpHeaders headers = new HttpHeaders(); + jettySession.getUpgradeResponse().getHeaders().forEach(headers::put); + String protocol = headers.getFirst("Sec-WebSocket-Protocol"); + return new HandshakeInfo(url, headers, Mono.empty(), protocol); } diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/ReactorNettyWebSocketClient.java b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/ReactorNettyWebSocketClient.java index 0f8215e1ec..9b0ef3ce80 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/ReactorNettyWebSocketClient.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/ReactorNettyWebSocketClient.java @@ -76,24 +76,26 @@ public class ReactorNettyWebSocketClient implements WebSocketClient { @Override public Mono execute(URI url, HttpHeaders requestHeaders, WebSocketHandler handler) { - if (logger.isDebugEnabled()) { - logger.debug("Connecting to " + url); - } return getHttpClient() .headers(nettyHeaders -> setNettyHeaders(requestHeaders, nettyHeaders)) .websocket(StringUtils.collectionToCommaDelimitedString(handler.getSubProtocols())) .uri(url.toString()) .handle((inbound, outbound) -> { - if (logger.isDebugEnabled()) { - logger.debug("Connected to " + url); - } HttpHeaders responseHeaders = toHttpHeaders(inbound); String protocol = responseHeaders.getFirst("Sec-WebSocket-Protocol"); HandshakeInfo info = new HandshakeInfo(url, responseHeaders, Mono.empty(), protocol); NettyDataBufferFactory factory = new NettyDataBufferFactory(outbound.alloc()); WebSocketSession session = new ReactorNettyWebSocketSession(inbound, outbound, info, factory); + if (logger.isDebugEnabled()) { + logger.debug("Started session '" + session.getId() + "' for " + url); + } return handler.handle(session); }) + .doOnRequest(n -> { + if (logger.isDebugEnabled()) { + logger.debug("Connecting to " + url); + } + }) .next(); } diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/StandardWebSocketClient.java b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/StandardWebSocketClient.java index 15b45d7db6..5000aadf2c 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/StandardWebSocketClient.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/StandardWebSocketClient.java @@ -115,15 +115,14 @@ public class StandardWebSocketClient implements WebSocketClient { private StandardWebSocketHandlerAdapter createEndpoint(URI url, WebSocketHandler handler, MonoProcessor completion, DefaultConfigurator configurator) { - return new StandardWebSocketHandlerAdapter(handler, session -> { - if (logger.isDebugEnabled()) { - logger.debug("Connected to " + url); - } - HttpHeaders responseHeaders = configurator.getResponseHeaders(); - String protocol = responseHeaders.getFirst("Sec-WebSocket-Protocol"); - HandshakeInfo info = new HandshakeInfo(url, responseHeaders, Mono.empty(), protocol); - return createWebSocketSession(session, info, completion); - }); + return new StandardWebSocketHandlerAdapter(handler, session -> + createWebSocketSession(session, createHandshakeInfo(url, configurator), completion)); + } + + private HandshakeInfo createHandshakeInfo(URI url, DefaultConfigurator configurator) { + HttpHeaders responseHeaders = configurator.getResponseHeaders(); + String protocol = responseHeaders.getFirst("Sec-WebSocket-Protocol"); + return new HandshakeInfo(url, responseHeaders, Mono.empty(), protocol); } protected StandardWebSocketSession createWebSocketSession(Session session, HandshakeInfo info, diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/UndertowWebSocketClient.java b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/UndertowWebSocketClient.java index f9ca84d532..a020677325 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/UndertowWebSocketClient.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/UndertowWebSocketClient.java @@ -196,12 +196,7 @@ public class UndertowWebSocketClient implements WebSocketClient { private void handleChannel(URI url, WebSocketHandler handler, MonoProcessor completion, DefaultNegotiation negotiation, WebSocketChannel channel) { - if (logger.isDebugEnabled()) { - logger.debug("Connected to " + url); - } - HttpHeaders responseHeaders = negotiation.getResponseHeaders(); - String protocol = responseHeaders.getFirst("Sec-WebSocket-Protocol"); - HandshakeInfo info = new HandshakeInfo(url, responseHeaders, Mono.empty(), protocol); + HandshakeInfo info = createHandshakeInfo(url, negotiation); UndertowWebSocketSession session = new UndertowWebSocketSession(channel, info, this.bufferFactory, completion); UndertowWebSocketHandlerAdapter adapter = new UndertowWebSocketHandlerAdapter(session); @@ -211,6 +206,12 @@ public class UndertowWebSocketClient implements WebSocketClient { handler.handle(session).subscribe(session); } + private HandshakeInfo createHandshakeInfo(URI url, DefaultNegotiation negotiation) { + HttpHeaders responseHeaders = negotiation.getResponseHeaders(); + String protocol = responseHeaders.getFirst("Sec-WebSocket-Protocol"); + return new HandshakeInfo(url, responseHeaders, Mono.empty(), protocol); + } + private static final class DefaultNegotiation extends WebSocketClientNegotiation { diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/server/support/HandshakeWebSocketService.java b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/server/support/HandshakeWebSocketService.java index 354536e237..9aba095bf4 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/socket/server/support/HandshakeWebSocketService.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/socket/server/support/HandshakeWebSocketService.java @@ -16,6 +16,7 @@ package org.springframework.web.reactive.socket.server.support; +import java.net.URI; import java.security.Principal; import java.util.Collections; import java.util.List; @@ -269,8 +270,11 @@ public class HandshakeWebSocketService implements WebSocketService, Lifecycle { private HandshakeInfo createHandshakeInfo(ServerWebExchange exchange, ServerHttpRequest request, @Nullable String protocol, Map attributes) { + URI uri = request.getURI(); + HttpHeaders headers = request.getHeaders(); Mono principal = exchange.getPrincipal(); - return new HandshakeInfo(request.getURI(), request.getHeaders(), principal, protocol, attributes); + String logPrefix = exchange.getLogPrefix(); + return new HandshakeInfo(uri, headers, principal, protocol, attributes, logPrefix); } }