Migrate to JSpecify for nullability constraints (take 5)

Resolves #1150

Signed-off-by: Chris Bono <[email protected]>

Author: onobc

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java

@@ -130,14 +130,17 @@ protected AbstractPulsarMessageToSpringMessageAdapter<V> createMessageHandler(
 		HandlerAdapter handlerMethod = configureListenerAdapter(messageListener, messageHandlerMethodFactory);
 		messageListener.setHandlerMethod(handlerMethod);
 
-		var payloadTypeParams = Arrays.stream(handlerMethod.requireNonNullInvokerHandlerMethod().getMethodParameters())
-			.filter(methodParameter1 -> !methodParameter1.getParameterType().equals(Consumer.class)
-					|| !methodParameter1.getParameterType().equals(Acknowledgement.class)
-					|| !methodParameter1.hasParameterAnnotation(Header.class))
+		// Determine the single payload param to use
+		var methodParameters = handlerMethod.requireNonNullInvokerHandlerMethod().getMethodParameters();
+		var allPayloadParams = Arrays.stream(methodParameters)
+			.filter(param -> !param.getParameterType().equals(Consumer.class)
+					&& !param.getParameterType().equals(Acknowledgement.class)
+					&& !param.hasParameterAnnotation(Header.class))
 			.toList();
-		Assert.isTrue(payloadTypeParams.size() == 1,
-				() -> "Expected exactly 1 payload type param but found " + payloadTypeParams.size());
-		MethodParameter messageParameter = payloadTypeParams.get(0);
+		Assert.isTrue(allPayloadParams.size() == 1, "Expected 1 payload types but found " + allPayloadParams);
+		var messageParameter = allPayloadParams.stream()
+			.findFirst()
+			.orElseThrow(() -> new IllegalArgumentException("Unable to determine message parameter"));
 
 		DefaultReactivePulsarMessageListenerContainer<Object> containerInstance = (DefaultReactivePulsarMessageListenerContainer<Object>) container;
 		ReactivePulsarContainerProperties<Object> pulsarContainerProperties = containerInstance

spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java

@@ -145,17 +145,17 @@ protected AbstractPulsarMessageToSpringMessageAdapter<V> createMessageListener(
 				messageConverter);
 		HandlerAdapter handlerMethod = configureListenerAdapter(messageListener, messageHandlerMethodFactory);
 		messageListener.setHandlerMethod(handlerMethod);
-
-		// TODO: filter out the payload type by excluding Consumer, Message, Messages etc.
-
-		var payloadTypeParams = Arrays.stream(handlerMethod.requireNonNullInvokerHandlerMethod().getMethodParameters())
-			.filter(methodParameter1 -> !methodParameter1.getParameterType().equals(Consumer.class)
-					|| !methodParameter1.getParameterType().equals(Acknowledgement.class)
-					|| !methodParameter1.hasParameterAnnotation(Header.class))
+		// Determine the single payload param to use
+		var methodParameters = handlerMethod.requireNonNullInvokerHandlerMethod().getMethodParameters();
+		var allPayloadParams = Arrays.stream(methodParameters)
+			.filter(param -> !param.getParameterType().equals(Consumer.class)
+					&& !param.getParameterType().equals(Acknowledgement.class)
+					&& !param.hasParameterAnnotation(Header.class))
 			.toList();
-		Assert.isTrue(payloadTypeParams.size() == 1,
-				() -> "Expected exactly 1 payload type param but found " + payloadTypeParams.size());
-		MethodParameter messageParameter = payloadTypeParams.get(0);
+		Assert.isTrue(allPayloadParams.size() == 1, "Expected 1 payload types but found " + allPayloadParams);
+		var messageParameter = allPayloadParams.stream()
+			.findFirst()
+			.orElseThrow(() -> new IllegalArgumentException("Unable to determine message parameter"));
 
 		ConcurrentPulsarMessageListenerContainer<?> containerInstance = (ConcurrentPulsarMessageListenerContainer<?>) container;
 		PulsarContainerProperties pulsarContainerProperties = containerInstance.getContainerProperties();

spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarReaderEndpoint.java

@@ -115,16 +115,17 @@ protected AbstractPulsarMessageToSpringMessageAdapter<V> createReaderListener(
 		HandlerAdapter handlerMethod = configureListenerAdapter(readerListener, messageHandlerMethodFactory);
 		readerListener.setHandlerMethod(handlerMethod);
 
-		// TODO: filter out the payload type by excluding Consumer, Message, Messages etc.
-
-		var payloadTypeParams = Arrays.stream(handlerMethod.requireNonNullInvokerHandlerMethod().getMethodParameters())
-			.filter(methodParameter1 -> !methodParameter1.getParameterType().equals(Consumer.class)
-					|| !methodParameter1.getParameterType().equals(Acknowledgement.class)
-					|| !methodParameter1.hasParameterAnnotation(Header.class))
+		// Determine the single payload param to use
+		var methodParameters = handlerMethod.requireNonNullInvokerHandlerMethod().getMethodParameters();
+		var allPayloadParams = Arrays.stream(methodParameters)
+			.filter(param -> !param.getParameterType().equals(Consumer.class)
+					&& !param.getParameterType().equals(Acknowledgement.class)
+					&& !param.hasParameterAnnotation(Header.class))
 			.toList();
-		Assert.isTrue(payloadTypeParams.size() == 1,
-				() -> "Expected exactly 1 payload type param but found " + payloadTypeParams.size());
-		MethodParameter messageParameter = payloadTypeParams.get(0);
+		Assert.isTrue(allPayloadParams.size() == 1, "Expected 1 payload types but found " + allPayloadParams);
+		var messageParameter = allPayloadParams.stream()
+			.findFirst()
+			.orElseThrow(() -> new IllegalArgumentException("Unable to determine message parameter"));
 
 		DefaultPulsarMessageReaderContainer<?> containerInstance = (DefaultPulsarMessageReaderContainer<?>) container;
 		PulsarReaderContainerProperties pulsarContainerProperties = containerInstance.getContainerProperties();

spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java

@@ -617,13 +617,13 @@ protected void doInTransactionWithoutResult(TransactionStatus status) {
 			return resourceHolder != null ? resourceHolder.getTransaction() : null;
 		}
 
+		@SuppressWarnings("NullAway")
 		private @Nullable RuntimeException dispatchMessageToListener(Message<T> message, AtomicBoolean inRetryMode,
 				@Nullable Transaction txn) {
 			try {
-				if (this.listener instanceof PulsarAcknowledgingMessageListener
-						&& this.ackMode.equals(AckMode.MANUAL)) {
-					this.listener.received(this.consumer, message,
-							new ConsumerAcknowledgment(this.consumer, message, txn));
+				if (this.listener instanceof PulsarAcknowledgingMessageListener) {
+					this.listener.received(this.consumer, message, this.ackMode.equals(AckMode.MANUAL)
+							? new ConsumerAcknowledgment(this.consumer, message, txn) : null);
 				}
 				else if (this.listener != null) {
 					this.listener.received(this.consumer, message);