Polish and improve logging

spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerReadPublisher.java

@@ -69,9 +69,6 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 
 	@Override
 	public void subscribe(Subscriber<? super T> subscriber) {
-		if (this.logger.isTraceEnabled()) {
-			this.logger.trace(this.state + " subscribe: " + subscriber);
-		}
 		this.state.get().subscribe(this, subscriber);
 	}
 
@@ -79,24 +76,20 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 	// Methods for sub-classes to delegate to, when async I/O events occur...
 
 	public final void onDataAvailable() {
-		if (this.logger.isTraceEnabled()) {
+		this.logger.trace("I/O event onDataAvailable");
-			this.logger.trace(this.state + " onDataAvailable");
-		}
 		this.state.get().onDataAvailable(this);
 	}
 
 	public void onAllDataRead() {
-		if (this.logger.isTraceEnabled()) {
+		this.logger.trace("I/O event onAllDataRead");
-			this.logger.trace(this.state + " onAllDataRead");
-		}
 		this.state.get().onAllDataRead(this);
 	}
 
-	public final void onError(Throwable t) {
+	public final void onError(Throwable ex) {
 		if (this.logger.isTraceEnabled()) {
-			this.logger.trace(this.state + " onError: " + t);
+			this.logger.trace("I/O event onError: " + ex);
 		}
-		this.state.get().onError(this, t);
+		this.state.get().onError(this, ex);
 	}
 
 
@@ -142,10 +135,17 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 				if (r != Long.MAX_VALUE) {
 					DEMAND_FIELD_UPDATER.addAndGet(this, -1L);
 				}
-				Assert.state(this.subscriber != null, "No subscriber");
+				Subscriber<? super T> subscriber = this.subscriber;
-				this.subscriber.onNext(data);
+				Assert.state(subscriber != null, "No subscriber");
+				if (logger.isTraceEnabled()) {
+					logger.trace("Data item read, publishing..");
+				}
+				subscriber.onNext(data);
 			}
 			else {
+				if (logger.isTraceEnabled()) {
+					logger.trace("No more data to read");
+				}
 				return true;
 			}
 		}
@@ -153,7 +153,17 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 	}
 
 	private boolean changeState(State oldState, State newState) {
-		return this.state.compareAndSet(oldState, newState);
+		boolean result = this.state.compareAndSet(oldState, newState);
+		if (result && logger.isTraceEnabled()) {
+			logger.trace(oldState + " -> " + newState);
+		}
+		return result;
+	}
+
+	private void changeToDemandState(State oldState) {
+		if (changeState(oldState, State.DEMAND)) {
+			checkOnDataAvailable();
+		}
 	}
 
 	private Subscription createSubscription() {
@@ -170,7 +180,7 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 		@Override
 		public final void request(long n) {
 			if (logger.isTraceEnabled()) {
-				logger.trace(state + " request: " + n);
+				logger.trace("Signal request(" + n + ")");
 			}
 			state.get().request(AbstractListenerReadPublisher.this, n);
 		}
@@ -178,7 +188,7 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 		@Override
 		public final void cancel() {
 			if (logger.isTraceEnabled()) {
-				logger.trace(state + " cancel");
+				logger.trace("Signal cancel()");
 			}
 			state.get().cancel(AbstractListenerReadPublisher.this);
 		}
@@ -217,10 +227,14 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 					publisher.changeState(SUBSCRIBING, NO_DEMAND);
 					// Now safe to check "beforeDemand" flags, they won't change once in NO_DEMAND
 					if (publisher.completionBeforeDemand) {
+						publisher.logger.trace("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.state.get().onError(publisher, ex);
 					}
 				}
@@ -249,9 +263,7 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 			<T> void request(AbstractListenerReadPublisher<T> publisher, long n) {
 				if (Operators.validate(n)) {
 					Operators.addCap(DEMAND_FIELD_UPDATER, publisher, n);
-					if (publisher.changeState(this, DEMAND)) {
+					publisher.changeToDemandState(this);
-						publisher.checkOnDataAvailable();
-					}
 				}
 			}
 
@@ -271,10 +283,7 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 			<T> void request(AbstractListenerReadPublisher<T> publisher, long n) {
 				if (Operators.validate(n)) {
 					Operators.addCap(DEMAND_FIELD_UPDATER, publisher, n);
-					if (publisher.changeState(this, DEMAND)) {
+					publisher.changeToDemandState(this);
-						publisher.checkOnDataAvailable();
-					}
-					// or else we completed at the same time...
 				}
 			}
 		},
@@ -285,9 +294,7 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 				if (Operators.validate(n)) {
 					Operators.addCap(DEMAND_FIELD_UPDATER, publisher, n);
 					// Did a concurrent read transition to NO_DEMAND just before us?
-					if (publisher.changeState(NO_DEMAND, this)) {
+					publisher.changeToDemandState(NO_DEMAND);
-						publisher.checkOnDataAvailable();
-					}
 				}
 			}
 
@@ -297,17 +304,15 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 					try {
 						boolean demandAvailable = publisher.readAndPublish();
 						if (demandAvailable) {
-							if (publisher.changeState(READING, DEMAND)) {
+							publisher.changeToDemandState(READING);
-								publisher.checkOnDataAvailable();
-							}
 						}
 						else {
 							publisher.readingPaused();
 							if (publisher.changeState(READING, NO_DEMAND)) {
 								// Demand may have arrived since readAndPublish returned
 								long r = publisher.demand;
-								if (r > 0  && publisher.changeState(NO_DEMAND, this)) {
+								if (r > 0) {
-									publisher.checkOnDataAvailable();
+									publisher.changeToDemandState(NO_DEMAND);
 								}
 							}
 						}
@@ -326,9 +331,7 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 				if (Operators.validate(n)) {
 					Operators.addCap(DEMAND_FIELD_UPDATER, publisher, n);
 					// Did a concurrent read transition to NO_DEMAND just before us?
-					if (publisher.changeState(NO_DEMAND, DEMAND)) {
+					publisher.changeToDemandState(NO_DEMAND);
-						publisher.checkOnDataAvailable();
-					}
 				}
 			}
 		},
@@ -372,8 +375,9 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 
 		<T> void onAllDataRead(AbstractListenerReadPublisher<T> publisher) {
 			if (publisher.changeState(this, COMPLETED)) {
-				if (publisher.subscriber != null) {
+				Subscriber<? super T> s = publisher.subscriber;
-					publisher.subscriber.onComplete();
+				if (s != null) {
+					s.onComplete();
 				}
 			}
 			else {
@@ -383,8 +387,9 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 
 		<T> void onError(AbstractListenerReadPublisher<T> publisher, Throwable t) {
 			if (publisher.changeState(this, COMPLETED)) {
-				if (publisher.subscriber != null) {
+				Subscriber<? super T> s = publisher.subscriber;
-					publisher.subscriber.onError(t);
+				if (s != null) {
+					s.onError(t);
 				}
 			}
 			else {

spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java

@@ -53,40 +53,47 @@ public abstract class AbstractListenerWriteFlushProcessor<T> implements Processo
 	private final WriteResultPublisher resultPublisher = new WriteResultPublisher();
 
 
-	// Subscriber methods...
+	// Subscriber methods and methods to notify of async I/O events...
 
 	@Override
 	public final void onSubscribe(Subscription subscription) {
-		if (logger.isTraceEnabled()) {
-			logger.trace(this.state + " onSubscribe: " + subscription);
-		}
 		this.state.get().onSubscribe(this, subscription);
 	}
 
 	@Override
 	public final void onNext(Publisher<? extends T> publisher) {
-		if (logger.isTraceEnabled()) {
+		logger.trace("Received onNext publisher");
-			logger.trace(this.state + " onNext: " + publisher);
-		}
 		this.state.get().onNext(this, publisher);
 	}
 
+	/**
+	 * Notify of an error. This can come from the upstream write Publisher or
+	 * from sub-classes as a result of an I/O error.
+	 */
 	@Override
-	public final void onError(Throwable t) {
+	public final void onError(Throwable ex) {
 		if (logger.isTraceEnabled()) {
-			logger.trace(this.state + " onError: " + t);
+			logger.trace("Received onError: " + ex);
 		}
-		this.state.get().onError(this, t);
+		this.state.get().onError(this, ex);
 	}
 
+	/**
+	 * Notify of completion. This can come from the upstream write Publisher or
+	 * from sub-classes as a result of an I/O completion event.
+	 */
 	@Override
 	public final void onComplete() {
-		if (logger.isTraceEnabled()) {
+		logger.trace("Received onComplete");
-			logger.trace(this.state + " onComplete");
-		}
 		this.state.get().onComplete(this);
 	}
 
+	protected void cancel() {
+		this.logger.trace("Received request to cancel");
+		if (this.subscription != null) {
+			this.subscription.cancel();
+		}
+	}
 
 	// Publisher method...
 
@@ -96,15 +103,6 @@ public abstract class AbstractListenerWriteFlushProcessor<T> implements Processo
 	}
 
 
-	// Methods for sub-classes to delegate to, when async I/O events occur...
-
-	protected void cancel() {
-		if (this.subscription != null) {
-			this.subscription.cancel();
-		}
-	}
-
-
 	// Methods for sub-classes to implement or override...
 
 	/**
@@ -147,11 +145,19 @@ public abstract class AbstractListenerWriteFlushProcessor<T> implements Processo
 	// Private methods for use in State...
 
 	private boolean changeState(State oldState, State newState) {
-		return this.state.compareAndSet(oldState, newState);
+		boolean result = this.state.compareAndSet(oldState, newState);
+		if (result && logger.isTraceEnabled()) {
+			logger.trace(oldState + " -> " + newState);
+		}
+		return result;
 	}
 
 	private void flushIfPossible() {
-		if (isWritePossible()) {
+		boolean result = isWritePossible();
+		if (logger.isTraceEnabled()) {
+			logger.trace("isWritePossible[" + result + "]");
+		}
+		if (result) {
 			onFlushPossible();
 		}
 	}

spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java

@@ -58,61 +58,60 @@ public abstract class AbstractListenerWriteProcessor<T> implements Processor<T,
 	private final WriteResultPublisher resultPublisher = new WriteResultPublisher();
 
 
-	// Subscriber methods...
+	// Subscriber methods and methods to notify of async I/O events...
 
 	@Override
 	public final void onSubscribe(Subscription subscription) {
-		if (logger.isTraceEnabled()) {
-			logger.trace(this.state + " onSubscribe: " + subscription);
-		}
 		this.state.get().onSubscribe(this, subscription);
 	}
 
 	@Override
 	public final void onNext(T data) {
-		if (logger.isTraceEnabled()) {
+		logger.trace("Received onNext data item");
-			logger.trace(this.state + " onNext: " + data);
-		}
 		this.state.get().onNext(this, data);
 	}
 
+	/**
+	 * Notify of an error. This can come from the upstream write Publisher or
+	 * from sub-classes as a result of an I/O error.
+	 */
 	@Override
-	public final void onError(Throwable t) {
+	public final void onError(Throwable ex) {
 		if (logger.isTraceEnabled()) {
-			logger.trace(this.state + " onError: " + t);
+			logger.trace("Received onError: " + ex);
 		}
-		this.state.get().onError(this, t);
+		this.state.get().onError(this, ex);
 	}
 
+	/**
+	 * Notify of completion. This can come from the upstream write Publisher or
+	 * from sub-classes as a result of an I/O completion event.
+	 */
 	@Override
 	public final void onComplete() {
-		if (logger.isTraceEnabled()) {
+		logger.trace("Received onComplete");
-			logger.trace(this.state + " onComplete");
-		}
 		this.state.get().onComplete(this);
 	}
 
-
-	// Publisher method...
-
-	@Override
-	public final void subscribe(Subscriber<? super Void> subscriber) {
-		this.resultPublisher.subscribe(subscriber);
-	}
-
-
-	// Methods for sub-classes to delegate to, when async I/O events occur...
-
 	public final void onWritePossible() {
+		this.logger.trace("Received onWritePossible");
 		this.state.get().onWritePossible(this);
 	}
 
 	public void cancel() {
+		this.logger.trace("Received request to cancel");
 		if (this.subscription != null) {
 			this.subscription.cancel();
 		}
 	}
 
+	// Publisher method...
+
+	@Override
+	public final void subscribe(Subscriber<? super Void> subscriber) {
+		this.resultPublisher.subscribe(subscriber);
+	}
+
 
 	// Methods for sub-classes to implement or override...
 
@@ -175,7 +174,11 @@ public abstract class AbstractListenerWriteProcessor<T> implements Processor<T,
 	// Private methods for use in State...
 
 	private boolean changeState(State oldState, State newState) {
-		return this.state.compareAndSet(oldState, newState);
+		boolean result = this.state.compareAndSet(oldState, newState);
+		if (result && logger.isTraceEnabled()) {
+			logger.trace(oldState + " -> " + newState);
+		}
+		return result;
 	}
 
 	private void changeStateToComplete(State oldState) {
@@ -189,7 +192,11 @@ public abstract class AbstractListenerWriteProcessor<T> implements Processor<T,
 	}
 
 	private void writeIfPossible() {
-		if (isWritePossible()) {
+		boolean result = isWritePossible();
+		if (logger.isTraceEnabled()) {
+			logger.trace("isWritePossible[" + result + "]");
+		}
+		if (result) {
 			onWritePossible();
 		}
 	}

spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpRequest.java

@@ -195,7 +195,7 @@ class ServletServerHttpRequest extends AbstractServerHttpRequest {
 	protected DataBuffer readFromInputStream() throws IOException {
 		int read = this.request.getInputStream().read(this.buffer);
 		if (logger.isTraceEnabled()) {
-			logger.trace("read:" + read);
+			logger.trace("InputStream read returned " + read + (read != -1 ? " bytes" : ""));
 		}
 
 		if (read > 0) {

spring-web/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpRequest.java

@@ -170,7 +170,7 @@ class UndertowServerHttpRequest extends AbstractServerHttpRequest {
 
 				int read = this.channel.read(byteBuffer);
 				if (logger.isTraceEnabled()) {
-					logger.trace("read:" + read);
+					logger.trace("Channel read returned " + read + (read != -1 ? " bytes" : ""));
 				}
 
 				if (read > 0) {

spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/AbstractListenerWebSocketSession.java

@@ -224,8 +224,12 @@ public abstract class AbstractListenerWebSocketSession<T> extends AbstractWebSoc
 		protected void checkOnDataAvailable() {
 			resumeReceiving();
 			if (!this.pendingMessages.isEmpty()) {
+				logger.trace("checkOnDataAvailable, processing pending messages");
 				onDataAvailable();
 			}
+			else {
+				logger.trace("checkOnDataAvailable, no pending messages");
+			}
 		}
 
 		@Override
@@ -239,8 +243,11 @@ public abstract class AbstractListenerWebSocketSession<T> extends AbstractWebSoc
 			return (WebSocketMessage) this.pendingMessages.poll();
 		}
 
-		void handleMessage(WebSocketMessage webSocketMessage) {
+		void handleMessage(WebSocketMessage message) {
-			if (!this.pendingMessages.offer(webSocketMessage)) {
+			if (logger.isTraceEnabled()) {
+				logger.trace("Received message: " + message);
+			}
+			if (!this.pendingMessages.offer(message)) {
 				throw new IllegalStateException("Too many messages received. " +
 						"Please ensure WebSocketSession.receive() is subscribed to.");
 			}
@@ -255,6 +262,9 @@ public abstract class AbstractListenerWebSocketSession<T> extends AbstractWebSoc
 
 		@Override
 		protected boolean write(WebSocketMessage message) throws IOException {
+			if (logger.isTraceEnabled()) {
+				logger.trace("Sending message " + message);
+			}
 			return sendMessage(message);
 		}
 
@@ -279,6 +289,9 @@ public abstract class AbstractListenerWebSocketSession<T> extends AbstractWebSoc
 		 * async completion callback into simple flow control.
 		 */
 		public void setReadyToSend(boolean ready) {
+			if (ready) {
+				logger.trace("Send succeeded, ready to send again");
+			}
 			this.isReady = ready;
 		}
 	}