Avoid invoking writeFunction under lock in ChannelSendOperator

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

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

@@ -176,7 +176,9 @@ class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
 				requiredWriteSubscriber().onNext(item);
 				return;
 			}
-			// FIXME revisit in case of reentrant sync deadlock
+
+			boolean invokeWriteFunction = false;
+
 			synchronized (this) {
 				if (this.state == State.READY_TO_WRITE) {
 					requiredWriteSubscriber().onNext(item);
@@ -184,15 +186,7 @@ class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
 				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);
+					invokeWriteFunction = true;
 				}
 				else {
 					if (this.subscription != null) {
@@ -201,6 +195,20 @@ class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
 					this.writeCompletionBarrier.onError(new IllegalStateException("Unexpected item."));
 				}
 			}
+
+			if (!invokeWriteFunction) {
+				return;
+			}
+
+			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,6 +242,9 @@ class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
 				requiredWriteSubscriber().onComplete();
 				return;
 			}
+
+			boolean invokeWriteFunction = false;
+
 			synchronized (this) {
 				if (this.state == State.READY_TO_WRITE) {
 					requiredWriteSubscriber().onComplete();
@@ -241,20 +252,26 @@ class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
 				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);
+					invokeWriteFunction = true;
 				}
 				else {
 					this.completed = true;
 				}
 			}
+
+			if (!invokeWriteFunction) {
+				return;
+			}
+
+			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,7 +168,9 @@ public class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
 				requiredWriteSubscriber().onNext(item);
 				return;
 			}
-			// FIXME revisit in case of reentrant sync deadlock
+
+			boolean invokeWriteFunction = false;
+
 			synchronized (this) {
 				if (this.state == State.READY_TO_WRITE) {
 					requiredWriteSubscriber().onNext(item);
@@ -176,15 +178,7 @@ public class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
 				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);
+					invokeWriteFunction = true;
 				}
 				else {
 					if (this.subscription != null) {
@@ -193,6 +187,20 @@ public class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
 					this.writeCompletionBarrier.onError(new IllegalStateException("Unexpected item."));
 				}
 			}
+
+			if (!invokeWriteFunction) {
+				return;
+			}
+
+			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,6 +234,9 @@ public class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
 				requiredWriteSubscriber().onComplete();
 				return;
 			}
+
+			boolean invokeWriteFunction = false;
+
 			synchronized (this) {
 				if (this.state == State.READY_TO_WRITE) {
 					requiredWriteSubscriber().onComplete();
@@ -233,20 +244,26 @@ public class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
 				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);
+					invokeWriteFunction = true;
 				}
 				else {
 					this.completed = true;
 				}
 			}
+
+			if (!invokeWriteFunction) {
+				return;
+			}
+
+			Publisher<Void> result;
+			try {
+				result = writeFunction.apply(this);
+			}
+			catch (Throwable ex) {
+				this.writeCompletionBarrier.onError(ex);
+				return;
+			}
+			result.subscribe(this.writeCompletionBarrier);
 		}
 
 		@Override