GH-10083: Apply Nullability to STMOP module

anthologia · web-flow · commit 8fdee30f54ab · 2025-08-15T09:30:32.000-04:00
Related to: #10083 Signed-off-by: Jooyoung Pyoung <pyoungjy@gmail.com>
diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/AbstractStompSessionManager.java
@@ -31,6 +31,7 @@
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.DisposableBean;
@@ -39,7 +40,6 @@
 import org.springframework.context.SmartLifecycle;
 import org.springframework.integration.stomp.event.StompConnectionFailedEvent;
 import org.springframework.integration.stomp.event.StompSessionConnectedEvent;
-import org.springframework.lang.Nullable;
 import org.springframework.messaging.simp.stomp.StompClientSupport;
 import org.springframework.messaging.simp.stomp.StompCommand;
 import org.springframework.messaging.simp.stomp.StompHeaders;
@@ -94,23 +94,24 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
 
 	private int phase = Integer.MAX_VALUE / 2;
 
+	@SuppressWarnings("NullAway.Init")
 	private ApplicationEventPublisher applicationEventPublisher;
 
-	private StompHeaders connectHeaders;
+	private @Nullable StompHeaders connectHeaders;
 
 	private boolean autoReceipt;
 
 	private long recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
 
-	private String name;
+	private @Nullable String name;
 
 	private volatile boolean connecting;
 
 	private volatile boolean connected;
 
-	private volatile CompletableFuture<StompSession> stompSessionFuture;
+	private volatile @Nullable CompletableFuture<StompSession> stompSessionFuture;
 
-	private volatile ScheduledFuture<?> reconnectFuture;
+	private volatile @Nullable ScheduledFuture<?> reconnectFuture;
 
 	public AbstractStompSessionManager(StompClientSupport stompClient) {
 		Assert.notNull(stompClient, "'stompClient' is required.");
@@ -226,6 +227,8 @@ private void connect() {
 	}
 
 	private CountDownLatch addStompSessionCallback(int currentEpoch) {
+		Assert.notNull(this.stompSessionFuture, "'stompSessionFuture' is required.");
+
 		CountDownLatch connectLatch = new CountDownLatch(1);
 		this.stompSessionFuture.whenComplete((stompSession, throwable) -> {
 					if (throwable == null) {
@@ -254,7 +257,7 @@ private CountDownLatch addStompSessionCallback(int currentEpoch) {
 		return connectLatch;
 	}
 
-	private void scheduleReconnect(Throwable e) {
+	private void scheduleReconnect(@Nullable Throwable e) {
 		this.epoch.incrementAndGet();
 		this.connecting = false;
 		this.connected = false;
@@ -353,7 +356,7 @@ public void disconnect(StompSessionHandler handler) {
 		this.compositeStompSessionHandler.removeHandler(handler);
 	}
 
-	protected StompHeaders getConnectHeaders() {
+	protected @Nullable StompHeaders getConnectHeaders() {
 		return this.connectHeaders;
 	}
 
@@ -374,7 +377,7 @@ private class CompositeStompSessionHandler extends StompSessionHandlerAdapter {
 
 		private final Lock delegatesMonitor = new ReentrantLock();
 
-		private volatile StompSession session;
+		private volatile @Nullable StompSession session;
 
 		CompositeStompSessionHandler() {
 		}
@@ -383,7 +386,11 @@ void addHandler(StompSessionHandler delegate) {
 			this.delegatesMonitor.lock();
 			try {
 				if (this.session != null) {
-					delegate.afterConnected(this.session, getConnectHeaders());
+					StompHeaders headers = getConnectHeaders();
+					if (headers == null) {
+						headers = new StompHeaders();
+					}
+					delegate.afterConnected(this.session, headers);
 				}
 				this.delegates.add(delegate);
 			}
@@ -443,7 +450,7 @@ public void handleTransportError(StompSession session, Throwable exception) {
 		}
 
 		@Override
-		public void handleFrame(StompHeaders headers, Object payload) {
+		public void handleFrame(StompHeaders headers, @Nullable Object payload) {
 			this.delegatesMonitor.lock();
 			try {
 				for (StompSessionHandler delegate : this.delegates) {
diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/WebSocketStompSessionManager.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/WebSocketStompSessionManager.java
@@ -19,6 +19,8 @@
 import java.util.Arrays;
 import java.util.concurrent.CompletableFuture;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.messaging.simp.stomp.StompSession;
 import org.springframework.messaging.simp.stomp.StompSessionHandler;
 import org.springframework.util.Assert;
@@ -41,13 +43,14 @@ public class WebSocketStompSessionManager extends AbstractStompSessionManager {
 
 	private final Object[] uriVariables;
 
-	private volatile WebSocketHttpHeaders handshakeHeaders;
+	private volatile @Nullable WebSocketHttpHeaders handshakeHeaders;
 
 	public WebSocketStompSessionManager(WebSocketStompClient webSocketStompClient, String url, Object... uriVariables) {
 		super(webSocketStompClient);
 		Assert.hasText(url, "'url' must not be empty.");
 		this.url = url;
-		this.uriVariables = uriVariables != null ? Arrays.copyOf(uriVariables, uriVariables.length) : null;
+		this.uriVariables = uriVariables != null ?
+				Arrays.copyOf(uriVariables, uriVariables.length) : new Object[0];
 	}
 
 	public void setHandshakeHeaders(WebSocketHttpHeaders handshakeHeaders) {
diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/package-info.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/config/package-info.java
@@ -1,4 +1,5 @@
 /**
  * Contains parser classes for the STOMP namespace support.
  */
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.stomp.config;
diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompConnectionFailedEvent.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompConnectionFailedEvent.java
@@ -16,6 +16,8 @@
 
 package org.springframework.integration.stomp.event;
 
+import org.jspecify.annotations.Nullable;
+
 /**
  * The {@link StompIntegrationEvent} implementation for the failed connection exceptions.
  *
@@ -25,7 +27,7 @@
 @SuppressWarnings("serial")
 public class StompConnectionFailedEvent extends StompIntegrationEvent {
 
-	public StompConnectionFailedEvent(Object source, Throwable cause) {
+	public StompConnectionFailedEvent(Object source, @Nullable Throwable cause) {
 		super(source, cause);
 	}
 
diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompExceptionEvent.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompExceptionEvent.java
@@ -16,6 +16,8 @@
 
 package org.springframework.integration.stomp.event;
 
+import org.jspecify.annotations.Nullable;
+
 /**
  * The {@link StompIntegrationEvent} implementation for the exception from STOMP Adapters.
  *
@@ -25,7 +27,7 @@
 @SuppressWarnings("serial")
 public class StompExceptionEvent extends StompIntegrationEvent {
 
-	public StompExceptionEvent(Object source, Throwable cause) {
+	public StompExceptionEvent(Object source, @Nullable Throwable cause) {
 		super(source, cause);
 	}
 
diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompIntegrationEvent.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompIntegrationEvent.java
@@ -16,6 +16,8 @@
 
 package org.springframework.integration.stomp.event;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.integration.events.IntegrationEvent;
 
 /**
@@ -33,7 +35,7 @@ public StompIntegrationEvent(Object source) {
 		super(source);
 	}
 
-	public StompIntegrationEvent(Object source, Throwable cause) {
+	public StompIntegrationEvent(Object source, @Nullable Throwable cause) {
 		super(source, cause);
 	}
 
diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompReceiptEvent.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/StompReceiptEvent.java
@@ -16,6 +16,8 @@
 
 package org.springframework.integration.stomp.event;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.messaging.Message;
 import org.springframework.messaging.simp.stomp.StompCommand;
 
@@ -31,30 +33,31 @@
 @SuppressWarnings("serial")
 public class StompReceiptEvent extends StompIntegrationEvent {
 
-	private final String destination;
+	private final @Nullable String destination;
 
-	private final String receiptId;
+	private final @Nullable String receiptId;
 
 	private final StompCommand stompCommand;
 
 	private final boolean lost;
 
-	private Message<?> message;
+	private @Nullable Message<?> message;
+
+	public StompReceiptEvent(Object source, @Nullable String destination, @Nullable String receiptId,
+		StompCommand stompCommand, boolean lost) {
 
-	public StompReceiptEvent(Object source, String destination, String receiptId, StompCommand stompCommand,
-			boolean lost) {
 		super(source);
 		this.destination = destination;
 		this.receiptId = receiptId;
 		this.stompCommand = stompCommand;
 		this.lost = lost;
 	}
 
-	public String getDestination() {
+	public @Nullable String getDestination() {
 		return this.destination;
 	}
 
-	public String getReceiptId() {
+	public @Nullable String getReceiptId() {
 		return this.receiptId;
 	}
 
@@ -66,7 +69,7 @@ public boolean isLost() {
 		return this.lost;
 	}
 
-	public Message<?> getMessage() {
+	public @Nullable Message<?> getMessage() {
 		return this.message;
 	}
 
diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/package-info.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/event/package-info.java
@@ -0,0 +1,5 @@
+/**
+ * Provides classes which represent STOMP events.
+ */
+@org.jspecify.annotations.NullMarked
+package org.springframework.integration.stomp.event;
diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/StompInboundChannelAdapter.java
@@ -25,6 +25,8 @@
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.context.Lifecycle;
@@ -39,7 +41,6 @@
 import org.springframework.jmx.export.annotation.ManagedAttribute;
 import org.springframework.jmx.export.annotation.ManagedOperation;
 import org.springframework.jmx.export.annotation.ManagedResource;
-import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.simp.stomp.StompCommand;
@@ -82,13 +83,14 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement
 
 	private final Lock destinationLock = new ReentrantLock();
 
+	@SuppressWarnings("NullAway.Init")
 	private ApplicationEventPublisher applicationEventPublisher;
 
 	private Class<?> payloadType = String.class;
 
 	private HeaderMapper<StompHeaders> headerMapper = new StompHeaderMapper();
 
-	private volatile StompSession stompSession;
+	private volatile @Nullable StompSession stompSession;
 
 	public StompInboundChannelAdapter(StompSessionManager stompSessionManager, String... destinations) {
 		Assert.notNull(stompSessionManager, "'stompSessionManager' is required.");
@@ -248,14 +250,14 @@ public void afterConnected(StompSession session, StompHeaders connectedHeaders)
 		}
 
 		@Override
-		public void handleException(StompSession session, StompCommand command, StompHeaders headers, byte[] payload,
+		public void handleException(StompSession session, @Nullable StompCommand command, StompHeaders headers, byte[] payload,
 				Throwable exception) {
 
 			String exceptionMessage = "STOMP Frame handling error in the [" + StompInboundChannelAdapter.this + ']';
 
 			MessageChannel errorChannel = getErrorChannel();
 
-			if (errorChannel != null) {
+			if (command != null && errorChannel != null) {
 				StompHeaderAccessor headerAccessor = StompHeaderAccessor.create(command);
 				headerAccessor.copyHeaders(StompInboundChannelAdapter.this.headerMapper.toHeaders(headers));
 				Message<byte[]> failedMessage =
diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/package-info.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/inbound/package-info.java
@@ -1,4 +1,5 @@
 /**
  * Provides classes which represent inbound STOMP components.
  */
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.stomp.inbound;
diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/StompMessageHandler.java
@@ -21,6 +21,8 @@
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.expression.EvaluationContext;
@@ -34,7 +36,6 @@
 import org.springframework.integration.stomp.event.StompReceiptEvent;
 import org.springframework.integration.stomp.support.StompHeaderMapper;
 import org.springframework.integration.support.management.ManageableLifecycle;
-import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.MessagingException;
@@ -72,17 +73,19 @@ public class StompMessageHandler extends AbstractMessageHandler
 
 	private HeaderMapper<StompHeaders> headerMapper = new StompHeaderMapper();
 
-	private Expression destinationExpression;
+	private @Nullable Expression destinationExpression;
 
+	@SuppressWarnings("NullAway.Init")
 	private EvaluationContext evaluationContext;
 
+	@SuppressWarnings("NullAway.Init")
 	private ApplicationEventPublisher applicationEventPublisher;
 
 	private long connectTimeout = DEFAULT_CONNECT_TIMEOUT;
 
-	private volatile StompSession stompSession;
+	private volatile @Nullable StompSession stompSession;
 
-	private volatile Throwable transportError;
+	private volatile @Nullable Throwable transportError;
 
 	private volatile boolean running;
 
@@ -155,6 +158,7 @@ protected void handleMessageInternal(final Message<?> message) {
 			stompHeaders.setDestination(destination);
 		}
 
+		@SuppressWarnings("NullAway") // Dataflow analysis limitation
 		final StompSession.Receiptable receiptable = session.send(stompHeaders, message.getPayload());
 		if (receiptable.getReceiptId() != null) {
 			final String destination = stompHeaders.getDestination();
@@ -195,7 +199,7 @@ private void connectIfNecessary() throws InterruptedException {
 							throw (ConnectionLostException) this.transportError;
 						}
 						else {
-							throw new ConnectionLostException(this.transportError.getMessage());
+							throw new ConnectionLostException("Failed to connect to STOMP broker", this.transportError);
 						}
 					}
 					else {
@@ -239,7 +243,7 @@ public void afterConnected(StompSession session, StompHeaders connectedHeaders)
 		}
 
 		@Override
-		public void handleFrame(StompHeaders headers, Object payload) {
+		public void handleFrame(StompHeaders headers, @Nullable Object payload) {
 			Object thePayload = payload;
 			if (thePayload == null) {
 				thePayload = headers.getFirst(StompHeaderAccessor.STOMP_MESSAGE_HEADER);
diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/package-info.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/outbound/package-info.java
@@ -1,4 +1,5 @@
 /**
  * Provides classes which represent outbound STOMP components.
  */
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.stomp.outbound;
diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/package-info.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/package-info.java
@@ -1,4 +1,5 @@
 /**
  * Provides core classes STOMP components.
  */
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.stomp;
diff --git a/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/package-info.java b/spring-integration-stomp/src/main/java/org/springframework/integration/stomp/support/package-info.java
@@ -3,4 +3,5 @@
  * @author Artem Bilan
  * @since 4.2
  */
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.stomp.support;

Original file line number	Diff line number	Diff line change
`@@ -16,6 +16,8 @@`
`16`	`16`
`17`	`17`	`package org.springframework.integration.stomp.event;`
`18`	`18`
	`19`	`+import org.jspecify.annotations.Nullable;`
	`20`	`+`
`19`	`21`	`/**`
`20`	`22`	`* The {@link StompIntegrationEvent} implementation for the failed connection exceptions.`
`21`	`23`	`*`
`@@ -25,7 +27,7 @@`
`25`	`27`	`@SuppressWarnings("serial")`
`26`	`28`	`public class StompConnectionFailedEvent extends StompIntegrationEvent {`
`27`	`29`
`28`		`- public StompConnectionFailedEvent(Object source, Throwable cause) {`
	`30`	`+ public StompConnectionFailedEvent(Object source, @Nullable Throwable cause) {`
`29`	`31`	`super(source, cause);`
`30`	`32`	`}`
`31`	`33`