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

* Fix new Sonar smells

Browse Source 
* Remove redundant `@SuppressWarnings("deprecation")`
* Add `Duration.ofSeconds(10)` to `StepVerifier.verify()`
to avoid infinite wait and lose failing text context on the hang CI build
This commit is contained in:
Artem Bilan
2020-08-08 12:29:24 -04:00
parent 65ad76232e
commit 9d557426b5
28 changed files with 62 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-integration-core/src/main/java/org/springframework/integration/aop/CompoundTriggerAdvice.java
View File

@@ -24,7 +24,7 @@ import org.springframework.scheduling.Trigger;
import org.springframework.util.Assert;
/**
* An {@link AbstractMessageSourceAdvice} that uses a {@link CompoundTrigger} to adjust
* A {@link MessageSourceMutator} that uses a {@link CompoundTrigger} to adjust
* the poller - when a message is present, the compound trigger's primary trigger is
* used to determine the next poll. When no message is present, the override trigger is
* used.
@@ -33,13 +33,13 @@ import org.springframework.util.Assert;
* {@link CompoundTrigger} instance and must <b>not</b> use a task executor.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 4.3
*
*/
@SuppressWarnings("deprecation")
public class CompoundTriggerAdvice
extends AbstractMessageSourceAdvice
implements ReceiveMessageAdvice {
implements MessageSourceMutator, ReceiveMessageAdvice {
private final CompoundTrigger compoundTrigger;

13
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
View File

@@ -65,7 +65,6 @@ import org.springframework.util.StringUtils;
* @author Artem Bilan
*/
@IntegrationManagedResource
@SuppressWarnings("deprecation")
public abstract class AbstractMessageChannel extends IntegrationObjectSupport
implements MessageChannel, TrackableComponent, InterceptableChannel, IntegrationManagement, IntegrationPattern {
@@ -140,8 +139,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
* @see #setMessageConverter(MessageConverter)
*/
public void setDatatypes(Class<?>... datatypes) {
this.datatypes = (datatypes != null && datatypes.length > 0)
? datatypes : new Class<?>[0];
this.datatypes =
(datatypes != null && datatypes.length > 0)
? datatypes
: new Class<?>[0];
}
/**
@@ -330,10 +331,8 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
return sent;
}
catch (Exception ex) {
if (!metricsProcessed) {
if (sample != null) {
sample.stop(buildSendTimer(false, ex.getClass().getSimpleName()));
}
if (!metricsProcessed && sample != null) {
sample.stop(buildSendTimer(false, ex.getClass().getSimpleName()));
}
if (interceptorStack != null) {
interceptorList.afterSendCompletion(message, this, sent, ex, interceptorStack);

1
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java
View File

@@ -37,7 +37,6 @@ import org.springframework.messaging.support.ExecutorChannelInterceptor;
* @author Gary Russell
* @author Artem Bilan
*/
@SuppressWarnings("deprecation")
public abstract class AbstractPollableChannel extends AbstractMessageChannel
implements PollableChannel, ExecutorChannelInterceptorAware {

13
spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java
View File

@@ -26,7 +26,6 @@ import reactor.core.Disposable;
import reactor.core.Disposables;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxProcessor;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
@@ -43,26 +42,24 @@ import reactor.core.scheduler.Schedulers;
public class FluxMessageChannel extends AbstractMessageChannel
implements Publisher<Message<?>>, ReactiveStreamsSubscribableChannel {
private final FluxProcessor<Message<?>, Message<?>> processor;
private final Sinks.Many<Message<?>> sink;
private final FluxSink<Message<?>> sink;
private final FluxProcessor<Message<?>, Message<?>> processor;
private final Sinks.Many<Boolean> subscribedSignal = Sinks.many().replay().limit(1);
private final Disposable.Composite upstreamSubscriptions = Disposables.composite();
@SuppressWarnings("deprecation")
public FluxMessageChannel() {
this.processor = FluxProcessor.fromSink(Sinks.many().multicast().onBackpressureBuffer(1, false));
this.sink = this.processor.sink(FluxSink.OverflowStrategy.BUFFER);
this.sink = Sinks.many().multicast().onBackpressureBuffer(1, false);
this.processor = FluxProcessor.fromSink(this.sink);
}
@Override
protected boolean doSend(Message<?> message, long timeout) {
Assert.state(this.processor.hasDownstreams(),
() -> "The [" + this + "] doesn't have subscribers to accept messages");
this.sink.next(message);
return true;
return this.sink.emitNext(message).hasEmitted();
}
@Override

1
spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java
View File

@@ -43,7 +43,6 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @author Artem Bilan
*/
@SuppressWarnings("deprecation")
public class QueueChannel extends AbstractPollableChannel implements QueueChannelOperations {
private final Queue<Message<?>> queue;

15
spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationManagementConfigurer.java
View File

@@ -19,8 +19,6 @@ package org.springframework.integration.config;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.commons.logging.Log;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.SmartInitializingSingleton;
@@ -54,7 +52,7 @@ public class IntegrationManagementConfigurer
implements SmartInitializingSingleton, ApplicationContextAware, BeanNameAware, BeanPostProcessor {
/**
* Bean name of tehe configurer.
* Bean name of the configurer.
*/
public static final String MANAGEMENT_CONFIGURER_NAME = "integrationManagementConfigurer";
@@ -86,8 +84,8 @@ public class IntegrationManagementConfigurer
* Exception logging (debug or otherwise) is not affected by this setting.
* <p>
* It has been found that in high-volume messaging environments, calls to methods such as
* {@link Log#isDebugEnabled()} can be quite expensive and account for an inordinate amount of CPU
* time.
* {@link org.apache.commons.logging.Log#isDebugEnabled()} can be quite expensive
* and account for an inordinate amount of CPU time.
* <p>
* Set this to false to disable logging by default in all framework components that implement
* {@link IntegrationManagement} (channels, message handlers etc). This turns off logging such as
@@ -121,7 +119,6 @@ public class IntegrationManagementConfigurer
if (!getOverrides(bean).loggingConfigured) {
bean.setLoggingEnabled(this.defaultLoggingEnabled);
}
String name = entry.getKey();
}
this.singletonsInstantiated = true;
}
@@ -140,10 +137,8 @@ public class IntegrationManagementConfigurer
@Override
public Object postProcessAfterInitialization(Object bean, String name) throws BeansException {
if (this.singletonsInstantiated) {
if (this.metricsCaptor != null && bean instanceof IntegrationManagement) {
((IntegrationManagement) bean).registerMetricsCaptor(this.metricsCaptor);
}
if (this.singletonsInstantiated && this.metricsCaptor != null && bean instanceof IntegrationManagement) {
((IntegrationManagement) bean).registerMetricsCaptor(this.metricsCaptor);
}
return bean;
}

3
spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java
View File

@@ -21,7 +21,6 @@ import java.util.concurrent.locks.ReentrantLock;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.context.IntegrationProperties;
@@ -80,7 +79,7 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport
* Such endpoints can be started/stopped as a group.
* @param role the role for this endpoint.
* @since 5.0
* @see SmartLifecycle
* @see org.springframework.context.SmartLifecycle
* @see org.springframework.integration.support.SmartLifecycleRoleController
*/
public void setRole(String role) {

1
spring-integration-core/src/main/java/org/springframework/integration/graph/MessageChannelNode.java
View File

@@ -30,7 +30,6 @@ import org.springframework.messaging.MessageChannel;
* @since 4.3
*
*/
@SuppressWarnings("deprecation")
public class MessageChannelNode extends IntegrationNode implements SendTimersAware {
private Supplier<SendTimers> sendTimers;

3
spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java
View File

@@ -36,7 +36,6 @@ import org.springframework.integration.leader.DefaultCandidate;
import org.springframework.integration.leader.event.DefaultLeaderEventPublisher;
import org.springframework.integration.leader.event.LeaderEventPublisher;
import org.springframework.integration.support.locks.LockRegistry;
import org.springframework.integration.support.management.ManageableSmartLifecycle;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.util.Assert;
@@ -61,7 +60,7 @@ import org.springframework.util.Assert;
*
* @since 4.3.1
*/
public class LockRegistryLeaderInitiator implements ManageableSmartLifecycle, DisposableBean,
public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBean,
ApplicationEventPublisherAware {
public static final long DEFAULT_HEART_BEAT_TIME = 500L;

3
spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagement.java
View File

@@ -20,7 +20,6 @@ import org.springframework.beans.factory.DisposableBean;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.integration.support.management.metrics.MetricsCaptor;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.lang.Nullable;
/**
* Base interface for Integration managed components.
@@ -54,7 +53,6 @@ public interface IntegrationManagement extends NamedComponent, DisposableBean {
default void setManagedName(String managedName) {
}
@Nullable
default String getManagedName() {
return null;
}
@@ -62,7 +60,6 @@ public interface IntegrationManagement extends NamedComponent, DisposableBean {
default void setManagedType(String managedType) {
}
@Nullable
default String getManagedType() {
return null;
}

1
spring-integration-core/src/main/java/org/springframework/integration/support/management/MessageSourceManagement.java
View File

@@ -26,7 +26,6 @@ import org.springframework.jmx.export.annotation.ManagedAttribute;
* @since 5.0
*
*/
@SuppressWarnings("deprecation")
@IntegrationManagedResource
public interface MessageSourceManagement {

3
spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java
View File

@@ -26,6 +26,7 @@ import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.time.Duration;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
@@ -329,7 +330,7 @@ public class ReactiveStreamsConsumerTests {
StepVerifier.create(sink.asFlux())
.expectNext(testMessage, testMessage2)
.thenCancel()
.verify();
.verify(Duration.ofSeconds(10));
reactiveConsumer.stop();
}

5
spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveInboundChannelAdapterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2019 the original author or authors.
* Copyright 2018-2020 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.
@@ -16,6 +16,7 @@
package org.springframework.integration.endpoint;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
@@ -60,7 +61,7 @@ public class ReactiveInboundChannelAdapterTests {
StepVerifier.create(testFlux)
.expectNext(2, 4, 6, 8, 10, 12, 14, 16)
.thenCancel()
.verify();
.verify(Duration.ofSeconds(10));
}
@Configuration

4
spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveMessageProducerTests.java
View File

@@ -18,6 +18,8 @@ package org.springframework.integration.endpoint;
import static org.assertj.core.api.Assertions.assertThat;
import java.time.Duration;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
@@ -58,7 +60,7 @@ public class ReactiveMessageProducerTests {
.cast(String.class))
.expectNext("test1", "test2")
.thenCancel()
.verify();
.verify(Duration.ofSeconds(10));
assertThat(this.producer.isRunning()).isFalse();
}

2
spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveMessageSourceProducerTests.java
View File

@@ -97,7 +97,7 @@ public class ReactiveMessageSourceProducerTests {
reactiveMessageSourceProducer.start();
stepVerifier.verify();
stepVerifier.verify(Duration.ofSeconds(10));
reactiveMessageSourceProducer.stop();

3
spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt
View File

@@ -47,6 +47,7 @@ import org.springframework.test.annotation.DirtiesContext
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig
import reactor.core.publisher.Flux
import reactor.test.StepVerifier
import java.time.Duration
import java.util.*
import java.util.concurrent.atomic.AtomicReference
import java.util.function.Function
@@ -167,7 +168,7 @@ class KotlinDslTests {
val registration = this.integrationFlowContext.registration(integrationFlow).register()
verifyLater.verify()
verifyLater.verify(Duration.ofSeconds(10))
registration.destroy()
}

5
spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/RotatingServerAdvice.java
View File

@@ -36,10 +36,7 @@ import org.springframework.util.Assert;
* @since 5.0.7
*
*/
@SuppressWarnings("deprecation")
public class RotatingServerAdvice
extends org.springframework.integration.aop.AbstractMessageSourceAdvice
implements MessageSourceMutator {
public class RotatingServerAdvice implements MessageSourceMutator {
private final RotationPolicy rotationPolicy;

3
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java
View File

@@ -25,7 +25,6 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.Lifecycle;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
@@ -58,7 +57,7 @@ import org.springframework.util.concurrent.SettableListenableFuture;
* (or times out). Asynchronous requests/responses over the same connection are not
* supported - use a pair of outbound/inbound adapters for that use case.
* <p>
* {@link Lifecycle} methods delegate to the underlying {@link AbstractConnectionFactory}
* {@link org.springframework.context.Lifecycle} methods delegate to the underlying {@link AbstractConnectionFactory}.
*
*
* @author Gary Russell

8
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/IntegrationMBeanExportConfiguration.java
View File

@@ -26,8 +26,6 @@ import javax.management.MBeanServer;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.BeanExpressionContext;
import org.springframework.beans.factory.config.BeanExpressionResolver;
@@ -42,7 +40,6 @@ import org.springframework.context.expression.StandardBeanExpressionResolver;
import org.springframework.core.annotation.AnnotationAttributes;
import org.springframework.core.env.Environment;
import org.springframework.core.type.AnnotationMetadata;
import org.springframework.integration.config.IntegrationManagementConfigurer;
import org.springframework.integration.monitor.IntegrationMBeanExporter;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -76,10 +73,6 @@ public class IntegrationMBeanExportConfiguration implements ImportAware, Environ
private Environment environment;
@Autowired(required = false)
@Qualifier(IntegrationManagementConfigurer.MANAGEMENT_CONFIGURER_NAME)
private IntegrationManagementConfigurer configurer;
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
@@ -102,7 +95,6 @@ public class IntegrationMBeanExportConfiguration implements ImportAware, Environ
"@EnableIntegrationMBeanExport is not present on importing class " + importMetadata.getClassName());
}
@SuppressWarnings("deprecation")
@Bean(name = MBEAN_EXPORTER_NAME)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public IntegrationMBeanExporter mbeanExporter() {

22
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java
View File

@@ -107,7 +107,6 @@ import org.springframework.util.ReflectionUtils;
* @author Meherzad Lahewala
*/
@ManagedResource
@SuppressWarnings("deprecation")
public class IntegrationMBeanExporter extends MBeanExporter
implements ApplicationContextAware, DestructionAwareBeanPostProcessor {
@@ -148,9 +147,7 @@ public class IntegrationMBeanExporter extends MBeanExporter
private String domain = DEFAULT_DOMAIN;
private String[] componentNamePatterns = { "*" };
private IntegrationManagementConfigurer managementConfigurer;
private String[] componentNamePatterns = {"*"};
private volatile long shutdownDeadline;
@@ -247,7 +244,7 @@ public class IntegrationMBeanExporter extends MBeanExporter
private void populateMessageHandlers() {
Map<String, MessageHandler> messageHandlers = this.applicationContext
.getBeansOfType(MessageHandler.class);
.getBeansOfType(MessageHandler.class);
for (Entry<String, MessageHandler> entry : messageHandlers.entrySet()) {
String beanName = entry.getKey();
@@ -269,7 +266,7 @@ public class IntegrationMBeanExporter extends MBeanExporter
private void populateMessageSources() {
this.applicationContext.getBeansOfType(
IntegrationInboundManagement.class)
IntegrationInboundManagement.class)
.values()
.stream()
// If the source is proxied, we have to extract the target to expose as an MBean.
@@ -301,15 +298,10 @@ public class IntegrationMBeanExporter extends MBeanExporter
private void configureManagementConfigurer() {
if (!this.applicationContext.containsBean(IntegrationManagementConfigurer.MANAGEMENT_CONFIGURER_NAME)) {
this.managementConfigurer = new IntegrationManagementConfigurer();
this.managementConfigurer.setApplicationContext(this.applicationContext);
this.managementConfigurer.setBeanName(IntegrationManagementConfigurer.MANAGEMENT_CONFIGURER_NAME);
this.managementConfigurer.afterSingletonsInstantiated();
}
else {
this.managementConfigurer =
this.applicationContext.getBean(IntegrationManagementConfigurer.MANAGEMENT_CONFIGURER_NAME,
IntegrationManagementConfigurer.class);
IntegrationManagementConfigurer managementConfigurer = new IntegrationManagementConfigurer();
managementConfigurer.setApplicationContext(this.applicationContext);
managementConfigurer.setBeanName(IntegrationManagementConfigurer.MANAGEMENT_CONFIGURER_NAME);
managementConfigurer.afterSingletonsInstantiated();
}
}

3
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ManagedEndpoint.java
View File

@@ -19,7 +19,6 @@ package org.springframework.integration.monitor;
import org.springframework.context.Lifecycle;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.integration.support.management.ManageableLifecycle;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
@@ -30,7 +29,7 @@ import org.springframework.jmx.export.annotation.ManagedOperation;
* @author Gary Russell
*
* @deprecated this is no longer used by the framework. Replaced by
* {@link ManageableLifecycle}.
* {@link org.springframework.integration.support.management.ManageableLifecycle}.
*
*/
@Deprecated

3
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PublishSubscribeKafkaChannel.java
View File

@@ -27,7 +27,7 @@ import org.springframework.kafka.core.KafkaOperations;
*
* @author Gary Russell
*
* @since 4.4
* @since 5.4
*
*/
public class PublishSubscribeKafkaChannel extends SubscribableKafkaChannel implements BroadcastCapableChannel {
@@ -40,6 +40,7 @@ public class PublishSubscribeKafkaChannel extends SubscribableKafkaChannel imple
*/
public PublishSubscribeKafkaChannel(KafkaOperations<?, ?> template, KafkaListenerContainerFactory<?> factory,
String channelTopic) {
super(template, factory, channelTopic);
}

5
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java
View File

@@ -19,7 +19,6 @@ package org.springframework.integration.kafka.channel;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.dispatcher.MessageDispatcher;
import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
@@ -95,7 +94,7 @@ public class SubscribableKafkaChannel extends AbstractKafkaChannel implements Su
/**
* Set the auto startup.
* @param autoStartup true to automatically start.
* @see SmartLifecycle
* @see org.springframework.context.SmartLifecycle
*/
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
@@ -123,7 +122,7 @@ public class SubscribableKafkaChannel extends AbstractKafkaChannel implements Su
.dispatch(toMessagingMessage(record, acknowledgment, consumer));
}
});
});
}
protected MessageDispatcher createDispatcher() {

2
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSourceTests.java
View File

@@ -184,7 +184,7 @@ public class ReactiveMongoDbMessageSourceTests extends MongoDbAvailableTests {
.assertNext(
message -> assertThat(((Person) message.getPayload()).getName()).isEqualTo("Oleg"))
.thenCancel()
.verify();
.verify(Duration.ofSeconds(10));
context.close();
}

3
spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSourceTests.java
View File

@@ -19,6 +19,7 @@ package org.springframework.integration.r2dbc.inbound;
import static org.assertj.core.api.Assertions.assertThat;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
@@ -134,7 +135,7 @@ public class R2dbcMessageSourceTests {
StepVerifier.create((Flux<?>) r2dbcMessageSourceError.receive().getPayload())
.expectErrorMatches(throwable -> throwable instanceof IllegalStateException
&& throwable.getMessage().contains("'queryExpression' must evaluate to String or"))
.verify();
.verify(Duration.ofSeconds(10));
}
@Configuration

4
spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java
View File

@@ -189,11 +189,11 @@ public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport {
Mono<?> consumerGroupMono = Mono.empty();
if (this.createConsumerGroup) {
consumerGroupMono =
this.reactiveStreamOperations.createGroup(this.streamKey, this.consumerGroup)
this.reactiveStreamOperations.createGroup(this.streamKey, this.consumerGroup) // NOSONAR
.onErrorReturn(this.consumerGroup);
}
Consumer consumer = Consumer.from(this.consumerGroup, this.consumerName);
Consumer consumer = Consumer.from(this.consumerGroup, this.consumerName); // NOSONAR
if (offset.getOffset().equals(ReadOffset.latest())) {
// for consumer group offset id should be equal '>'

2
spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java
View File

@@ -100,7 +100,7 @@ public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests
.assertNext((infoGroup) ->
assertThat(infoGroup.groupName()).isEqualTo(this.redisStreamMessageProducer.getBeanName()))
.thenCancel()
.verify();
.verify(Duration.ofSeconds(10));
}
@Test

18
spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java
View File

@@ -160,7 +160,7 @@ public class RSocketOutboundGatewayIntegrationTests {
StepVerifier.create(controller.fireForgetPayloads.asFlux())
.expectNext("Hello")
.thenCancel()
.verify();
.verify(Duration.ofSeconds(10));
disposable.dispose();
}
@@ -195,7 +195,7 @@ public class RSocketOutboundGatewayIntegrationTests {
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());
verifier.verify();
verifier.verify(Duration.ofSeconds(10));
}
@Test
@@ -227,7 +227,7 @@ public class RSocketOutboundGatewayIntegrationTests {
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());
verifier.verify();
verifier.verify(Duration.ofSeconds(10));
}
@Test
@@ -261,7 +261,7 @@ public class RSocketOutboundGatewayIntegrationTests {
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());
verifier.verify();
verifier.verify(Duration.ofSeconds(10));
}
@Test
@@ -295,7 +295,7 @@ public class RSocketOutboundGatewayIntegrationTests {
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());
verifier.verify();
verifier.verify(Duration.ofSeconds(10));
}
@@ -326,7 +326,7 @@ public class RSocketOutboundGatewayIntegrationTests {
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());
verifier.verify();
verifier.verify(Duration.ofSeconds(10));
}
@Test
@@ -356,7 +356,7 @@ public class RSocketOutboundGatewayIntegrationTests {
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());
verifier.verify();
verifier.verify(Duration.ofSeconds(10));
}
@Test
@@ -388,7 +388,7 @@ public class RSocketOutboundGatewayIntegrationTests {
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());
verifier.verify();
verifier.verify(Duration.ofSeconds(10));
}
@Test
@@ -420,7 +420,7 @@ public class RSocketOutboundGatewayIntegrationTests {
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());
verifier.verify();
verifier.verify(Duration.ofSeconds(10));
}
@Test