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

Add package-info.java setting all things @NonNull

Browse Source 
- Adjusted producer pieces to get rid of warnings
- Other areas to be updated as we encounter the warnings
This commit is contained in:
Chris Bono
2022-10-22 11:18:01 -05:00
committed by Soby Chacko
parent 979cb11d31
commit fed6a0c40f
21 changed files with 174 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-pulsar-sample-apps/src/main/java/app1/package-info.java Normal file
View File

@@ -0,0 +1,9 @@
/**
* Package containing sample app for the framework.
*/
@NonNullApi
@NonNullFields
package app1;
import org.springframework.lang.NonNullApi;
import org.springframework.lang.NonNullFields;

9
spring-pulsar-sample-apps/src/main/java/app2/package-info.java Normal file
View File

@@ -0,0 +1,9 @@
/**
* Package containing sample app for the framework.
*/
@NonNullApi
@NonNullFields
package app2;
import org.springframework.lang.NonNullApi;
import org.springframework.lang.NonNullFields;

11
spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/package-info.java Normal file
View File

@@ -0,0 +1,11 @@
/**
* Package containing the Spring Boot
* {@link org.springframework.boot.autoconfigure.AutoConfiguration} for the Spring for
* Apache Pulsar framework.
*/
@NonNullApi
@NonNullFields
package org.springframework.pulsar.autoconfigure;
import org.springframework.lang.NonNullApi;
import org.springframework.lang.NonNullFields;

9
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/package-info.java Normal file
View File

@@ -0,0 +1,9 @@
/**
* Package containing annotations used by the framework.
*/
@NonNullApi
@NonNullFields
package org.springframework.pulsar.annotation;
import org.springframework.lang.NonNullApi;
import org.springframework.lang.NonNullFields;

3
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientFactoryBean.java
View File

@@ -22,6 +22,7 @@ import org.apache.pulsar.client.api.PulsarClient;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.core.log.LogAccessor;
import org.springframework.lang.Nullable;
/**
* {@link FactoryBean} implementation for the {@link PulsarClient}.
@@ -50,7 +51,7 @@ public class PulsarClientFactoryBean extends AbstractFactoryBean<PulsarClient> {
}
@Override
protected void destroyInstance(PulsarClient instance) throws Exception {
protected void destroyInstance(@Nullable PulsarClient instance) throws Exception {
if (instance != null) {
this.logger.info(() -> "Closing client " + instance);
instance.close();

9
spring-pulsar/src/main/java/org/springframework/pulsar/config/package-info.java Normal file
View File

@@ -0,0 +1,9 @@
/**
* Package containing Spring configuration classes for the framework.
*/
@NonNullApi
@NonNullFields
package org.springframework.pulsar.config;
import org.springframework.lang.NonNullApi;
import org.springframework.lang.NonNullFields;

17
spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java
View File

@@ -38,7 +38,6 @@ import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.core.log.LogAccessor;
import org.springframework.lang.NonNull;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -92,9 +91,9 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
}
@Override
protected Producer<T> doCreateProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
List<ProducerInterceptor> producerInterceptors,
List<ProducerBuilderCustomizer<T>> producerBuilderCustomizers) {
protected Producer<T> doCreateProducer(@Nullable String topic, Schema<T> schema,
@Nullable MessageRouter messageRouter, @Nullable List<ProducerInterceptor> producerInterceptors,
@Nullable List<ProducerBuilderCustomizer<T>> producerBuilderCustomizers) {
final String topicName = ProducerUtils.resolveTopicName(topic, this);
ProducerCacheKey<T> producerCacheKey = new ProducerCacheKey<>(schema, topicName, messageRouter,
producerInterceptors);
@@ -102,9 +101,9 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
st.interceptors, producerBuilderCustomizers));
}
private Producer<T> createCacheableProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
List<ProducerInterceptor> producerInterceptors,
List<ProducerBuilderCustomizer<T>> producerBuilderCustomizers) {
private Producer<T> createCacheableProducer(String topic, Schema<T> schema, @Nullable MessageRouter messageRouter,
@Nullable List<ProducerInterceptor> producerInterceptors,
@Nullable List<ProducerBuilderCustomizer<T>> producerBuilderCustomizers) {
try {
Producer<T> producer = super.doCreateProducer(topic, schema, messageRouter, producerInterceptors,
producerBuilderCustomizers);
@@ -124,7 +123,7 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
factory.addAdvice(new MethodInterceptor() {
@Nullable
@Override
public Object invoke(@NonNull MethodInvocation invocation) throws Throwable {
public Object invoke(MethodInvocation invocation) throws Throwable {
if (invocation.getMethod().getName().equals("close")) {
closeCallback.accept((Producer<T>) invocation.getThis());
return null;
@@ -171,8 +170,10 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
private final String topic;
@Nullable
private final MessageRouter router;
@Nullable
private final List<ProducerInterceptor> interceptors;
/**

21
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java
View File

@@ -30,6 +30,7 @@ import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
import org.springframework.core.log.LogAccessor;
import org.springframework.lang.Nullable;
import org.springframework.util.CollectionUtils;
/**
@@ -56,26 +57,26 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
}
@Override
public Producer<T> createProducer(String topic, Schema<T> schema) throws PulsarClientException {
public Producer<T> createProducer(@Nullable String topic, Schema<T> schema) throws PulsarClientException {
return doCreateProducer(topic, schema, null, null, null);
}
@Override
public Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter)
public Producer<T> createProducer(@Nullable String topic, Schema<T> schema, @Nullable MessageRouter messageRouter)
throws PulsarClientException {
return doCreateProducer(topic, schema, messageRouter, null, null);
}
@Override
public Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
List<ProducerInterceptor> producerInterceptors) throws PulsarClientException {
public Producer<T> createProducer(@Nullable String topic, Schema<T> schema, @Nullable MessageRouter messageRouter,
@Nullable List<ProducerInterceptor> producerInterceptors) throws PulsarClientException {
return doCreateProducer(topic, schema, messageRouter, producerInterceptors, null);
}
@Override
public Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
List<ProducerInterceptor> producerInterceptors,
List<ProducerBuilderCustomizer<T>> producerBuilderCustomizers) throws PulsarClientException {
public Producer<T> createProducer(@Nullable String topic, Schema<T> schema, @Nullable MessageRouter messageRouter,
@Nullable List<ProducerInterceptor> producerInterceptors,
@Nullable List<ProducerBuilderCustomizer<T>> producerBuilderCustomizers) throws PulsarClientException {
return doCreateProducer(topic, schema, messageRouter, producerInterceptors, producerBuilderCustomizers);
}
@@ -91,9 +92,9 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
* @return the created producer
* @throws PulsarClientException if any error occurs
*/
protected Producer<T> doCreateProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
List<ProducerInterceptor> producerInterceptors,
List<ProducerBuilderCustomizer<T>> producerBuilderCustomizers) throws PulsarClientException {
protected Producer<T> doCreateProducer(@Nullable String topic, Schema<T> schema,
@Nullable MessageRouter messageRouter, @Nullable List<ProducerInterceptor> producerInterceptors,
@Nullable List<ProducerBuilderCustomizer<T>> producerBuilderCustomizers) throws PulsarClientException {
final String resolvedTopic = ProducerUtils.resolveTopicName(topic, this);
this.logger.trace(() -> String.format("Creating producer for '%s' topic", resolvedTopic));
final ProducerBuilder<T> producerBuilder = this.pulsarClient.newProducer(schema);

3
spring-pulsar/src/main/java/org/springframework/pulsar/core/ProducerUtils.java
View File

@@ -21,6 +21,7 @@ import java.util.Optional;
import org.apache.pulsar.client.api.Producer;
import org.springframework.core.log.LogAccessor;
import org.springframework.lang.Nullable;
import org.springframework.util.StringUtils;
/**
@@ -37,7 +38,7 @@ final class ProducerUtils {
return String.format("(%s:%s)", producer.getProducerName(), producer.getTopic());
}
static <T> String resolveTopicName(String userSpecifiedTopic, PulsarProducerFactory<T> producerFactory) {
static <T> String resolveTopicName(@Nullable String userSpecifiedTopic, PulsarProducerFactory<T> producerFactory) {
if (StringUtils.hasText(userSpecifiedTopic)) {
return userSpecifiedTopic;
}

6
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarOperations.java
View File

@@ -22,6 +22,8 @@ import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageRouter;
import org.apache.pulsar.client.api.PulsarClientException;
import org.springframework.lang.Nullable;
/**
* The basic Pulsar operations contract.
*
@@ -47,7 +49,7 @@ public interface PulsarOperations<T> {
* @return the id assigned by the broker to the published message
* @throws PulsarClientException if an error occurs
*/
MessageId send(String topic, T message) throws PulsarClientException;
MessageId send(@Nullable String topic, T message) throws PulsarClientException;
/**
* Sends a message to the default topic in a non-blocking manner.
@@ -65,7 +67,7 @@ public interface PulsarOperations<T> {
* @return a future that holds the id assigned by the broker to the published message
* @throws PulsarClientException if an error occurs
*/
CompletableFuture<MessageId> sendAsync(String topic, T message) throws PulsarClientException;
CompletableFuture<MessageId> sendAsync(@Nullable String topic, T message) throws PulsarClientException;
/**
* Create a {@link SendMessageBuilder builder} for configuring and sending a message.

8
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java
View File

@@ -25,6 +25,8 @@ import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
import org.springframework.lang.Nullable;
/**
* The strategy to create a {@link Producer} instance(s).
*
@@ -82,9 +84,9 @@ public interface PulsarProducerFactory<T> {
* @return the producer
* @throws PulsarClientException if any error occurs
*/
Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
List<ProducerInterceptor> producerInterceptors,
List<ProducerBuilderCustomizer<T>> producerBuilderCustomizers) throws PulsarClientException;
Producer<T> createProducer(@Nullable String topic, Schema<T> schema, @Nullable MessageRouter messageRouter,
@Nullable List<ProducerInterceptor> producerInterceptors,
@Nullable List<ProducerBuilderCustomizer<T>> producerBuilderCustomizers) throws PulsarClientException;
/**
* Return a map of configuration options to use when creating producers.

33
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java
View File

@@ -53,14 +53,18 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, BeanNameAware {
private final PulsarProducerFactory<T> producerFactory;
@Nullable
private final List<ProducerInterceptor> interceptors;
@Nullable
private final ObservationRegistry observationRegistry;
@Nullable
private final PulsarTemplateObservationConvention observationConvention;
private String beanName;
private String beanName = "";
@Nullable
private Schema<T> schema;
/**
@@ -76,7 +80,7 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, BeanNameAware {
* @param producerFactory the factory used to create the backing Pulsar producers.
* @param interceptors the interceptors to add to the producer.
*/
public PulsarTemplate(PulsarProducerFactory<T> producerFactory, List<ProducerInterceptor> interceptors) {
public PulsarTemplate(PulsarProducerFactory<T> producerFactory, @Nullable List<ProducerInterceptor> interceptors) {
this(producerFactory, interceptors, null, null);
}
@@ -105,7 +109,7 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, BeanNameAware {
}
@Override
public MessageId send(String topic, T message) throws PulsarClientException {
public MessageId send(@Nullable String topic, T message) throws PulsarClientException {
return doSend(topic, message, null, null, null);
}
@@ -115,7 +119,7 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, BeanNameAware {
}
@Override
public CompletableFuture<MessageId> sendAsync(String topic, T message) throws PulsarClientException {
public CompletableFuture<MessageId> sendAsync(@Nullable String topic, T message) throws PulsarClientException {
return doSendAsync(topic, message, null, null, null);
}
@@ -137,8 +141,10 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, BeanNameAware {
this.schema = schema;
}
private MessageId doSend(String topic, T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer,
MessageRouter messageRouter, ProducerBuilderCustomizer<T> producerCustomizer) throws PulsarClientException {
private MessageId doSend(@Nullable String topic, T message,
@Nullable TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer,
@Nullable MessageRouter messageRouter, @Nullable ProducerBuilderCustomizer<T> producerCustomizer)
throws PulsarClientException {
try {
return doSendAsync(topic, message, typedMessageBuilderCustomizer, messageRouter, producerCustomizer).get();
}
@@ -147,9 +153,10 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, BeanNameAware {
}
}
private CompletableFuture<MessageId> doSendAsync(String topic, T message,
TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer, MessageRouter messageRouter,
ProducerBuilderCustomizer<T> producerCustomizer) throws PulsarClientException {
private CompletableFuture<MessageId> doSendAsync(@Nullable String topic, T message,
@Nullable TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer,
@Nullable MessageRouter messageRouter, @Nullable ProducerBuilderCustomizer<T> producerCustomizer)
throws PulsarClientException {
final String topicName = ProducerUtils.resolveTopicName(topic, this.producerFactory);
this.logger.trace(() -> String.format("Sending msg to '%s' topic", topicName));
@@ -193,8 +200,8 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, BeanNameAware {
DefaultPulsarTemplateObservationConvention.INSTANCE, () -> senderContext, this.observationRegistry);
}
private Producer<T> prepareProducerForSend(String topic, T message, MessageRouter messageRouter,
ProducerBuilderCustomizer<T> producerCustomizer) throws PulsarClientException {
private Producer<T> prepareProducerForSend(@Nullable String topic, T message, @Nullable MessageRouter messageRouter,
@Nullable ProducerBuilderCustomizer<T> producerCustomizer) throws PulsarClientException {
Schema<T> schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message);
return this.producerFactory.createProducer(topic, schema, messageRouter, this.interceptors,
producerCustomizer == null ? Collections.emptyList() : Collections.singletonList(producerCustomizer));
@@ -206,12 +213,16 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, BeanNameAware {
private final T message;
@Nullable
private String topic;
@Nullable
private TypedMessageBuilderCustomizer<T> messageCustomizer;
@Nullable
private MessageRouter messageRouter;
@Nullable
private ProducerBuilderCustomizer<T> producerCustomizer;
SendMessageBuilderImpl(PulsarTemplate<T> template, T message) {

9
spring-pulsar/src/main/java/org/springframework/pulsar/core/package-info.java Normal file
View File

@@ -0,0 +1,9 @@
/**
* Package containing the core components of the framework.
*/
@NonNullApi
@NonNullFields
package org.springframework.pulsar.core;
import org.springframework.lang.NonNullApi;
import org.springframework.lang.NonNullFields;

9
spring-pulsar/src/main/java/org/springframework/pulsar/event/package-info.java Normal file
View File

@@ -0,0 +1,9 @@
/**
* Package containing events produced by the framework.
*/
@NonNullApi
@NonNullFields
package org.springframework.pulsar.event;
import org.springframework.lang.NonNullApi;
import org.springframework.lang.NonNullFields;

9
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/package-info.java Normal file
View File

@@ -0,0 +1,9 @@
/**
* Package containing components for adapting Pulsar messages to message listeners.
*/
@NonNullApi
@NonNullFields
package org.springframework.pulsar.listener.adapter;
import org.springframework.lang.NonNullApi;
import org.springframework.lang.NonNullFields;

9
spring-pulsar/src/main/java/org/springframework/pulsar/listener/package-info.java Normal file
View File

@@ -0,0 +1,9 @@
/**
* Package containing listener components for receiving Pulsar messages.
*/
@NonNullApi
@NonNullFields
package org.springframework.pulsar.listener;
import org.springframework.lang.NonNullApi;
import org.springframework.lang.NonNullFields;

9
spring-pulsar/src/main/java/org/springframework/pulsar/observation/package-info.java Normal file
View File

@@ -0,0 +1,9 @@
/**
* Package containing components that provide support for the Observation API.
*/
@NonNullApi
@NonNullFields
package org.springframework.pulsar.observation;
import org.springframework.lang.NonNullApi;
import org.springframework.lang.NonNullFields;

9
spring-pulsar/src/main/java/org/springframework/pulsar/package-info.java Normal file
View File

@@ -0,0 +1,9 @@
/**
* Package containing the Spring for Apache Pulsar framework classes.
*/
@NonNullApi
@NonNullFields
package org.springframework.pulsar;
import org.springframework.lang.NonNullApi;
import org.springframework.lang.NonNullFields;

2
spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarRecordMessageConverter.java
View File

@@ -20,7 +20,6 @@ import java.lang.reflect.Type;
import org.apache.pulsar.client.api.Consumer;
import org.springframework.lang.NonNull;
import org.springframework.messaging.Message;
import org.springframework.pulsar.support.MessageConverter;
@@ -32,7 +31,6 @@ import org.springframework.pulsar.support.MessageConverter;
*/
public interface PulsarRecordMessageConverter<T> extends MessageConverter {
@NonNull
Message<?> toMessage(org.apache.pulsar.client.api.Message<T> record, Consumer<T> consumer, Type payloadType);
T fromMessage(Message<?> message, String defaultTopic);

9
spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/package-info.java Normal file
View File

@@ -0,0 +1,9 @@
/**
* Package containing message converters.
*/
@NonNullApi
@NonNullFields
package org.springframework.pulsar.support.converter;
import org.springframework.lang.NonNullApi;
import org.springframework.lang.NonNullFields;

9
spring-pulsar/src/main/java/org/springframework/pulsar/support/package-info.java Normal file
View File

@@ -0,0 +1,9 @@
/**
* Package containing general support components.
*/
@NonNullApi
@NonNullFields
package org.springframework.pulsar.support;
import org.springframework.lang.NonNullApi;
import org.springframework.lang.NonNullFields;