Fix new Sonar smells

Author: artembilan

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.rsocket;
 
+import java.lang.reflect.Method;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -38,7 +39,6 @@
 import org.springframework.util.RouteMatcher;
 
 import io.rsocket.RSocketFactory;
-import io.rsocket.SocketAcceptor;
 import io.rsocket.transport.ServerTransport;
 import io.rsocket.transport.netty.server.CloseableChannel;
 import io.rsocket.transport.netty.server.TcpServerTransport;
@@ -50,7 +50,7 @@
 /**
  * A server {@link AbstractRSocketConnector} extension to accept and manage client RSocket connections.
  * <p>
- * Note: the {@link RSocketFactory.ServerRSocketFactory#acceptor(SocketAcceptor)}
+ * Note: the {@link RSocketFactory.ServerRSocketFactory#acceptor(io.rsocket.SocketAcceptor)}
  * in the provided {@link #factoryConfigurer} is overridden with an internal
  * {@link ServerRSocketMessageHandler#serverAcceptor()}
  * for the proper Spring Integration channel adapter mappings.
@@ -175,16 +175,18 @@ public void afterSingletonsInstantiated() {
 
 	private static class ServerRSocketMessageHandler extends IntegrationRSocketMessageHandler {
 
+		private static final Method HANDLE_CONNECTION_SETUP_METHOD =
+				ReflectionUtils.findMethod(ServerRSocketMessageHandler.class, "handleConnectionSetup", Message.class);
+
+
 		private final Map<Object, RSocketRequester> clientRSocketRequesters = new HashMap<>();
 
 		private BiFunction<String, DataBuffer, Object> clientRSocketKeyStrategy = (destination, data) -> destination;
 
 		private ApplicationEventPublisher applicationEventPublisher;
 
 		private void registerHandleConnectionSetupMethod() {
-			registerHandlerMethod(this,
-					ReflectionUtils.findMethod(ServerRSocketMessageHandler.class, "handleConnectionSetup", // NOSONAR
-							Message.class),
+			registerHandlerMethod(this, HANDLE_CONNECTION_SETUP_METHOD,
 					new CompositeMessageCondition(
 							RSocketFrameTypeMessageCondition.CONNECT_CONDITION,
 							new DestinationPatternsMessageCondition(new String[] { "*" }, getRouteMatcher())));

spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/ChannelSendOperator.java

@@ -157,7 +157,7 @@ private class WriteBarrier implements CoreSubscriber<T>, Subscription, Publisher
 		private long demandBeforeReadyToWrite;
 
 		/** Current state. */
-		private State state = State.NEW;
+		private volatile State state = State.NEW;
 
 		/** The actual writeSubscriber from the HTTP server adapter. */
 		@Nullable
@@ -198,7 +198,7 @@ else if (this.state == State.NEW) {
 					try {
 						result = ChannelSendOperator.this.writeFunction.apply(this);
 					}
-					catch (Throwable ex) {
+					catch (Throwable ex) { // NOSONAR
 						this.writeCompletionBarrier.onError(ex);
 						return;
 					}
@@ -214,8 +214,9 @@ else if (this.state == State.NEW) {
 		}
 
 		private Subscriber<? super T> requiredWriteSubscriber() {
-			Assert.state(this.writeSubscriber != null, "No write subscriber");
-			return this.writeSubscriber;
+			Subscriber<? super T> writeSubscriberToReturn = this.writeSubscriber;
+			Assert.state(writeSubscriberToReturn != null, "No write subscriber");
+			return writeSubscriberToReturn;
 		}
 
 		@Override
@@ -255,7 +256,7 @@ else if (this.state == State.NEW) {
 					try {
 						result = ChannelSendOperator.this.writeFunction.apply(this);
 					}
-					catch (Throwable ex) {
+					catch (Throwable ex) { // NOSONAR
 						this.writeCompletionBarrier.onError(ex);
 						return;
 					}
@@ -277,27 +278,28 @@ public Context currentContext() {
 
 		@Override
 		public void request(long n) {
+			long requests = n;
 			Subscription s = this.subscription;
 			if (s == null) {
 				return;
 			}
 			if (this.state == State.READY_TO_WRITE) {
-				s.request(n);
+				s.request(requests);
 				return;
 			}
 			synchronized (this) {
 				if (this.writeSubscriber != null) {
 					if (this.state == State.EMITTING_CACHED_SIGNALS) {
-						this.demandBeforeReadyToWrite = n;
+						this.demandBeforeReadyToWrite = requests;
 						return;
 					}
 					try {
 						this.state = State.EMITTING_CACHED_SIGNALS;
 						if (emitCachedSignals()) {
 							return;
 						}
-						n = n + this.demandBeforeReadyToWrite - 1;
-						if (n == 0) {
+						requests = requests + this.demandBeforeReadyToWrite - 1;
+						if (requests == 0) {
 							return;
 						}
 					}
@@ -306,7 +308,7 @@ public void request(long n) {
 					}
 				}
 			}
-			s.request(n);
+			s.request(requests);
 		}
 
 		private boolean emitCachedSignals() {
@@ -319,10 +321,13 @@ private boolean emitCachedSignals() {
 				}
 				return true;
 			}
-			T item = this.item;
-			this.item = null;
-			if (item != null) {
-				requiredWriteSubscriber().onNext(item);
+			T itemToUse;
+			synchronized (this) {
+				itemToUse = this.item;
+				this.item = null;
+			}
+			if (itemToUse != null) {
+				requiredWriteSubscriber().onNext(itemToUse);
 			}
 			if (this.completed) {
 				requiredWriteSubscriber().onComplete();
@@ -347,9 +352,9 @@ public void cancel() {
 
 		private void releaseCachedItem() {
 			synchronized (this) {
-				Object item = this.item;
-				if (item instanceof DataBuffer) {
-					DataBufferUtils.release((DataBuffer) item);
+				Object itemToRelease = this.item;
+				if (itemToRelease instanceof DataBuffer) {
+					DataBufferUtils.release((DataBuffer) itemToRelease);
 				}
 				this.item = null;
 			}
@@ -451,9 +456,9 @@ public void request(long n) {
 		@Override
 		public void cancel() {
 			this.writeBarrier.cancel();
-			Subscription subscription = this.subscription;
-			if (subscription != null) {
-				subscription.cancel();
+			Subscription subscriptionToCancel = this.subscription;
+			if (subscriptionToCancel != null) {
+				subscriptionToCancel.cancel();
 			}
 		}