diff --git a/consumer/log-consumer/src/test/java/org/springframework/cloud/fn/consumer/log/LogConsumerApplicationTests.java b/consumer/log-consumer/src/test/java/org/springframework/cloud/fn/consumer/log/LogConsumerApplicationTests.java
index ca7fedf1..08e555ec 100644
--- a/consumer/log-consumer/src/test/java/org/springframework/cloud/fn/consumer/log/LogConsumerApplicationTests.java
+++ b/consumer/log-consumer/src/test/java/org/springframework/cloud/fn/consumer/log/LogConsumerApplicationTests.java
@@ -80,7 +80,7 @@ class LogConsumerApplicationTests {
assertThat(captorMessage.getPayload()).isEqualTo(expectedPayload);
MessageHeaders messageHeaders = captorMessage.getHeaders();
- assertThat(messageHeaders).hasSize(3);
+ assertThat(messageHeaders).hasSize(4);
assertThat(messageHeaders)
.containsEntry(MessageHeaders.CONTENT_TYPE, message.getHeaders().get(MessageHeaders.CONTENT_TYPE));
diff --git a/spring-functions-parent/pom.xml b/spring-functions-parent/pom.xml
index b9e8c578..e5c59e0f 100644
--- a/spring-functions-parent/pom.xml
+++ b/spring-functions-parent/pom.xml
@@ -15,7 +15,7 @@
pom
- 2.3.0.M4
+ 2.3.0.RELEASE
3.0.3.RELEASE
diff --git a/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java b/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java
index b1fe331a..919a3db3 100644
--- a/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java
+++ b/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java
@@ -16,8 +16,12 @@
package org.springframework.cloud.fn.supplier.rabbit;
+import java.util.function.Supplier;
+
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Envelope;
+import reactor.core.publisher.Flux;
+
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
@@ -41,9 +45,6 @@ import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.messaging.Message;
import org.springframework.retry.interceptor.RetryOperationsInterceptor;
import org.springframework.util.Assert;
-import reactor.core.publisher.Flux;
-
-import java.util.function.Supplier;
/**
* A source module that receives data from RabbitMQ.
@@ -55,143 +56,144 @@ import java.util.function.Supplier;
@EnableConfigurationProperties(RabbitSupplierProperties.class)
public class RabbitSupplierConfiguration implements DisposableBean {
- private static final MessagePropertiesConverter inboundMessagePropertiesConverter =
- new DefaultMessagePropertiesConverter() {
+ private static final MessagePropertiesConverter inboundMessagePropertiesConverter =
+ new DefaultMessagePropertiesConverter() {
- @Override
- public MessageProperties toMessageProperties(AMQP.BasicProperties source,
- Envelope envelope,
- String charset) {
- MessageProperties properties = super.toMessageProperties(source, envelope, charset);
- properties.setDeliveryMode(null);
- return properties;
- }
- };
+ @Override
+ public MessageProperties toMessageProperties(AMQP.BasicProperties source,
+ Envelope envelope,
+ String charset) {
+ MessageProperties properties = super.toMessageProperties(source, envelope, charset);
+ properties.setDeliveryMode(null);
+ return properties;
+ }
+ };
- @Autowired
- private RabbitProperties rabbitProperties;
+ @Autowired
+ private RabbitProperties rabbitProperties;
- @Autowired
- private ObjectProvider connectionNameStrategy;
+ @Autowired
+ private ObjectProvider connectionNameStrategy;
- @Autowired
- private RabbitSupplierProperties properties;
+ @Autowired
+ private RabbitSupplierProperties properties;
- @Autowired
- private ConnectionFactory rabbitConnectionFactory;
+ @Autowired
+ private ConnectionFactory rabbitConnectionFactory;
- private CachingConnectionFactory ownConnectionFactory;
+ private CachingConnectionFactory ownConnectionFactory;
- @Bean
- public SimpleMessageListenerContainer container() {
- ConnectionFactory connectionFactory = this.properties.isOwnConnection()
- ? buildLocalConnectionFactory()
- : this.rabbitConnectionFactory;
- SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
- container.setAutoStartup(false);
- RabbitProperties.SimpleContainer simpleContainer = this.rabbitProperties.getListener().getSimple();
+ @Bean
+ public SimpleMessageListenerContainer container() {
+ ConnectionFactory connectionFactory = this.properties.isOwnConnection()
+ ? buildLocalConnectionFactory()
+ : this.rabbitConnectionFactory;
+ SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
+ container.setAutoStartup(false);
+ RabbitProperties.SimpleContainer simpleContainer = this.rabbitProperties.getListener().getSimple();
- AcknowledgeMode acknowledgeMode = simpleContainer.getAcknowledgeMode();
- if (acknowledgeMode != null) {
- container.setAcknowledgeMode(acknowledgeMode);
- }
- Integer concurrency = simpleContainer.getConcurrency();
- if (concurrency != null) {
- container.setConcurrentConsumers(concurrency);
- }
- Integer maxConcurrency = simpleContainer.getMaxConcurrency();
- if (maxConcurrency != null) {
- container.setMaxConcurrentConsumers(maxConcurrency);
- }
- Integer prefetch = simpleContainer.getPrefetch();
- if (prefetch != null) {
- container.setPrefetchCount(prefetch);
- }
- Integer transactionSize = simpleContainer.getBatchSize();
- if (transactionSize != null) {
- container.setBatchSize(transactionSize);
- }
+ AcknowledgeMode acknowledgeMode = simpleContainer.getAcknowledgeMode();
+ if (acknowledgeMode != null) {
+ container.setAcknowledgeMode(acknowledgeMode);
+ }
+ Integer concurrency = simpleContainer.getConcurrency();
+ if (concurrency != null) {
+ container.setConcurrentConsumers(concurrency);
+ }
+ Integer maxConcurrency = simpleContainer.getMaxConcurrency();
+ if (maxConcurrency != null) {
+ container.setMaxConcurrentConsumers(maxConcurrency);
+ }
+ Integer prefetch = simpleContainer.getPrefetch();
+ if (prefetch != null) {
+ container.setPrefetchCount(prefetch);
+ }
+ Integer transactionSize = simpleContainer.getBatchSize();
+ if (transactionSize != null) {
+ container.setBatchSize(transactionSize);
+ }
- container.setDefaultRequeueRejected(this.properties.getRequeue());
- container.setChannelTransacted(this.properties.getTransacted());
- String[] queues = this.properties.getQueues();
- Assert.state(queues.length > 0, "At least one queue is required");
- Assert.noNullElements(queues, "queues cannot have null elements");
- container.setQueueNames(queues);
- if (this.properties.isEnableRetry()) {
- container.setAdviceChain(rabbitSourceRetryInterceptor());
- }
- container.setMessagePropertiesConverter(inboundMessagePropertiesConverter);
- return container;
- }
+ container.setDefaultRequeueRejected(this.properties.getRequeue());
+ container.setChannelTransacted(this.properties.getTransacted());
+ String[] queues = this.properties.getQueues();
+ Assert.state(queues.length > 0, "At least one queue is required");
+ Assert.noNullElements(queues, "queues cannot have null elements");
+ container.setQueueNames(queues);
+ if (this.properties.isEnableRetry()) {
+ container.setAdviceChain(rabbitSourceRetryInterceptor());
+ }
+ container.setMessagePropertiesConverter(inboundMessagePropertiesConverter);
+ return container;
+ }
- @Bean
- public AmqpInboundChannelAdapter adapter(SimpleMessageListenerContainer container,
- FluxMessageChannel channel) {
- return Amqp.inboundAdapter(container)
- .autoStartup(false)
- .outputChannel(channel)
- .mappedRequestHeaders(properties.getMappedRequestHeaders())
- .get();
- }
+ @Bean
+ public AmqpInboundChannelAdapter adapter(SimpleMessageListenerContainer container,
+ FluxMessageChannel channel) {
+ return Amqp.inboundAdapter(container)
+ .autoStartup(false)
+ .outputChannel(channel)
+ .mappedRequestHeaders(properties.getMappedRequestHeaders())
+ .get();
+ }
- @Bean
- public Supplier>> rabbitSupplier(AmqpInboundChannelAdapter adapter,
- FluxMessageChannel channel) {
- return () -> Flux.from(channel).doOnSubscribe(subscription -> adapter.start());
- }
+ @Bean
+ public Supplier>> rabbitSupplier(AmqpInboundChannelAdapter adapter,
+ FluxMessageChannel channel) {
+ return () -> Flux.from(channel).doOnSubscribe(subscription -> adapter.start());
+ }
- @Bean
- public FluxMessageChannel output() {
- return new FluxMessageChannel();
- }
+ @Bean
+ public FluxMessageChannel output() {
+ return new FluxMessageChannel();
+ }
- @Bean
- public RetryOperationsInterceptor rabbitSourceRetryInterceptor() {
- return RetryInterceptorBuilder.stateless()
- .maxAttempts(this.properties.getMaxAttempts())
- .backOffOptions(this.properties.getInitialRetryInterval(), this.properties.getRetryMultiplier(),
- this.properties.getMaxRetryInterval())
- .recoverer(new RejectAndDontRequeueRecoverer())
- .build();
- }
+ @Bean
+ public RetryOperationsInterceptor rabbitSourceRetryInterceptor() {
+ return RetryInterceptorBuilder.stateless()
+ .maxAttempts(this.properties.getMaxAttempts())
+ .backOffOptions(this.properties.getInitialRetryInterval(), this.properties.getRetryMultiplier(),
+ this.properties.getMaxRetryInterval())
+ .recoverer(new RejectAndDontRequeueRecoverer())
+ .build();
+ }
- @Override
- public void destroy() throws Exception {
- if (this.ownConnectionFactory != null) {
- this.ownConnectionFactory.destroy();
- }
- }
+ @Override
+ public void destroy() throws Exception {
+ if (this.ownConnectionFactory != null) {
+ this.ownConnectionFactory.destroy();
+ }
+ }
- private ConnectionFactory buildLocalConnectionFactory() {
- try {
- this.ownConnectionFactory = new AutoConfig.Creator().rabbitConnectionFactory(this.rabbitProperties,
- this.connectionNameStrategy);
- } catch (Exception exception) {
- throw new IllegalStateException("Error building connection factory", exception);
- }
+ private ConnectionFactory buildLocalConnectionFactory() {
+ try {
+ this.ownConnectionFactory = new AutoConfig.Creator().rabbitConnectionFactory(this.rabbitProperties,
+ this.connectionNameStrategy);
+ }
+ catch (Exception exception) {
+ throw new IllegalStateException("Error building connection factory", exception);
+ }
- return this.ownConnectionFactory;
- }
+ return this.ownConnectionFactory;
+ }
}
class AutoConfig extends RabbitAutoConfiguration {
- static class Creator extends RabbitConnectionFactoryCreator {
+ static class Creator extends RabbitConnectionFactoryCreator {
- @Override
- public CachingConnectionFactory rabbitConnectionFactory(RabbitProperties config,
- ObjectProvider connectionNameStrategy) throws Exception {
- CachingConnectionFactory cf = super.rabbitConnectionFactory(config, connectionNameStrategy);
- cf.setConnectionNameStrategy(new ConnectionNameStrategy() {
+ @Override
+ public CachingConnectionFactory rabbitConnectionFactory(RabbitProperties config,
+ ObjectProvider connectionNameStrategy) throws Exception {
+ CachingConnectionFactory cf = super.rabbitConnectionFactory(config, connectionNameStrategy);
+ cf.setConnectionNameStrategy(new ConnectionNameStrategy() {
- @Override
- public String obtainNewConnectionName(ConnectionFactory connectionFactory) {
- return "rabbit.supplier.own.connection";
- }
- });
- cf.afterPropertiesSet();
- return cf;
- }
- }
+ @Override
+ public String obtainNewConnectionName(ConnectionFactory connectionFactory) {
+ return "rabbit.supplier.own.connection";
+ }
+ });
+ cf.afterPropertiesSet();
+ return cf;
+ }
+ }
}
diff --git a/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierProperties.java b/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierProperties.java
index 3c3677c8..9c9be3b4 100644
--- a/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierProperties.java
+++ b/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierProperties.java
@@ -16,146 +16,146 @@
package org.springframework.cloud.fn.supplier.rabbit;
-import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.validation.annotation.Validated;
-
import javax.validation.constraints.NotNull;
import javax.validation.constraints.Size;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
@ConfigurationProperties("rabbit.supplier")
@Validated
public class RabbitSupplierProperties {
- /**
- * Whether rejected messages should be requeued.
- */
- private boolean requeue = true;
+ /**
+ * Whether rejected messages should be requeued.
+ */
+ private boolean requeue = true;
- /**
- * Whether the channel is transacted.
- */
- private boolean transacted = false;
+ /**
+ * Whether the channel is transacted.
+ */
+ private boolean transacted = false;
- /**
- * The queues to which the source will listen for messages.
- */
- private String[] queues;
+ /**
+ * The queues to which the source will listen for messages.
+ */
+ private String[] queues;
- /**
- * Headers that will be mapped.
- */
- private String[] mappedRequestHeaders = {"STANDARD_REQUEST_HEADERS"};
+ /**
+ * Headers that will be mapped.
+ */
+ private String[] mappedRequestHeaders = {"STANDARD_REQUEST_HEADERS"};
- /**
- * Initial retry interval when retry is enabled.
- */
- private int initialRetryInterval = 1000;
+ /**
+ * Initial retry interval when retry is enabled.
+ */
+ private int initialRetryInterval = 1000;
- /**
- * Max retry interval when retry is enabled.
- */
- private int maxRetryInterval = 30000;
+ /**
+ * Max retry interval when retry is enabled.
+ */
+ private int maxRetryInterval = 30000;
- /**
- * Retry backoff multiplier when retry is enabled.
- */
- private double retryMultiplier = 2.0;
+ /**
+ * Retry backoff multiplier when retry is enabled.
+ */
+ private double retryMultiplier = 2.0;
- /**
- * The maximum delivery attempts when retry is enabled.
- */
- private int maxAttempts = 3;
+ /**
+ * The maximum delivery attempts when retry is enabled.
+ */
+ private int maxAttempts = 3;
- /**
- * true to enable retry.
- */
- private boolean enableRetry = false;
+ /**
+ * true to enable retry.
+ */
+ private boolean enableRetry = false;
- /**
- * When true, use a separate connection based on the boot properties.
- */
- private boolean ownConnection;
+ /**
+ * When true, use a separate connection based on the boot properties.
+ */
+ private boolean ownConnection;
- public boolean getRequeue() {
- return requeue;
- }
+ public boolean getRequeue() {
+ return requeue;
+ }
- public void setRequeue(boolean requeue) {
- this.requeue = requeue;
- }
+ public void setRequeue(boolean requeue) {
+ this.requeue = requeue;
+ }
- public boolean getTransacted() {
- return transacted;
- }
+ public boolean getTransacted() {
+ return transacted;
+ }
- public void setTransacted(boolean transacted) {
- this.transacted = transacted;
- }
+ public void setTransacted(boolean transacted) {
+ this.transacted = transacted;
+ }
- @NotNull(message = "queue(s) are required")
- @Size(min = 1, message = "At least one queue is required")
- public String[] getQueues() {
- return queues;
- }
+ @NotNull(message = "queue(s) are required")
+ @Size(min = 1, message = "At least one queue is required")
+ public String[] getQueues() {
+ return queues;
+ }
- public void setQueues(String[] queues) {
- this.queues = queues;
- }
+ public void setQueues(String[] queues) {
+ this.queues = queues;
+ }
- @NotNull
- public String[] getMappedRequestHeaders() {
- return mappedRequestHeaders;
- }
+ @NotNull
+ public String[] getMappedRequestHeaders() {
+ return mappedRequestHeaders;
+ }
- public void setMappedRequestHeaders(String[] mappedRequestHeaders) {
- this.mappedRequestHeaders = mappedRequestHeaders;
- }
+ public void setMappedRequestHeaders(String[] mappedRequestHeaders) {
+ this.mappedRequestHeaders = mappedRequestHeaders;
+ }
- public int getInitialRetryInterval() {
- return initialRetryInterval;
- }
+ public int getInitialRetryInterval() {
+ return initialRetryInterval;
+ }
- public void setInitialRetryInterval(int initialRetryInterval) {
- this.initialRetryInterval = initialRetryInterval;
- }
+ public void setInitialRetryInterval(int initialRetryInterval) {
+ this.initialRetryInterval = initialRetryInterval;
+ }
- public int getMaxRetryInterval() {
- return maxRetryInterval;
- }
+ public int getMaxRetryInterval() {
+ return maxRetryInterval;
+ }
- public void setMaxRetryInterval(int maxRetryInterval) {
- this.maxRetryInterval = maxRetryInterval;
- }
+ public void setMaxRetryInterval(int maxRetryInterval) {
+ this.maxRetryInterval = maxRetryInterval;
+ }
- public double getRetryMultiplier() {
- return retryMultiplier;
- }
+ public double getRetryMultiplier() {
+ return retryMultiplier;
+ }
- public void setRetryMultiplier(double retryMultiplier) {
- this.retryMultiplier = retryMultiplier;
- }
+ public void setRetryMultiplier(double retryMultiplier) {
+ this.retryMultiplier = retryMultiplier;
+ }
- public int getMaxAttempts() {
- return maxAttempts;
- }
+ public int getMaxAttempts() {
+ return maxAttempts;
+ }
- public void setMaxAttempts(int maxAttempts) {
- this.maxAttempts = maxAttempts;
- }
+ public void setMaxAttempts(int maxAttempts) {
+ this.maxAttempts = maxAttempts;
+ }
- public boolean isEnableRetry() {
- return enableRetry;
- }
+ public boolean isEnableRetry() {
+ return enableRetry;
+ }
- public void setEnableRetry(boolean enableRetry) {
- this.enableRetry = enableRetry;
- }
+ public void setEnableRetry(boolean enableRetry) {
+ this.enableRetry = enableRetry;
+ }
- public boolean isOwnConnection() {
- return this.ownConnection;
- }
+ public boolean isOwnConnection() {
+ return this.ownConnection;
+ }
- public void setOwnConnection(boolean ownConnection) {
- this.ownConnection = ownConnection;
- }
+ public void setOwnConnection(boolean ownConnection) {
+ this.ownConnection = ownConnection;
+ }
}