Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-10083: Migrate ZeroMQ module to Jspecify

Browse Source 
Related to: https://github.com/spring-projects/spring-integration/issues/10083

- Replaced `org.springframework.lang.Nullable` with `org.jspecify.annotations.Nullable`

* Complete ZeroMQ module migration to JSpecify

- Migrate `package-info.java` files to use `@NullMarked` annotation
- Add `@SuppressWarnings("NullAway.Init")` for fields initialized in lifecycle methods

* Fix NullAway violations in ZeroMqChannel

- Extract `this.zeroMqProxy` to local variable inside `Mono.defer()` block

Signed-off-by: Jooyoung Pyoung <pyoungjy@gmail.com>
This commit is contained in:
Jooyoung Pyoung
2025-06-11 04:24:48 +09:00
committed by GitHub
parent 260073d18a
commit d63dd2637c
9 changed files with 28 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqProxy.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2023 the original author or authors.
* Copyright 2020-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -26,6 +26,7 @@ import java.util.function.Consumer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jspecify.annotations.Nullable;
import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;
@@ -35,7 +36,6 @@ import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
@@ -82,8 +82,10 @@ public class ZeroMqProxy implements InitializingBean, SmartLifecycle, BeanNameAw
private final AtomicInteger backendPort = new AtomicInteger();
@SuppressWarnings("NullAway.Init")
private String controlAddress;
@SuppressWarnings("NullAway.Init")
private Executor proxyExecutor;
@Nullable
@@ -97,6 +99,7 @@ public class ZeroMqProxy implements InitializingBean, SmartLifecycle, BeanNameAw
@Nullable
private String captureAddress;
@SuppressWarnings("NullAway.Init")
private String beanName;
private boolean autoStartup = true;

15
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/ZeroMqChannel.java
View File

@@ -23,6 +23,7 @@ import java.util.Map;
import java.util.function.Consumer;
import java.util.function.Supplier;
import org.jspecify.annotations.Nullable;
import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;
@@ -36,7 +37,6 @@ import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.mapping.BytesMessageMapper;
import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper;
import org.springframework.integration.zeromq.ZeroMqProxy;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.SubscribableChannel;
@@ -66,6 +66,7 @@ import org.springframework.util.Assert;
* concurrency primitives for multi-publisher(subscriber) communication within the same application.
*
* @author Artem Bilan
* @author Jooyoung Pyoung
*
* @since 5.4
*/
@@ -142,19 +143,19 @@ public class ZeroMqChannel extends AbstractMessageChannel implements Subscribabl
this.subscriberData = prepareSubscriberDataFlux();
}
@SuppressWarnings("this-escape")
private Mono<Integer> prepareProxyMono() {
return Mono.defer(() -> {
if (this.zeroMqProxy != null) {
return Mono.fromCallable(() -> this.zeroMqProxy.getBackendPort())
ZeroMqProxy zeroMqProxyToUse = this.zeroMqProxy;
if (zeroMqProxyToUse != null) {
return Mono.fromCallable(zeroMqProxyToUse::getBackendPort)
.filter((proxyPort) -> proxyPort > 0)
.repeatWhenEmpty(100, (repeat) -> repeat.delayElements(Duration.ofMillis(100))) // NOSONAR
.doOnNext((proxyPort) ->
setConnectUrl("tcp://localhost:" + this.zeroMqProxy.getFrontendPort() +
':' + this.zeroMqProxy.getBackendPort()))
setConnectUrl("tcp://localhost:" + zeroMqProxyToUse.getFrontendPort() +
':' + zeroMqProxyToUse.getBackendPort()))
.doOnError((error) ->
logger.error(error,
() -> "The provided '" + this.zeroMqProxy + "' has not been started"));
() -> "The provided '" + zeroMqProxyToUse + "' has not been started"));
}
else {
return Mono.empty();

2
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/package-info.java
View File

@@ -1,5 +1,5 @@
/**
* Provides classes for message channels support over ZeroMQ.
*/
@org.springframework.lang.NonNullApi
@org.jspecify.annotations.NullMarked
package org.springframework.integration.zeromq.channel;

3
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/package-info.java
View File

@@ -1,6 +1,5 @@
/**
* Provides classes for supporting ZeroMQ component via Java DSL.
*/
@org.springframework.lang.NonNullApi
@org.springframework.lang.NonNullFields
@org.jspecify.annotations.NullMarked
package org.springframework.integration.zeromq.dsl;

6
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2024 the original author or authors.
* Copyright 2020-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -24,6 +24,7 @@ import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import org.jspecify.annotations.Nullable;
import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZFrame;
@@ -43,7 +44,6 @@ import org.springframework.integration.zeromq.ZeroMqHeaders;
import org.springframework.integration.zeromq.ZeroMqUtils;
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.converter.MessageConverter;
import org.springframework.util.Assert;
@@ -77,6 +77,7 @@ public class ZeroMqMessageProducer extends MessageProducerSupport {
private final SocketType socketType;
@SuppressWarnings("NullAway.Init")
private InboundMessageMapper<byte[]> messageMapper;
private Consumer<ZMQ.Socket> socketConfigurer = (socket) -> {
@@ -91,6 +92,7 @@ public class ZeroMqMessageProducer extends MessageProducerSupport {
@Nullable
private String connectUrl;
@SuppressWarnings("NullAway.Init")
private volatile Mono<ZMQ.Socket> socketMono;
private volatile boolean unwrapTopic = true;

2
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/package-info.java
View File

@@ -1,5 +1,5 @@
/**
* Provides classes for inbound channel adapters over ZeroMQ.
*/
@org.springframework.lang.NonNullApi
@org.jspecify.annotations.NullMarked
package org.springframework.integration.zeromq.inbound;

8
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2024 the original author or authors.
* Copyright 2020-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,6 +22,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Supplier;
import org.jspecify.annotations.Nullable;
import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZFrame;
@@ -45,7 +46,6 @@ import org.springframework.integration.mapping.OutboundMessageMapper;
import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter;
import org.springframework.integration.support.management.ManageableLifecycle;
import org.springframework.integration.zeromq.ZeroMqUtils;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.util.Assert;
@@ -77,8 +77,10 @@ public class ZeroMqMessageHandler extends AbstractReactiveMessageHandler
private final Scheduler publisherScheduler = Schedulers.newSingle("zeroMqMessageHandlerScheduler");
@SuppressWarnings("NullAway.Init")
private volatile Mono<ZMQ.Socket> socketMono;
@SuppressWarnings("NullAway.Init")
private OutboundMessageMapper<byte[]> messageMapper;
private Consumer<ZMQ.Socket> socketConfigurer = (socket) -> {
@@ -86,10 +88,12 @@ public class ZeroMqMessageHandler extends AbstractReactiveMessageHandler
private Expression topicExpression = new SupplierExpression<>(() -> null);
@SuppressWarnings("NullAway.Init")
private EvaluationContext evaluationContext;
private volatile boolean initialized;
@SuppressWarnings("NullAway.Init")
private volatile Disposable socketMonoSubscriber;
private volatile boolean wrapTopic = true;

2
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/package-info.java
View File

@@ -1,5 +1,5 @@
/**
* Provides classes for outbound channel adapters over ZeroMQ.
*/
@org.springframework.lang.NonNullApi
@org.jspecify.annotations.NullMarked
package org.springframework.integration.zeromq.outbound;

2
spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/package-info.java
View File

@@ -1,5 +1,5 @@
/**
* Provides common classes for supporting ZeroMQ components.
*/
@org.springframework.lang.NonNullApi
@org.jspecify.annotations.NullMarked
package org.springframework.integration.zeromq;