Avoid invoking writeFunction under lock in ChannelSendOperator

Signed-off-by: l2yuPa <jeungwon28@gmail.com>

Author: l2yujw

spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/reactive/ChannelSendOperator.java

@@ -176,31 +176,34 @@ public final void onNext(T item) {
 				requiredWriteSubscriber().onNext(item);
 				return;
 			}
-			// FIXME revisit in case of reentrant sync deadlock
+
 			synchronized (this) {
 				if (this.state == State.READY_TO_WRITE) {
 					requiredWriteSubscriber().onNext(item);
+					return;
 				}
-				else if (this.state == State.NEW) {
-					this.item = item;
-					this.state = State.FIRST_SIGNAL_RECEIVED;
-					Publisher<Void> result;
-					try {
-						result = writeFunction.apply(this);
-					}
-					catch (Throwable ex) {
-						this.writeCompletionBarrier.onError(ex);
-						return;
-					}
-					result.subscribe(this.writeCompletionBarrier);
-				}
-				else {
+
+				if (this.state != State.NEW) {
 					if (this.subscription != null) {
 						this.subscription.cancel();
 					}
 					this.writeCompletionBarrier.onError(new IllegalStateException("Unexpected item."));
+					return;
 				}
+
+				this.item = item;
+				this.state = State.FIRST_SIGNAL_RECEIVED;
 			}
+
+			Publisher<Void> result;
+			try {
+				result = writeFunction.apply(this);
+			}
+			catch (Throwable ex) {
+				this.writeCompletionBarrier.onError(ex);
+				return;
+			}
+			result.subscribe(this.writeCompletionBarrier);
 		}
 
 		private Subscriber<? super T> requiredWriteSubscriber() {
@@ -234,27 +237,31 @@ public final void onComplete() {
 				requiredWriteSubscriber().onComplete();
 				return;
 			}
+
 			synchronized (this) {
 				if (this.state == State.READY_TO_WRITE) {
 					requiredWriteSubscriber().onComplete();
+					return;
 				}
-				else if (this.state == State.NEW) {
-					this.completed = true;
-					this.state = State.FIRST_SIGNAL_RECEIVED;
-					Publisher<Void> result;
-					try {
-						result = writeFunction.apply(this);
-					}
-					catch (Throwable ex) {
-						this.writeCompletionBarrier.onError(ex);
-						return;
-					}
-					result.subscribe(this.writeCompletionBarrier);
-				}
-				else {
+
+				if (this.state != State.NEW) {
 					this.completed = true;
+					return;
 				}
+
+				this.completed = true;
+				this.state = State.FIRST_SIGNAL_RECEIVED;
+			}
+
+			Publisher<Void> result;
+			try {
+				result = writeFunction.apply(this);
+			}
+			catch (Throwable ex) {
+				this.writeCompletionBarrier.onError(ex);
+				return;
 			}
+			result.subscribe(this.writeCompletionBarrier);
 		}
 
 		@Override

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

@@ -168,31 +168,34 @@ public final void onNext(T item) {
 				requiredWriteSubscriber().onNext(item);
 				return;
 			}
-			// FIXME revisit in case of reentrant sync deadlock
+
 			synchronized (this) {
 				if (this.state == State.READY_TO_WRITE) {
 					requiredWriteSubscriber().onNext(item);
+					return;
 				}
-				else if (this.state == State.NEW) {
-					this.item = item;
-					this.state = State.FIRST_SIGNAL_RECEIVED;
-					Publisher<Void> result;
-					try {
-						result = writeFunction.apply(this);
-					}
-					catch (Throwable ex) {
-						this.writeCompletionBarrier.onError(ex);
-						return;
-					}
-					result.subscribe(this.writeCompletionBarrier);
-				}
-				else {
+
+				if (this.state != State.NEW) {
 					if (this.subscription != null) {
 						this.subscription.cancel();
 					}
 					this.writeCompletionBarrier.onError(new IllegalStateException("Unexpected item."));
+					return;
 				}
+
+				this.item = item;
+				this.state = State.FIRST_SIGNAL_RECEIVED;
 			}
+
+			Publisher<Void> result;
+			try {
+				result = writeFunction.apply(this);
+			}
+			catch (Throwable ex) {
+				this.writeCompletionBarrier.onError(ex);
+				return;
+			}
+			result.subscribe(this.writeCompletionBarrier);
 		}
 
 		private Subscriber<? super T> requiredWriteSubscriber() {
@@ -226,27 +229,31 @@ public final void onComplete() {
 				requiredWriteSubscriber().onComplete();
 				return;
 			}
+
 			synchronized (this) {
 				if (this.state == State.READY_TO_WRITE) {
 					requiredWriteSubscriber().onComplete();
+					return;
 				}
-				else if (this.state == State.NEW) {
-					this.completed = true;
-					this.state = State.FIRST_SIGNAL_RECEIVED;
-					Publisher<Void> result;
-					try {
-						result = writeFunction.apply(this);
-					}
-					catch (Throwable ex) {
-						this.writeCompletionBarrier.onError(ex);
-						return;
-					}
-					result.subscribe(this.writeCompletionBarrier);
-				}
-				else {
+
+				if (this.state != State.NEW) {
 					this.completed = true;
+					return;
 				}
+
+				this.completed = true;
+				this.state = State.FIRST_SIGNAL_RECEIVED;
+			}
+
+			Publisher<Void> result;
+			try {
+				result = writeFunction.apply(this);
+			}
+			catch (Throwable ex) {
+				this.writeCompletionBarrier.onError(ex);
+				return;
 			}
+			result.subscribe(this.writeCompletionBarrier);
 		}
 
 		@Override