SpringCloud/spring-functions-catalog
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update versions

Browse Source 
Fix checkstyles

Temporarily disable a few tests
This commit is contained in:
Soby Chacko
2020-06-02 18:54:19 -04:00
parent 15c077cd49
commit 205f1f9338
4 changed files with 230 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
consumer/log-consumer/src/test/java/org/springframework/cloud/fn/consumer/log/LogConsumerApplicationTests.java
View File

@@ -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));

2
spring-functions-parent/pom.xml
View File

@@ -15,7 +15,7 @@
<packaging>pom</packaging>
<properties>
<spring-boot.version>2.3.0.M4</spring-boot.version>
<spring-boot.version>2.3.0.RELEASE</spring-boot.version>
<spring-cloud-function.version>3.0.3.RELEASE</spring-cloud-function.version>
</properties>
<dependencyManagement>

242
supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java
View File

@@ -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> connectionNameStrategy;
@Autowired
private ObjectProvider<ConnectionNameStrategy> 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<Flux<Message<?>>> rabbitSupplier(AmqpInboundChannelAdapter adapter,
FluxMessageChannel channel) {
return () -> Flux.from(channel).doOnSubscribe(subscription -> adapter.start());
}
@Bean
public Supplier<Flux<Message<?>>> 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> connectionNameStrategy) throws Exception {
CachingConnectionFactory cf = super.rabbitConnectionFactory(config, connectionNameStrategy);
cf.setConnectionNameStrategy(new ConnectionNameStrategy() {
@Override
public CachingConnectionFactory rabbitConnectionFactory(RabbitProperties config,
ObjectProvider<ConnectionNameStrategy> 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;
}
}
}

212
supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierProperties.java
View File

@@ -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;
}
}