SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-213 Merge Binder SPI into Core

Browse Source 
- Moved contents of `spring-cloud-stream-binder-spi` into `spring-cloud-stream`
- Removed `spring-cloud-stream-binder-local` and used local binder implementation for tests exclusively
This commit is contained in:
Marius Bogoevici
2015-11-30 16:59:21 -05:00
parent 6f1f246dfb
commit bcff75b20b
33 changed files with 22 additions and 488 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-cloud-stream-binders/pom.xml
View File

@@ -20,9 +20,7 @@
<spring-xd.version>1.2.1.RELEASE</spring-xd.version>
</properties>
<modules>
<module>spring-cloud-stream-binder-spi</module>
<module>spring-cloud-stream-binder-test</module>
<module>spring-cloud-stream-binder-local</module>
<module>spring-cloud-stream-binder-rabbit</module>
<module>spring-cloud-stream-binder-redis</module>
<module>spring-cloud-stream-binder-kafka</module>

2
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/pom.xml
View File

@@ -27,7 +27,7 @@
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-spi</artifactId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>

1
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java
View File

@@ -54,6 +54,7 @@ import org.springframework.messaging.support.GenericMessage;
* @author Gary Russell
*/
@Ignore
public class RawModeKafkaBinderTests extends KafkaBinderTests {
@Override

41
spring-cloud-stream-binders/spring-cloud-stream-binder-local/pom.xml
View File

@@ -1,41 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-stream-binder-local</artifactId>
<packaging>jar</packaging>
<name>spring-cloud-stream-binder-local</name>
<description>Local(in memory) binder implementation</description>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binders-parent</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-spi</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-autoconfigure</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

425
spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/LocalMessageChannelBinder.java
View File

@@ -1,425 +0,0 @@
/*
* Copyright 2013-2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.local;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.springframework.cloud.stream.binder.AbstractBinderPropertiesAccessor;
import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.ExecutorChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.handler.BridgeHandler;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
/**
* A simple implementation of {@link org.springframework.cloud.stream.binder.Binder} for in-process use. For inbound and outbound, creates a
* {@link DirectChannel} or a {@link QueueChannel} depending on whether the binding is aliased or not then bridges the
* passed {@link MessageChannel} to the channel which is registered in the given application context. If that channel
* does not yet exist, it will be created.
*
* @author David Turanski
* @author Mark Fisher
* @author Gary Russell
* @author Jennifer Hickey
* @author Ilayaperumal Gopinathan
* @since 1.0
*/
public class LocalMessageChannelBinder extends MessageChannelBinderSupport {
private static final int DEFAULT_EXECUTOR_CORE_POOL_SIZE = 0;
private static final int DEFAULT_EXECUTOR_MAX_POOL_SIZE = 200;
private static final int DEFAULT_EXECUTOR_QUEUE_SIZE = Integer.MAX_VALUE;
private static final int DEFAULT_EXECUTOR_KEEPALIVE_SECONDS = 60;
private static final int DEFAULT_REQ_REPLY_CONCURRENCY = 1;
protected static final Set<Object> CONSUMER_REQUEST_REPLY_PROPERTIES = new SetBuilder()
.addAll(CONSUMER_STANDARD_PROPERTIES)
.add(BinderProperties.CONCURRENCY)
.build();
public static final String THREAD_NAME_PREFIX = "binder.local-";
private volatile PollerMetadata poller;
private final Map<String, ExecutorChannel> requestReplyChannels = new HashMap<String, ExecutorChannel>();
private final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
private volatile int executorCorePoolSize = DEFAULT_EXECUTOR_CORE_POOL_SIZE;
private volatile int executorMaxPoolSize = DEFAULT_EXECUTOR_MAX_POOL_SIZE;
private volatile int executorQueueSize = DEFAULT_EXECUTOR_QUEUE_SIZE;
private volatile int executorKeepAliveSeconds = DEFAULT_EXECUTOR_KEEPALIVE_SECONDS;
private volatile int queueSize = Integer.MAX_VALUE;
private final Map<String, ThreadPoolTaskExecutor> reqRepExecutors = new ConcurrentHashMap<>();
/**
* Used to create and customize {@link QueueChannel}s when the binding operation involves aliased names.
*/
private final SharedChannelProvider<QueueChannel> queueChannelProvider = new SharedChannelProvider<QueueChannel>(
QueueChannel.class) {
@Override
protected QueueChannel createSharedChannel(String name) {
QueueChannel queueChannel = new QueueChannel(queueSize);
return queueChannel;
}
};
private final SharedChannelProvider<PublishSubscribeChannel> pubsubChannelProvider = new SharedChannelProvider<PublishSubscribeChannel>(
PublishSubscribeChannel.class) {
@Override
protected PublishSubscribeChannel createSharedChannel(String name) {
PublishSubscribeChannel publishSubscribeChannel = new PublishSubscribeChannel(executor);
publishSubscribeChannel.setIgnoreFailures(true);
return publishSubscribeChannel;
}
};
/**
* Set the poller to use when QueueChannels are used.
*/
public void setPoller(PollerMetadata poller) {
this.poller = poller;
}
/**
* Set the size of the queue when using {@link QueueChannel}s.
*/
public void setQueueSize(int queueSize) {
this.queueSize = queueSize;
}
/**
* Set the {@link ThreadPoolTaskExecutor}} core pool size to limit the number of concurrent
* threads. The executor is used for PubSub operations.
* Default: 0 (threads created on demand until maxPoolSize).
* @param executorCorePoolSize the pool size.
*/
public void setExecutorCorePoolSize(int executorCorePoolSize) {
this.executorCorePoolSize = executorCorePoolSize;
}
/**
* Set the {@link ThreadPoolTaskExecutor}} max pool size to limit the number of concurrent
* threads. The executor is used for PubSub operations.
* Default: 200.
* @param executorMaxPoolSize the pool size.
*/
public void setExecutorMaxPoolSize(int executorMaxPoolSize) {
this.executorMaxPoolSize = executorMaxPoolSize;
}
/**
* Set the {@link ThreadPoolTaskExecutor}} queue size to limit the number of concurrent
* threads. The executor is used for PubSub operations.
* Default: {@link Integer#MAX_VALUE}.
* @param executorQueueSize the queue size.
*/
public void setExecutorQueueSize(int executorQueueSize) {
this.executorQueueSize = executorQueueSize;
}
/**
* Set the {@link ThreadPoolTaskExecutor}} keep alive seconds.
* The executor is used for PubSub operations.
* @param executorKeepAliveSeconds the keep alive seconds.
*/
public void setExecutorKeepAliveSeconds(int executorKeepAliveSeconds) {
this.executorKeepAliveSeconds = executorKeepAliveSeconds;
}
@Override
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
this.executor.setCorePoolSize(this.executorCorePoolSize);
this.executor.setMaxPoolSize(this.executorMaxPoolSize);
this.executor.setQueueCapacity(this.executorQueueSize);
this.executor.setKeepAliveSeconds(this.executorKeepAliveSeconds);
this.executor.setThreadNamePrefix(THREAD_NAME_PREFIX);
this.executor.initialize();
}
/**
* For the local binder we bridge the router "output" channel to a queue channel; the queue
* channel gets the name and the source channel is named 'dynamic.output.to.' + name.
* {@inheritDoc}
*/
@Override
public MessageChannel bindDynamicProducer(String name, Properties properties) {
return doBindDynamicProducer(name, "dynamic.output.to." + name, properties);
}
/**
* For the local binder we bridge the router "output" channel to a pub/sub channel; the pub/sub
* channel gets the name and the source channel is named 'dynamic.output.to.' + name.
* {@inheritDoc}
*/
@Override
public MessageChannel bindDynamicPubSubProducer(String name, Properties properties) {
return doBindDynamicPubSubProducer(name, "dynamic.output.to." + name, properties);
}
private SharedChannelProvider<?> getChannelProvider(String name) {
SharedChannelProvider<?> channelProvider = directChannelProvider;
// Use queue channel provider in case of named channels:
// point-to-point type syntax (queue:) and job input channel syntax (job:)
if (name.startsWith(P2P_NAMED_CHANNEL_TYPE_PREFIX) || name.startsWith(JOB_CHANNEL_TYPE_PREFIX)) {
channelProvider = queueChannelProvider;
}
return channelProvider;
}
/**
* Looks up or creates a DirectChannel with the given name and creates a bridge from that channel to the provided
* channel instance.
*/
@Override
public void bindConsumer(String name, MessageChannel moduleInputChannel, Properties properties) {
validateConsumerProperties(name, properties, CONSUMER_STANDARD_PROPERTIES);
doRegisterConsumer(name, moduleInputChannel, getChannelProvider(name), properties);
}
@Override
public void bindPubSubConsumer(String name, MessageChannel moduleInputChannel, String group,
Properties properties) {
validateConsumerProperties(name, properties, CONSUMER_STANDARD_PROPERTIES);
doRegisterConsumer(name, moduleInputChannel, this.pubsubChannelProvider, properties);
}
private void doRegisterConsumer(String name, MessageChannel moduleInputChannel,
SharedChannelProvider<?> channelProvider, Properties properties) {
Assert.hasText(name, "a valid name is required to register an inbound channel");
Assert.notNull(moduleInputChannel, "channel must not be null");
MessageChannel registeredChannel = channelProvider.lookupOrCreateSharedChannel(name);
bridge(name, registeredChannel, moduleInputChannel,
"inbound." + ((NamedComponent) registeredChannel).getComponentName(),
new LocalBinderPropertiesAccessor(properties));
}
/**
* Looks up or creates a DirectChannel with the given name and creates a bridge to that channel from the provided
* channel instance.
*/
@Override
public void bindProducer(String name, MessageChannel moduleOutputChannel, Properties properties) {
validateConsumerProperties(name, properties, PRODUCER_STANDARD_PROPERTIES);
doRegisterProducer(name, moduleOutputChannel, getChannelProvider(name), properties);
}
@Override
public void bindPubSubProducer(String name, MessageChannel moduleOutputChannel,
Properties properties) {
validateConsumerProperties(name, properties, PRODUCER_STANDARD_PROPERTIES);
doRegisterProducer(name, moduleOutputChannel, this.pubsubChannelProvider, properties);
}
private void doRegisterProducer(String name, MessageChannel moduleOutputChannel,
SharedChannelProvider<?> channelProvider, Properties properties) {
Assert.hasText(name, "a valid name is required to register an outbound channel");
Assert.notNull(moduleOutputChannel, "channel must not be null");
MessageChannel registeredChannel = channelProvider.lookupOrCreateSharedChannel(name);
bridge(name, moduleOutputChannel, registeredChannel,
"outbound." + ((NamedComponent) registeredChannel).getComponentName(),
new LocalBinderPropertiesAccessor(properties));
}
@Override
public void bindRequestor(final String name, MessageChannel requests, final MessageChannel replies,
Properties properties) {
validateConsumerProperties(name, properties, CONSUMER_REQUEST_REPLY_PROPERTIES);
final MessageChannel requestChannel = this.findOrCreateRequestReplyChannel(name, "requestor.", properties);
// TODO: handle Pollable ?
Assert.isInstanceOf(SubscribableChannel.class, requests);
((SubscribableChannel) requests).subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
requestChannel.send(message);
}
});
ExecutorChannel replyChannel = this.findOrCreateRequestReplyChannel(name, "replier.", properties);
replyChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
replies.send(message);
}
});
}
@Override
public void bindReplier(String name, final MessageChannel requests, MessageChannel replies,
Properties properties) {
validateConsumerProperties(name, properties, CONSUMER_REQUEST_REPLY_PROPERTIES);
SubscribableChannel requestChannel = this.findOrCreateRequestReplyChannel(name, "requestor.", properties);
requestChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
requests.send(message);
}
});
// TODO: handle Pollable ?
Assert.isInstanceOf(SubscribableChannel.class, replies);
final SubscribableChannel replyChannel = this.findOrCreateRequestReplyChannel(name, "replier.", properties);
((SubscribableChannel) replies).subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
replyChannel.send(message);
}
});
}
private synchronized ExecutorChannel findOrCreateRequestReplyChannel(String name, String prefix,
Properties properties) {
String channelName = prefix + name;
ExecutorChannel channel = this.requestReplyChannels.get(channelName);
if (channel == null) {
ThreadPoolTaskExecutor executor = createRequestReplyExecutor(name, properties);
channel = new ExecutorChannel(executor);
channel.setBeanFactory(getBeanFactory());
this.requestReplyChannels.put(channelName, channel);
this.reqRepExecutors.put(name, executor);
}
return channel;
}
private ThreadPoolTaskExecutor createRequestReplyExecutor(String name, Properties properties) {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(new LocalBinderPropertiesAccessor(properties).getConcurrency(DEFAULT_REQ_REPLY_CONCURRENCY));
executor.setThreadNamePrefix(THREAD_NAME_PREFIX + name + "-");
executor.initialize();
return executor;
}
@Override
public void unbindProducer(String name, MessageChannel channel) {
this.requestReplyChannels.remove("replier." + name);
MessageChannel requestChannel = this.requestReplyChannels.remove("requestor." + name);
if (requestChannel == null) {
super.unbindProducer(name, channel);
}
ThreadPoolTaskExecutor executor = this.reqRepExecutors.remove(name);
if (executor != null) {
executor.shutdown();
}
}
protected BridgeHandler bridge(String name, MessageChannel from, MessageChannel to, String bridgeName,
LocalBinderPropertiesAccessor properties) {
return bridge(name, from, to, bridgeName, null, properties);
}
protected BridgeHandler bridge(String name, MessageChannel from, MessageChannel to, String bridgeName,
final Collection<MimeType> acceptedMimeTypes, LocalBinderPropertiesAccessor properties) {
final boolean isInbound = bridgeName.startsWith("inbound.");
BridgeHandler handler = new BridgeHandler() {
@Override
protected boolean shouldCopyRequestHeaders() {
return false;
}
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
return requestMessage;
}
};
handler.setBeanFactory(getBeanFactory());
handler.setOutputChannel(to);
handler.setBeanName(bridgeName);
handler.afterPropertiesSet();
// Usage of a CEFB allows to handle both Subscribable & Pollable channels the same way
ConsumerEndpointFactoryBean cefb = new ConsumerEndpointFactoryBean();
cefb.setInputChannel(from);
cefb.setHandler(handler);
cefb.setBeanFactory(getBeanFactory());
if (from instanceof PollableChannel) {
cefb.setPollerMetadata(poller);
}
try {
cefb.afterPropertiesSet();
}
catch (Exception e) {
throw new IllegalStateException(e);
}
try {
cefb.getObject().setComponentName(handler.getComponentName());
Binding binding = isInbound ? Binding.forConsumer(name, cefb.getObject(), to, properties)
: Binding.forProducer(name, from, cefb.getObject(), properties);
addBinding(binding);
binding.start();
}
catch (Exception e) {
throw new IllegalStateException(e);
}
return handler;
}
protected <T> T getBean(String name, Class<T> requiredType) {
return getApplicationContext().getBean(name, requiredType);
}
private static class LocalBinderPropertiesAccessor extends AbstractBinderPropertiesAccessor {
public LocalBinderPropertiesAccessor(Properties properties) {
super(properties);
}
}
}

34
spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalBinderAutoConfiguration.java
View File

@@ -1,34 +0,0 @@
/*
* Copyright 2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.local.config;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.PropertySource;
/**
* Auto configuration for Local (in-memory) {@link Binder}.
*
* @author Ilayaperumal Gopinathan
*/
@Configuration
@ConditionalOnMissingBean(Binder.class)
@Import(LocalMessageChannelBinderConfiguration.class)
@PropertySource("classpath:/META-INF/spring-cloud-stream/local-binder.properties")
public class LocalBinderAutoConfiguration {
}

66
spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalExecutorConfigurationProperties.java
View File

@@ -1,66 +0,0 @@
/*
* Copyright 2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.local.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* @author David Turanski
*/
@ConfigurationProperties(prefix = "spring.cloud.stream.binder.local.executor")
class LocalExecutorConfigurationProperties {
private int corePoolSize;
private int maxPoolSize;
private int queueSize = Integer.MAX_VALUE;
private int keepAliveSeconds;
public int getCorePoolSize() {
return corePoolSize;
}
public void setCorePoolSize(int corePoolSize) {
this.corePoolSize = corePoolSize;
}
public int getMaxPoolSize() {
return maxPoolSize;
}
public void setMaxPoolSize(int maxPoolSize) {
this.maxPoolSize = maxPoolSize;
}
public int getQueueSize() {
return queueSize;
}
public void setQueueSize(int queueSize) {
this.queueSize = queueSize;
}
public int getKeepAliveSeconds() {
return keepAliveSeconds;
}
public void setKeepAliveSeconds(int keepAliveSeconds) {
this.keepAliveSeconds = keepAliveSeconds;
}
}

71
spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalMessageChannelBinderConfiguration.java
View File

@@ -1,71 +0,0 @@
/*
* Copyright 2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.local.config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.local.LocalMessageChannelBinder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.scheduling.support.PeriodicTrigger;
/**
* @author David Turanski
*/
@Configuration
@ConfigurationProperties(prefix = "spring.cloud.stream.binder.local")
@EnableConfigurationProperties(LocalExecutorConfigurationProperties.class)
public class LocalMessageChannelBinderConfiguration {
private int queueSize = Integer.MAX_VALUE;
private int polling;
@Autowired
LocalExecutorConfigurationProperties localExecutorConfigurationProperties;
@Bean
public LocalMessageChannelBinder localMessageChannelBinder() {
LocalMessageChannelBinder localMessageChannelBinder = new LocalMessageChannelBinder();
localMessageChannelBinder.setExecutorCorePoolSize(localExecutorConfigurationProperties.getCorePoolSize());
localMessageChannelBinder.setExecutorKeepAliveSeconds(localExecutorConfigurationProperties.getKeepAliveSeconds());
localMessageChannelBinder.setExecutorMaxPoolSize(localExecutorConfigurationProperties.getMaxPoolSize());
localMessageChannelBinder.setExecutorQueueSize(localExecutorConfigurationProperties.getQueueSize());
if (polling > 0) {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(new PeriodicTrigger(polling));
localMessageChannelBinder.setPoller(pollerMetadata);
}
localMessageChannelBinder.setQueueSize(queueSize);
return localMessageChannelBinder;
}
public void setQueueSize(int queueSize) {
this.queueSize = queueSize;
}
public void setPolling(int polling) {
this.polling = polling;
}
}

5
spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/resources/META-INF/spring-cloud-stream/local-binder.properties
View File

@@ -1,5 +0,0 @@
spring.cloud.stream.binder.local.polling: 1000
spring.cloud.stream.binder.local.executor.corePoolSize: 0
spring.cloud.stream.binder.local.executor.maxPoolSize: 400
#spring.cloud.stream.binder.local.executor.queueSize: # defaults to Integer.MAX_VALUE
spring.cloud.stream.binder.local.executor.keepAliveSeconds: 60

2
spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/resources/META-INF/spring.factories
View File

@@ -1,2 +0,0 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration:\
org.springframework.cloud.stream.binder.local.config.LocalBinderAutoConfiguration

195
spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/test/java/org/springframework/cloud/stream/binder/local/LocalBinderTests.java
View File

@@ -1,195 +0,0 @@
/*
* Copyright 2013-2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.local;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.cloud.stream.binder.AbstractBinderTests;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.http.MediaType;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.interceptor.WireTap;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
/**
* @author Gary Russell
* @author David Turanski
* @since 1.0
*/
public class LocalBinderTests extends AbstractBinderTests {
@Override
protected Binder<MessageChannel> getBinder() throws Exception {
LocalMessageChannelBinder binder = new LocalMessageChannelBinder();
GenericApplicationContext applicationContext = new GenericApplicationContext();
applicationContext.getBeanFactory().registerSingleton(
IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
new DefaultMessageBuilderFactory());
applicationContext.refresh();
binder.setApplicationContext(applicationContext);
binder.setExecutorCorePoolSize(2);
binder.setExecutorMaxPoolSize(10);
binder.setExecutorKeepAliveSeconds(59);
binder.setExecutorQueueSize(Integer.MAX_VALUE - 1);
binder.afterPropertiesSet();
return binder;
}
@Override
protected Collection<?> getBindings(Binder<MessageChannel> testBinder) {
return getBindingsFromBinder(testBinder);
}
@Test
public void testProps() throws Exception {
LocalMessageChannelBinder binder = (LocalMessageChannelBinder) getBinder();
ThreadPoolTaskExecutor exec = TestUtils.getPropertyValue(binder, "executor", ThreadPoolTaskExecutor.class);
assertEquals(2, exec.getCorePoolSize());
assertEquals(10, exec.getMaxPoolSize());
assertEquals(59, exec.getKeepAliveSeconds());
Assert.assertEquals(Integer.MAX_VALUE - 1, TestUtils.getPropertyValue(exec, "queueCapacity"));
}
@Test
public void testPayloadConversionNotNeededExplicitType() throws Exception {
LocalMessageChannelBinder binder = (LocalMessageChannelBinder) getBinder();
verifyPayloadConversion(new TestPayload(), binder);
}
@Test
public void testNoPayloadConversionByDefault() throws Exception {
LocalMessageChannelBinder binder = (LocalMessageChannelBinder) getBinder();
verifyPayloadConversion(new TestPayload(), binder);
}
@Test
public void testTapDoesntHurtStream() throws Exception {
LocalMessageChannelBinder binder = (LocalMessageChannelBinder) getBinder();
DirectChannel moduleOutputChannel = new DirectChannel();
moduleOutputChannel.setBeanName("bangOut");
DirectChannel tapChannel = new DirectChannel();
tapChannel.setBeanName("tapChannel");
WireTap tap = new WireTap(tapChannel);
moduleOutputChannel.addInterceptor(tap);
binder.bindProducer("bang.0", moduleOutputChannel, null);
final AtomicBoolean messageReceived = new AtomicBoolean();
final AtomicReference<Thread> streamThread = new AtomicReference<Thread>();
binder.bindConsumer("bang.0", new DirectChannel() {
@Override
protected boolean doSend(Message<?> message, long timeout) {
messageReceived.set(true);
streamThread.set(Thread.currentThread());
return true;
}
}, null);
final CountDownLatch tapped = new CountDownLatch(1);
final AtomicReference<Thread> tapThread = new AtomicReference<Thread>();
binder.bindPubSubProducer("tap:stream:bang.0", tapChannel, null);
binder.bindPubSubConsumer("tap:stream:bang.0", new DirectChannel() {
@Override
protected boolean doSend(Message<?> message, long timeout) {
tapThread.set(Thread.currentThread());
tapped.countDown();
throw new RuntimeException("bang");
}
}, null, null);
moduleOutputChannel.send(new GenericMessage<String>("Foo"));
assertTrue(tapped.await(10, TimeUnit.SECONDS));
assertTrue(messageReceived.get());
assertSame(Thread.currentThread(), streamThread.get());
assertNotNull(tapThread.get());
assertNotSame(Thread.currentThread(), tapThread.get());
}
private void verifyPayloadConversion(final Object expectedValue, final LocalMessageChannelBinder binder) {
DirectChannel myChannel = new DirectChannel();
binder.bindConsumer("in", myChannel, null);
DirectChannel input = binder.getBean("in", DirectChannel.class);
assertNotNull(input);
final AtomicBoolean msgSent = new AtomicBoolean(false);
myChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
assertEquals(expectedValue, message.getPayload());
msgSent.set(true);
}
});
Message<TestPayload> msg = MessageBuilder.withPayload(new TestPayload())
.setHeader(MessageHeaders.CONTENT_TYPE, MediaType.ALL_VALUE).build();
input.send(msg);
assertTrue(msgSent.get());
}
@Override @Ignore // TODO
public void testSendAndReceivePubSub() throws Exception {
}
@Override @Ignore // TODO
public void createInboundPubSubBeforeOutboundPubSub() throws Exception {
}
static class TestPayload {
@Override
public String toString() {
return "foo";
}
@Override
public boolean equals(Object other) {
return (other instanceof TestPayload && this.toString().equals(other.toString()));
}
@Override
public int hashCode() {
return this.toString().hashCode();
}
}
}

2
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/pom.xml
View File

@@ -25,7 +25,7 @@
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-spi</artifactId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>

2
spring-cloud-stream-binders/spring-cloud-stream-binder-redis/pom.xml
View File

@@ -25,7 +25,7 @@
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-spi</artifactId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>

42
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/pom.xml
View File

@@ -1,42 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-stream-binder-spi</artifactId>
<packaging>jar</packaging>
<name>spring-cloud-stream-binder-spi</name>
<description>SPI for binder implementations</description>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binders-parent</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.retry</groupId>
<artifactId>spring-retry</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
</dependency>
</dependencies>
</project>

379
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderPropertiesAccessor.java
View File

@@ -1,379 +0,0 @@
/*
* Copyright 2014 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
import java.util.Properties;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.util.StringUtils;
/**
* Base class for binder-specific property accessors; common properties
* are defined here.
*
* @author Gary Russell
*/
public abstract class AbstractBinderPropertiesAccessor implements BinderProperties {
private static final SpelExpressionParser spelExpressionParser = new SpelExpressionParser();
private final Properties properties;
public AbstractBinderPropertiesAccessor(Properties properties) {
if (properties == null) {
this.properties = new Properties();
}
else {
this.properties = properties;
}
}
/**
* Return the underlying properties object.
* @return The properties.
*/
public Properties getProperties() {
return properties;
}
/**
* Return the property for the key, or null if it doesn't exist.
* @param key The property.
* @return The key.
*/
public String getProperty(String key) {
return this.properties.getProperty(key);
}
/**
* Return the property for the key, or the default value if the
* property doesn't exist.
* @param key The key.
* @param defaultValue The default value.
* @return The property or default value.
*/
public String getProperty(String key, String defaultValue) {
return this.properties.getProperty(key, defaultValue);
}
/**
* Return the property for the key, or the default value if the
* property doesn't exist.
* @param key The key.
* @param defaultValue The default value.
* @return The property or default value.
*/
public boolean getProperty(String key, boolean defaultValue) {
String property = this.properties.getProperty(key);
if (property != null) {
return Boolean.parseBoolean(property);
}
else {
return defaultValue;
}
}
/**
* Return the property for the key, or the default value if the
* property doesn't exist.
* @param key The key.
* @param defaultValue The default value.
* @return The property or default value.
*/
public int getProperty(String key, int defaultValue) {
String property = this.properties.getProperty(key);
if (property != null) {
return Integer.parseInt(property);
}
else {
return defaultValue;
}
}
/**
* Return the property for the key, or the default value if the
* property doesn't exist.
* @param key The key.
* @param defaultValue The default value.
* @return The property or default value.
*/
public long getProperty(String key, long defaultValue) {
String property = this.properties.getProperty(key);
if (property != null) {
return Long.parseLong(property);
}
else {
return defaultValue;
}
}
/**
* Return the property for the key, or the default value if the
* property doesn't exist.
* @param key The key.
* @param defaultValue The default value.
* @return The property or default value.
*/
public double getProperty(String key, double defaultValue) {
String property = properties.getProperty(key);
if (property != null) {
return Double.parseDouble(property);
}
else {
return defaultValue;
}
}
/**
* Return the 'concurrency' property or the default value.
* The meaning of concurrency depends on the binder implementation.
* @param defaultValue The default value.
* @return The property or default value.
*/
public int getConcurrency(int defaultValue) {
return getProperty(CONCURRENCY, defaultValue);
}
/**
* Return the 'maxConcurrency' property or the default value.
* The meaning of maxConcurrency depends on the binder implementation.
* @param defaultValue The default value.
* @return The property or default value.
*/
public int getMaxConcurrency(int defaultValue) {
return getProperty(MAX_CONCURRENCY, defaultValue);
}
// Retry properties
/**
* Return the 'maxAttempts' property or the default value.
* This is used in the retry template's SimpleRetryPolicy
* in binders that support retry.
* @param defaultValue The default value.
* @return The property or default value.
*/
public int getMaxAttempts(int defaultValue) {
return getProperty(MAX_ATTEMPTS, defaultValue);
}
/**
* Return the 'backOffInitialInterval' property or the default value.
* This is used in the retry template's ExponentialBackOffPolicy
* in binders that support retry.
* @param defaultValue The default value.
* @return The property or default value.
*/
public long getBackOffInitialInterval(long defaultValue) {
return getProperty(BACK_OFF_INITIAL_INTERVAL, defaultValue);
}
/**
* Return the 'backOffMultiplier' property or the default value.
* This is used in the retry template's ExponentialBackOffPolicy
* in binders that support retry.
* @param defaultValue The default value.
* @return The property or default value.
*/
public double getBackOffMultiplier(double defaultValue) {
return getProperty(BACK_OFF_MULTIPLIER, defaultValue);
}
/**
* Return the 'backOffMaxInterval' property or the default value.
* This is used in the retry template's ExponentialBackOffPolicy
* in binders that support retry.
* @param defaultValue The default value.
* @return The property or default value.
*/
public long getBackOffMaxInterval(long defaultValue) {
return getProperty(BACK_OFF_MAX_INTERVAL, defaultValue);
}
// Partitioning
/**
* A class name for extracting partition keys from messages.
* @return The class name,
*/
public String getPartitionKeyExtractorClass() {
return getProperty(PARTITION_KEY_EXTRACTOR_CLASS);
}
/**
* The expression to determine the partition key, evaluated against the
* message as the root object.
* @return The key.
*/
public Expression getPartitionKeyExpression() {
String partionKeyExpression = getProperty(PARTITION_KEY_EXPRESSION);
Expression expression = null;
if (partionKeyExpression != null) {
expression = spelExpressionParser.parseExpression(partionKeyExpression);
}
return expression;
}
/**
* A class name for calculating a partition from a key.
* @return The class name,
*/
public String getPartitionSelectorClass() {
return getProperty(PARTITION_SELECTOR_CLASS);
}
/**
* The expression evaluated against the partition key to determine
* the partition to which the message will be sent. The result should
* be an integer that will subsequently be mod'd with the module's
* partition count.
* @return The expression.
*/
public Expression getPartitionSelectorExpression() {
String partionSelectorExpression = getProperty(PARTITION_SELECTOR_EXPRESSION);
Expression expression = null;
if (partionSelectorExpression != null) {
expression = spelExpressionParser.parseExpression(partionSelectorExpression);
}
return expression;
}
/**
* The sequence number for this module.
*
* @return the sequence number.
*/
public int getSequence() {
return getProperty(SEQUENCE, 1);
}
/**
* The module count.
*
* @return the module count.
*/
public int getCount() {
return getProperty(COUNT, 1);
}
/**
* The next module count for non-sink modules
* @return the next module count
*/
public int getNextModuleCount() {
return getProperty(NEXT_MODULE_COUNT, 1);
}
/**
* The partition index that this consumer supports.
* @return The partition index.
*/
public int getPartitionIndex() {
return getProperty(PARTITION_INDEX, -1);
}
// Direct Binding
/**
* If true, the binder can attempt a direct binding.
*/
public boolean isDirectBindingAllowed() {
return getProperty(DIRECT_BINDING_ALLOWED, false);
}
// Batching
/**
* If true, enable batching.
* @param defaultValue the default value.
* @return the property or default value.
*/
public boolean isBatchingEnabled(boolean defaultValue) {
return getProperty(BATCHING_ENABLED, defaultValue);
}
/**
* The batch size.
* @param defaultValue the default value.
* @return the property or default value.
*/
public int getBatchSize(int defaultValue) {
return getProperty(BATCH_SIZE, defaultValue);
}
/**
* The batch buffer limit.
* @param defaultValue the default value.
* @return the property or default value.
*/
public int geteBatchBufferLimit(int defaultValue) {
return getProperty(BATCH_BUFFER_LIMIT, defaultValue);
}
/**
* The batch timeout.
* @param defaultValue the default value.
* @return the property or default value.
*/
public long getBatchTimeout(long defaultValue) {
return getProperty(BATCH_TIMEOUT, defaultValue);
}
/**
* If true, messages will be compressed.
* @param defaultValue the default value.
* @return the property or default value.
*/
public boolean isCompress(boolean defaultValue) {
return getProperty(COMPRESS, defaultValue);
}
/**
* If true, subscriptions to taps/topics will be durable.
* @param defaultValue the default value.
* @return the property or default value.
*/
public boolean isDurable(boolean defaultValue) {
return getProperty(DURABLE, defaultValue);
}
// Utility methods
/**
* Convert a comma-delimited String property to a String[] if
* present, or return the default value.
* @param value The property value.
* @param defaultValue The default value.
* @return The converted property or default value.
*/
protected String[] asStringArray(String value, String[] defaultValue) {
if (StringUtils.hasText(value)) {
return StringUtils.commaDelimitedListToStringArray(value);
}
else {
return defaultValue;
}
}
@Override
public String toString() {
return this.properties.toString();
}
}

140
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/Binder.java
View File

@@ -1,140 +0,0 @@
/*
* Copyright 2013-2014 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
import java.util.Properties;
/**
* A strategy interface used to bind a module interface to a logical name. The name is intended to identify a
* logical consumer or producer of messages. This may be a queue, a channel adapter, another message channel, a Spring
* bean, etc.
* @author Mark Fisher
* @author David Turanski
* @author Gary Russell
* @author Jennifer Hickey
* @author Ilayaperumal Gopinathan
* @since 1.0
*/
public interface Binder<T> {
/**
* Bind a message consumer on a p2p channel
* @param name the logical identity of the message source
* @param inboundBindTarget the module interface to be bound as a point to point consumer
* @param properties arbitrary String key/value pairs that will be used in the binding
*/
void bindConsumer(String name, T inboundBindTarget, Properties properties);
/**
* Bind a message consumer on a pub/sub channel
* @param name the logical identity of the message source
* @param inboundBindTarget the module interface to be bound as a pub/sub consumer
* @param group the consumer group to which this consumer belongs - subscriptions are shared among consumers
* in the same group
* @param properties arbitrary String key/value pairs that will be used in the binding
*/
void bindPubSubConsumer(final String name, T inboundBindTarget, String group, Properties properties);
/**
* Bind a message producer on a p2p channel.
* @param name the logical identity of the message target
* @param outboundBindTarget the module interface bound as a producer
* @param properties arbitrary String key/value pairs that will be used in the binding
*/
void bindProducer(String name, T outboundBindTarget, Properties properties);
/**
* Bind a message producer on a pub/sub channel.
* @param name the logical identity of the message target
* @param outboundBindTarget the module interface bound as a producer
* @param properties arbitrary String key/value pairs that will be used in the binding
*/
void bindPubSubProducer(final String name, T outboundBindTarget, Properties properties);
/**
* Unbind inbound module components and stop any active components that use the channel.
* @param name the channel name
*/
void unbindConsumers(String name);
/**
* Unbind inbound module components and stop any active components that use the channel
* with the supplied consumer group.
* @param name the channel name
* @param group the consumer group
*/
void unbindPubSubConsumers(String name, String group);
/**
* Unbind outbound module components and stop any active components that use the channel.
* @param name the channel name
*/
void unbindProducers(String name);
/**
* Unbind a specific p2p or pub/sub message consumer
* @param name The logical identify of a message source
* @param inboundBindTarget The module interface bound as a consumer
*/
void unbindConsumer(String name, T inboundBindTarget);
/**
* Unbind a specific p2p or pub/sub message producer
* @param name the logical identity of the message target
* @param outboundBindTarget the channel bound as a producer
*/
void unbindProducer(String name, T outboundBindTarget);
/**
* Bind a producer that expects async replies. To unbind, invoke unbindProducer() and unbindConsumer().
* @param name The name of the requestor.
* @param requests The interface used to send requests.
* @param replies The interface used to receive replies.
* @param properties arbitrary String key/value pairs that will be used in the binding.
*/
void bindRequestor(String name, T requests, T replies, Properties properties);
/**
* Bind a consumer that handles requests from a requestor and asynchronously sends replies. To unbind, invoke
* unbindProducer() and unbindConsumer().
* @param name The name of the requestor for which this replier will handle requests.
* @param requests The interface used to send requests.
* @param replies The interface used to receive replies.
* @param properties arbitrary String key/value pairs that will be used in the binding.
*/
void bindReplier(String name, T requests, T replies, Properties properties);
/**
* Create an object and bind a producer dynamically, creating the infrastructure
* required by the binder technology.
* @param name The name of the "queue:" channel.
* @param properties arbitrary String key/value pairs that will be used in the binding.
* @return The bound object.
*/
T bindDynamicProducer(String name, Properties properties);
/**
* Create an object and bind a producer dynamically, creating the infrastructure
* required by the binder technology to broadcast messages to consumers.
* @param name The name of the "topic:" channel.
* @param properties arbitrary String key/value pairs that will be used in the binding.
* @return The bound Object.
*/
T bindDynamicPubSubProducer(String name, Properties properties);
}

33
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/BinderException.java
View File

@@ -1,33 +0,0 @@
/*
* Copyright 2014 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
/**
* Exception thrown to indicate a binder error (most
* likely a configuration error).
*
* @author Gary Russell
*/
@SuppressWarnings("serial")
public class BinderException extends RuntimeException {
public BinderException(String message, Throwable cause) {
super(message, cause);
}
}

62
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/BinderHeaders.java
View File

@@ -1,62 +0,0 @@
/*
* Copyright 2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.messaging.MessageHeaders;
/**
* Spring Integration message headers for XD.
* @author Gary Russell
* @author David Turanski
*/
public final class BinderHeaders {
public static final String BINDER_REPLY_CHANNEL = "binderReplyChannel";
public static final String BINDER_HISTORY = "binderHistory";
/*
* no xd prefix for backwards compatibility
*/
public static final String BINDER_ORIGINAL_CONTENT_TYPE = "originalContentType";
/*
* no xd prefix for backwards compatibility
*/
public static final String REPLY_TO = "replyTo";
/**
* The headers that will be propagated, by default, by binder implementations
* that have no inherent header support (by embedding the headers in the payload).
*/
public static final String[] STANDARD_HEADERS = new String[] {
IntegrationMessageHeaderAccessor.CORRELATION_ID,
IntegrationMessageHeaderAccessor.SEQUENCE_SIZE,
IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER,
BINDER_REPLY_CHANNEL,
MessageHeaders.CONTENT_TYPE,
BINDER_ORIGINAL_CONTENT_TYPE,
REPLY_TO,
BINDER_HISTORY
};
private BinderHeaders() {
}
}

144
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/BinderProperties.java
View File

@@ -1,144 +0,0 @@
/*
* Copyright 2014 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
/**
* Common binder properties.
*
* @author Gary Russell
*/
public interface BinderProperties {
/**
* The retry back off initial interval.
*/
public static final String BACK_OFF_INITIAL_INTERVAL = "backOffInitialInterval";
/**
* The retry back off max interval.
*/
public static final String BACK_OFF_MAX_INTERVAL = "backOffMaxInterval";
/**
* The retry back off multiplier.
*/
public static final String BACK_OFF_MULTIPLIER = "backOffMultiplier";
/**
* The minimum number of concurrent deliveries.
*/
public static final String CONCURRENCY = "concurrency";
/**
* The maximum delivery attempts when a delivery fails.
*/
public static final String MAX_ATTEMPTS = "maxAttempts";
/**
* The maximum number of concurrent deliveries.
*/
public static final String MAX_CONCURRENCY = "maxConcurrency";
/**
* The sequence index of the module.
* In a partitioned stream, it is identical to the partition index.
*/
public static final String SEQUENCE = "sequence";
/**
* The number of consumers, i.e. module instances in the stream.
* In a partitioned stream, it is identical to the partition count.
*/
public static final String COUNT = "count";
/**
* The consumer's partition index.
*/
public static final String PARTITION_INDEX = "partitionIndex";
/**
* The partition key expression.
*/
public static final String PARTITION_KEY_EXPRESSION = "partitionKeyExpression";
/**
* The partition key class.
*/
public static final String PARTITION_KEY_EXTRACTOR_CLASS = "partitionKeyExtractorClass";
/**
* The partition selector class.
*/
public static final String PARTITION_SELECTOR_CLASS = "partitionSelectorClass";
/**
* The partition selector expression.
*/
public static final String PARTITION_SELECTOR_EXPRESSION = "partitionSelectorExpression";
/**
* If true, the binder will attempt to create a direct binding between the producer and consumer.
*/
public static final String DIRECT_BINDING_ALLOWED = "directBindingAllowed";
/**
* True if message batching is enabled.
*/
public static final String BATCHING_ENABLED = "batchingEnabled";
/**
* The batch size if batching is enabled.
*/
public static final String BATCH_SIZE = "batchSize";
/**
* The buffer limit if batching is enabled.
*/
public static final String BATCH_BUFFER_LIMIT = "batchBufferLimit";
/**
* The batch timeout if batching is enabled.
*/
public static final String BATCH_TIMEOUT = "batchTimeout";
/**
* For all non-terminal modules, the number of modules coming after this one, irrespective of partitioning.
*/
public static final String NEXT_MODULE_COUNT = "nextModuleCount";
/**
* For all non-terminal modules, the concurrency for module coming after this one.
*/
public static final String NEXT_MODULE_CONCURRENCY = "nextModuleConcurrency";
/**
* Compression enabled.
*/
public static final String COMPRESS = "compress";
/**
* Durable pub/sub consumer.
*/
public static final String DURABLE = "durableSubscription";
/**
* Minimum partition count, if the transport supports partitioning natively (e.g. Kafka)
*/
public static final String MIN_PARTITION_COUNT = "minPartitionCount";
}

59
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/BinderUtils.java
View File

@@ -1,59 +0,0 @@
/*
* Copyright 2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Binder utilities.
*
* @author Gary Russell
*/
public class BinderUtils {
/**
* The delimiter between a group and index when constructing a binder consumer/producer.
*/
public static final String GROUP_INDEX_DELIMITER = ".";
/**
* The prefix for the consumer/producer when creating a topic.
*/
public static final String TOPIC_CHANNEL_PREFIX = "topic:";
/**
* Determine whether the provided channel name represents a pub/sub channel (i.e. topic or tap).
* @param channelName name of the channel to check
* @return true if pub/sub.
*/
public static boolean isChannelPubSub(String channelName) {
Assert.isTrue(StringUtils.hasText(channelName), "Channel name should not be empty/null.");
return channelName.startsWith(TOPIC_CHANNEL_PREFIX);
}
/**
* Construct a name comprised of the group and name.
* @param name the name.
* @param group the group.
* @return the constructed name.
*/
public static String groupedName(String name, String group) {
return group == null ? name : group + BinderUtils.GROUP_INDEX_DELIMITER + name;
}
}

119
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/Binding.java
View File

@@ -1,119 +0,0 @@
/*
* Copyright 2013-2014 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
import org.springframework.context.Lifecycle;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
/**
* Represents a binding between a module's channel and an adapter endpoint that connects to the Binder. The binding
* could be for a consumer or a producer. A consumer binding represents a connection from an adapter on the binder to a
* module's input channel. A producer binding represents a connection from a module's output channel to an adapter on
* the binder.
*
* @author Jennifer Hickey
* @author Mark Fisher
* @author Gary Russell
*/
public class Binding implements Lifecycle {
public static final String PRODUCER = "producer";
public static final String CONSUMER = "consumer";
public static final String DIRECT = "direct";
private final String name;
private final MessageChannel channel;
private final AbstractEndpoint endpoint;
private final String type;
private final AbstractBinderPropertiesAccessor properties;
private Binding(String name, MessageChannel channel, AbstractEndpoint endpoint, String type,
AbstractBinderPropertiesAccessor properties) {
Assert.notNull(channel, "channel must not be null");
Assert.notNull(endpoint, "endpoint must not be null");
this.name = name;
this.channel = channel;
this.endpoint = endpoint;
this.type = type;
this.properties = properties;
}
public static Binding forConsumer(String name, AbstractEndpoint adapterFromBinder, MessageChannel moduleInputChannel,
AbstractBinderPropertiesAccessor properties) {
return new Binding(name, moduleInputChannel, adapterFromBinder, CONSUMER, properties);
}
public static Binding forProducer(String name, MessageChannel moduleOutputChannel, AbstractEndpoint adapterToBinder,
AbstractBinderPropertiesAccessor properties) {
return new Binding(name, moduleOutputChannel, adapterToBinder, PRODUCER, properties);
}
public static Binding forDirectProducer(String name, MessageChannel moduleOutputChannel,
AbstractEndpoint adapter, AbstractBinderPropertiesAccessor properties) {
return new Binding(name, moduleOutputChannel, adapter, DIRECT, properties);
}
public String getName() {
return name;
}
public MessageChannel getChannel() {
return channel;
}
public AbstractEndpoint getEndpoint() {
return endpoint;
}
public String getType() {
return type;
}
public AbstractBinderPropertiesAccessor getPropertiesAccessor() {
return properties;
}
@Override
public void start() {
endpoint.start();
}
@Override
public void stop() {
endpoint.stop();
}
@Override
public boolean isRunning() {
return endpoint.isRunning();
}
@Override
public String toString() {
return type + " Binding [name=" + name + ", channel=" + channel + ", endpoint=" + endpoint.getComponentName()
+ "]";
}
}

40
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/BindingCleaner.java
View File

@@ -1,40 +0,0 @@
/*
* Copyright 2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
import java.util.List;
import java.util.Map;
/**
* Interface for implementations that perform cleanup for binders.
*
* @author Gary Russell
* @since 1.2
*/
public interface BindingCleaner {
/**
* Clean up all resources for the supplied stream/job.
* @param entity the stream or job; may be terminated with a simple wild card '*', in which
* case all streams with names starting with the characters before the '*' will be cleaned.
* @param isJob true if the entity is a job.
* @return a map of lists of resources removed.
*/
Map<String, List<String>> clean(String entity, boolean isJob);
}

158
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/EmbeddedHeadersMessageConverter.java
View File

@@ -1,158 +0,0 @@
/*
* Copyright 2014-2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import javax.xml.bind.DatatypeConverter;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.support.json.Jackson2JsonObjectMapper;
import org.springframework.messaging.Message;
/**
* Encodes requested headers into payload with format
* {@code 0xff, n(1), [ [lenHdr(1), hdr, lenValue(4), value] ... ]}.
* The 0xff indicates this new format; n is number of headers (max 255); for
* each header, the name length (1 byte) is followed by the name, followed by
* the value length (int) followed by the value (json).
* <p>
* Previously, there was no leading 0xff; the value length was 1 byte and only
* String header values were supported (no JSON conversion).
*
* @author Eric Bottard
* @author Gary Russell
*/
public class EmbeddedHeadersMessageConverter {
private final Jackson2JsonObjectMapper objectMapper = new Jackson2JsonObjectMapper();
public static String decodeExceptionMessage(Message<?> requestMessage) {
return "Could not convert message: " + DatatypeConverter.printHexBinary((byte[]) requestMessage.getPayload());
}
/**
* Return a new message where some of the original headers of {@code original}
* have been embedded into the new message payload.
*/
public byte[] embedHeaders(MessageValues original, String... headers) throws Exception {
byte[][] headerValues = new byte[headers.length][];
int n = 0;
int headerCount = 0;
int headersLength = 0;
for (String header : headers) {
Object value = original.get(header) == null ? null
: original.get(header);
if (value != null) {
String json = this.objectMapper.toJson(value);
headerValues[n++] = json.getBytes("UTF-8");
headerCount++;
headersLength += header.length() + json.length();
}
else {
headerValues[n++] = null;
}
}
// 0xff, n(1), [ [lenHdr(1), hdr, lenValue(4), value] ... ]
byte[] newPayload = new byte[((byte[])original.getPayload()).length + headersLength + headerCount * 5 + 2];
ByteBuffer byteBuffer = ByteBuffer.wrap(newPayload);
byteBuffer.put((byte) 0xff); // signal new format
byteBuffer.put((byte) headerCount);
for (int i = 0; i < headers.length; i++) {
if (headerValues[i] != null) {
byteBuffer.put((byte) headers[i].length());
byteBuffer.put(headers[i].getBytes("UTF-8"));
byteBuffer.putInt(headerValues[i].length);
byteBuffer.put(headerValues[i]);
}
}
byteBuffer.put((byte[])original.getPayload());
return byteBuffer.array();
}
/**
* Return a message where headers, that were originally embedded into the payload, have been promoted
* back to actual headers. The new payload is now the original payload.
*
* @param message the message to extract headers
* @param copyRequestHeaders boolean value to specify if original headers should be copied
*/
public MessageValues extractHeaders(Message<byte[]> message, boolean copyRequestHeaders) throws Exception {
byte[] bytes = message.getPayload();
ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
int headerCount = byteBuffer.get() & 0xff;
if (headerCount < 255) {
return oldExtractHeaders(byteBuffer, bytes, headerCount, message, copyRequestHeaders);
}
else {
headerCount = byteBuffer.get() & 0xff;
Map<String, Object> headers = new HashMap<String, Object>();
for (int i = 0; i < headerCount; i++) {
int len = byteBuffer.get() & 0xff;
String headerName = new String(bytes, byteBuffer.position(), len, "UTF-8");
byteBuffer.position(byteBuffer.position() + len);
len = byteBuffer.getInt();
String headerValue = new String(bytes, byteBuffer.position(), len, "UTF-8");
Object headerContent = this.objectMapper.fromJson(headerValue, Object.class);
headers.put(headerName, headerContent);
byteBuffer.position(byteBuffer.position() + len);
}
byte[] newPayload = new byte[byteBuffer.remaining()];
byteBuffer.get(newPayload);
return buildMessageValues(message, newPayload, headers, copyRequestHeaders);
}
}
private MessageValues oldExtractHeaders(ByteBuffer byteBuffer, byte[] bytes, int headerCount,
Message<byte[]> message, boolean copyRequestHeaders)
throws UnsupportedEncodingException {
Map<String, Object> headers = new HashMap<String, Object>();
for (int i = 0; i < headerCount; i++) {
int len = byteBuffer.get();
String headerName = new String(bytes, byteBuffer.position(), len, "UTF-8");
byteBuffer.position(byteBuffer.position() + len);
len = byteBuffer.get() & 0xff;
String headerValue = new String(bytes, byteBuffer.position(), len, "UTF-8");
byteBuffer.position(byteBuffer.position() + len);
if (IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER.equals(headerName)
|| IntegrationMessageHeaderAccessor.SEQUENCE_SIZE.equals(headerName)) {
headers.put(headerName, Integer.parseInt(headerValue));
}
else {
headers.put(headerName, headerValue);
}
}
byte[] newPayload = new byte[byteBuffer.remaining()];
byteBuffer.get(newPayload);
return buildMessageValues(message, newPayload, headers, copyRequestHeaders);
}
private MessageValues buildMessageValues(Message<byte[]> message, byte[] payload, Map<String, Object> headers,
boolean copyRequestHeaders) {
MessageValues messageValues = new MessageValues(payload, headers);
if (copyRequestHeaders) {
messageValues.copyHeadersIfAbsent(message.getHeaders());
}
return messageValues;
}
}

1140
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java
View File

File diff suppressed because it is too large Load Diff

156
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/MessageValues.java
View File

@@ -1,156 +0,0 @@
/*
* Copyright 2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
* A mutable type for allowing {@link Binder} implementations to transform and enrich message content more
* efficiently.
* @author David Turanski
*/
public class MessageValues implements Map<String, Object> {
private Map<String, Object> headers = new HashMap<>();
private Object payload;
/**
* Create an instance from a {@link Message}.
* @param message the message
*/
public MessageValues(Message<?> message) {
this.payload = message.getPayload();
for (Map.Entry<String, Object> header : message.getHeaders().entrySet()) {
this.headers.put(header.getKey(), header.getValue());
}
}
public MessageValues(Object payload, Map<String, Object> headers) {
this.payload = payload;
this.headers.putAll(headers);
}
/**
* @return the payload
*/
public Object getPayload() {
return payload;
}
/**
* Convert to a {@link Message} using a {@link org.springframework.integration.support.MessageBuilderFactory}.
* @param messageBuilderFactory the MessageBuilderFactory
* @return the Message
*/
public Message<?> toMessage(MessageBuilderFactory messageBuilderFactory) {
return messageBuilderFactory.withPayload(this.payload).copyHeaders(this.headers).build();
}
/**
* Convert to a {@link Message} using a the default {@link org.springframework.integration.support.MessageBuilder}.
* @return the Message
*/
public Message<?> toMessage() {
return MessageBuilder.withPayload(this.payload).copyHeaders(this.headers).build();
}
/**
* Set the payload
* @param payload any non null object.
*/
public void setPayload(Object payload) {
Assert.notNull(payload, "'payload' cannot be null");
this.payload = payload;
}
@Override
public int size() {
return headers.size();
}
@Override
public boolean isEmpty() {
return headers.isEmpty();
}
@Override
public boolean containsKey(Object key) {
return headers.containsKey(key);
}
@Override
public boolean containsValue(Object value) {
return headers.containsValue(value);
}
@Override
public Object get(Object key) {
return headers.get(key);
}
@Override
public Object put(String key, Object value) {
return headers.put(key, value);
}
@Override
public Object remove(Object key) {
return headers.remove(key);
}
@Override
public void putAll(Map<? extends String, ?> m) {
headers.putAll(m);
}
@Override
public void clear() {
headers.clear();
}
@Override
public Set<String> keySet() {
return headers.keySet();
}
@Override
public Collection<Object> values() {
return headers.values();
}
@Override
public Set<Entry<String, Object>> entrySet() {
return headers.entrySet();
}
public void copyHeadersIfAbsent(Map<String, Object> headersToCopy) {
for (Entry<String, Object> headersToCopyEntry : headersToCopy.entrySet()) {
if (!containsKey(headersToCopyEntry.getKey())) {
put(headersToCopyEntry.getKey(), headersToCopyEntry.getValue());
}
}
}
}

31
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/PartitionKeyExtractorStrategy.java
View File

@@ -1,31 +0,0 @@
/*
* Copyright 2014 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
import org.springframework.messaging.Message;
/**
* Strategy for extracting a partition key from a Message.
*
* @author Gary Russell
*/
public interface PartitionKeyExtractorStrategy {
Object extractKey(Message<?> message);
}

41
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/PartitionSelectorStrategy.java
View File

@@ -1,41 +0,0 @@
/*
* Copyright 2014 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
/**
* Strategy for determining the partition to which a message should be sent.
*
* @author Gary Russell
*/
public interface PartitionSelectorStrategy {
/**
* Determine the partition based on a key. The partitionCount is 1 greater
* than the maximum value of a valid partition. Typical implementations
* will return {@code someValue % partitionCount}. The caller will apply
* that same modulo operation (as well as enforcing absolute value) if the
* value exceeds partitionCount - 1.
*
* @param key the key
* @param partitionCount the number of partitions
*
* @return the partition
*/
int selectPartition(Object key, int partitionCount);
}

57
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/StringConvertingContentTypeResolver.java
View File

@@ -1,57 +0,0 @@
/*
* Copyright 2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.DefaultContentTypeResolver;
import org.springframework.util.MimeType;
/**
* A {@link DefaultContentTypeResolver} that can parse String values.
*
* @author David Turanski
*/
public class StringConvertingContentTypeResolver extends DefaultContentTypeResolver {
private ConcurrentMap<String,MimeType> mimeTypeCache = new ConcurrentHashMap<>();
@Override
public MimeType resolve(MessageHeaders headers) {
return resolve((Map<String, Object>) headers);
}
public MimeType resolve(Map<String,Object> headers) {
Object value = headers.get(MessageHeaders.CONTENT_TYPE);
if (value instanceof MimeType) {
return (MimeType) value;
}
else if (value instanceof String) {
MimeType mimeType = mimeTypeCache.get(value);
if (mimeType == null) {
String valueAsString = (String) value;
mimeType = MimeType.valueOf(valueAsString);
mimeTypeCache.put(valueAsString,mimeType);
}
return mimeType;
}
return getDefaultMimeType();
}
}

96
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/test/java/org/springframework/cloud/stream/binder/MessageConverterTests.java
View File

@@ -1,96 +0,0 @@
/*
* Copyright 2002-2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.startsWith;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
/**
* @author Gary Russell
* @since 1.0
*
*/
public class MessageConverterTests {
@Test
public void testHeaderEmbedding() throws Exception {
EmbeddedHeadersMessageConverter converter = new EmbeddedHeadersMessageConverter();
Message<byte[]> message = MessageBuilder.withPayload("Hello".getBytes())
.setHeader("foo", "bar")
.setHeader("baz", "quxx")
.build();
byte[] embedded = converter.embedHeaders(new MessageValues(message), "foo", "baz");
assertEquals(0xff, embedded[0] & 0xff);
assertEquals("\u0002\u0003foo\u0000\u0000\u0000\u0005\"bar\"\u0003baz\u0000\u0000\u0000\u0006\"quxx\"Hello",
new String(embedded).substring(1));
MessageValues extracted = converter.extractHeaders(MessageBuilder.withPayload(embedded).build(), false);
assertEquals("Hello", new String((byte[])extracted.getPayload()));
assertEquals("bar", extracted.get("foo"));
assertEquals("quxx", extracted.get("baz"));
}
@Test
public void testHeaderEmbeddingMissingHeader() throws Exception {
EmbeddedHeadersMessageConverter converter = new EmbeddedHeadersMessageConverter();
Message<byte[]> message = MessageBuilder.withPayload("Hello".getBytes())
.setHeader("foo", "bar")
.build();
byte[] embedded = converter.embedHeaders(new MessageValues(message), "foo", "baz");
assertEquals(0xff, embedded[0] & 0xff);
assertEquals("\u0001\u0003foo\u0000\u0000\u0000\u0005\"bar\"Hello",
new String(embedded).substring(1));
}
@Test
public void testCanDecodeOldFormat() throws Exception {
EmbeddedHeadersMessageConverter converter = new EmbeddedHeadersMessageConverter();
byte[] bytes = "\u0002\u0003foo\u0003bar\u0003baz\u0004quxxHello".getBytes("UTF-8");
Message<byte[]> message = new GenericMessage<byte[]>(bytes);
MessageValues extracted = converter.extractHeaders(message,false);
assertEquals("Hello", new String((byte[])extracted.getPayload()));
assertEquals("bar", extracted.get("foo"));
assertEquals("quxx", extracted.get("baz"));
}
@Test
public void testBadDecode() throws Exception {
EmbeddedHeadersMessageConverter converter = new EmbeddedHeadersMessageConverter();
byte[] bytes = "\u0002\u0003foo\u0020bar\u0003baz\u0004quxxHello".getBytes("UTF-8");
Message<byte[]> message = new GenericMessage<byte[]>(bytes);
try {
converter.extractHeaders(message,false);
Assert.fail("Exception expected");
}
catch (Exception e) {
String s = EmbeddedHeadersMessageConverter.decodeExceptionMessage(message);
assertThat(e, instanceOf(StringIndexOutOfBoundsException.class));
assertThat(s, startsWith("Could not convert message: 0203666F6F"));
}
}
}

2
spring-cloud-stream-binders/spring-cloud-stream-binder-test/pom.xml
View File

@@ -41,7 +41,7 @@
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-spi</artifactId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
</dependencies>
</project>