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

Refactored Binder and Codec configuration

Browse Source 
- replaced XML-based configuration with Java configuration
- removed optional on spring-cloud-stream-codec
- removed unnecessary test
This commit is contained in:
David Turanski
2015-08-05 09:07:34 -04:00
committed by Marius Bogoevici
parent 31daf48cc2
commit 0a84afd6ea
60 changed files with 1331 additions and 1520 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -26,6 +26,15 @@
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-spi</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-codec</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>
@@ -63,12 +72,6 @@
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-tuple</artifactId>
<version>${spring-xd.version}</version>
<exclusions>
<exclusion>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-codec</artifactId>
</exclusion>
</exclusions>
<scope>test</scope>
</dependency>
<dependency>

4
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -85,7 +85,6 @@ import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
import org.springframework.cloud.stream.binder.MessageValues;
import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
import scala.collection.Seq;
@@ -293,11 +292,10 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport {
private Mode mode = Mode.embeddedHeaders;
public KafkaMessageChannelBinder(ZookeeperConnect zookeeperConnect, String brokers, String zkAddress,
MultiTypeCodec<Object> codec, String... headersToMap) {
String... headersToMap) {
this.zookeeperConnect = zookeeperConnect;
this.brokers = brokers;
this.zkAddress = zkAddress;
setCodec(codec);
if (headersToMap.length > 0) {
String[] combinedHeadersToMap =
Arrays.copyOfRange(BinderHeaders.STANDARD_HEADERS, 0, BinderHeaders.STANDARD_HEADERS.length + headersToMap

125
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java Normal file
View File

@@ -0,0 +1,125 @@
/*
* 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.kafka.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* @author David Turanski
*/
@ConfigurationProperties(prefix = "spring.cloud.stream.binder.kafka.default")
class KafkaBinderConfigurationProperties {
private int batchSize;
private long batchTimeout;
private int requiredAcks;
private int replicationFactor;
private int concurrency;
private String compressionCodec;
private boolean autoCommitEnabled;
private int fetchSize;
private int minPartitionCount;
private int queueSize;
public int getBatchSize() {
return batchSize;
}
public void setBatchSize(int batchSize) {
this.batchSize = batchSize;
}
public long getBatchTimeout() {
return batchTimeout;
}
public void setBatchTimeout(long batchTimeout) {
this.batchTimeout = batchTimeout;
}
public int getRequiredAcks() {
return requiredAcks;
}
public void setRequiredAcks(int requiredAcks) {
this.requiredAcks = requiredAcks;
}
public int getReplicationFactor() {
return replicationFactor;
}
public void setReplicationFactor(int replicationFactor) {
this.replicationFactor = replicationFactor;
}
public int getConcurrency() {
return concurrency;
}
public void setConcurrency(int concurrency) {
this.concurrency = concurrency;
}
public String getCompressionCodec() {
return compressionCodec;
}
public void setCompressionCodec(String compressionCodec) {
this.compressionCodec = compressionCodec;
}
public boolean isAutoCommitEnabled() {
return autoCommitEnabled;
}
public void setAutoCommitEnabled(boolean autoCommitEnabled) {
this.autoCommitEnabled = autoCommitEnabled;
}
public int getFetchSize() {
return fetchSize;
}
public void setFetchSize(int fetchSize) {
this.fetchSize = fetchSize;
}
public int getMinPartitionCount() {
return minPartitionCount;
}
public void setMinPartitionCount(int minPartitionCount) {
this.minPartitionCount = minPartitionCount;
}
public int getQueueSize() {
return queueSize;
}
public void setQueueSize(int queueSize) {
this.queueSize = queueSize;
}
}

166
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaMessageChannelBinderConfiguration.java Normal file
View File

@@ -0,0 +1,166 @@
/*
* 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.kafka.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.kafka.KafkaMessageChannelBinder;
import org.springframework.cloud.stream.config.codec.CodecConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.integration.codec.Codec;
import org.springframework.integration.kafka.support.ZookeeperConnect;
/**
* @author David Turanski
*/
@Configuration
@Import(CodecConfiguration.class)
@EnableConfigurationProperties(KafkaBinderConfigurationProperties.class)
@ConfigurationProperties(prefix = "spring.cloud.stream.binder.kafka")
public class KafkaMessageChannelBinderConfiguration {
private String zkAddress;
private String brokers;
private KafkaMessageChannelBinder.Mode mode;
private String offsetStoreTopic;
private int offsetStoreSegmentSize;
private int offsetStoreRetentionTime;
private int offsetStoreRequiredAcks;
private int offsetStoreMaxFetchSize;
private int offsetStoreBatchBytes;
private int offsetStoreBatchTime;
private int offsetUpdateTimeWindow;
private int offsetUpdateCount;
private int offsetUpdateShutdownTimeout;
@Autowired
private Codec codec;
@Autowired
private KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties;
@Bean
ZookeeperConnect zookeeperConnect() {
ZookeeperConnect zookeeperConnect = new ZookeeperConnect();
zookeeperConnect.setZkConnect(zkAddress);
return zookeeperConnect;
}
@Bean
KafkaMessageChannelBinder kafkaMessageChannelBinder() {
KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder(zookeeperConnect(),
brokers, zkAddress, new String[0]);
kafkaMessageChannelBinder.setCodec(codec);
kafkaMessageChannelBinder.setMode(mode);
kafkaMessageChannelBinder.setOffsetStoreTopic(offsetStoreTopic);
kafkaMessageChannelBinder.setOffsetStoreSegmentSize(offsetStoreSegmentSize);
kafkaMessageChannelBinder.setOffsetStoreRetentionTime(offsetStoreRetentionTime);
kafkaMessageChannelBinder.setOffsetStoreRequiredAcks(offsetStoreRequiredAcks);
kafkaMessageChannelBinder.setOffsetStoreMaxFetchSize(offsetStoreMaxFetchSize);
kafkaMessageChannelBinder.setOffsetStoreBatchBytes(offsetStoreBatchBytes);
kafkaMessageChannelBinder.setOffsetStoreBatchTime(offsetStoreBatchTime);
kafkaMessageChannelBinder.setOffsetUpdateTimeWindow(offsetUpdateTimeWindow);
kafkaMessageChannelBinder.setOffsetUpdateCount(offsetUpdateCount);
kafkaMessageChannelBinder.setOffsetUpdateShutdownTimeout(offsetUpdateShutdownTimeout);
kafkaMessageChannelBinder.setDefaultAutoCommitEnabled(kafkaBinderConfigurationProperties.isAutoCommitEnabled());
kafkaMessageChannelBinder.setDefaultBatchSize(kafkaBinderConfigurationProperties.getBatchSize());
kafkaMessageChannelBinder.setDefaultBatchTimeout(kafkaBinderConfigurationProperties.getBatchTimeout());
kafkaMessageChannelBinder.setDefaultCompressionCodec(kafkaBinderConfigurationProperties
.getCompressionCodec());
kafkaMessageChannelBinder.setDefaultConcurrency(kafkaBinderConfigurationProperties.getConcurrency());
kafkaMessageChannelBinder.setDefaultFetchSize(kafkaBinderConfigurationProperties.getFetchSize());
kafkaMessageChannelBinder.setDefaultMinPartitionCount(kafkaBinderConfigurationProperties
.getMinPartitionCount());
kafkaMessageChannelBinder.setDefaultQueueSize(kafkaBinderConfigurationProperties.getQueueSize());
kafkaMessageChannelBinder.setDefaultReplicationFactor(kafkaBinderConfigurationProperties.getReplicationFactor());
kafkaMessageChannelBinder.setDefaultRequiredAcks(kafkaBinderConfigurationProperties.getRequiredAcks());
return kafkaMessageChannelBinder;
}
public void setZkAddress(String zkAddress) {
this.zkAddress = zkAddress;
}
public void setBrokers(String brokers) {
this.brokers = brokers;
}
public void setMode(KafkaMessageChannelBinder.Mode mode) {
this.mode = mode;
}
public void setOffsetStoreTopic(String offsetStoreTopic) {
this.offsetStoreTopic = offsetStoreTopic;
}
public void setOffsetStoreSegmentSize(int offsetStoreSegmentSize) {
this.offsetStoreSegmentSize = offsetStoreSegmentSize;
}
public void setOffsetStoreRetentionTime(int offsetStoreRetentionTime) {
this.offsetStoreRetentionTime = offsetStoreRetentionTime;
}
public void setOffsetStoreRequiredAcks(int offsetStoreRequiredAcks) {
this.offsetStoreRequiredAcks = offsetStoreRequiredAcks;
}
public void setOffsetStoreMaxFetchSize(int offsetStoreMaxFetchSize) {
this.offsetStoreMaxFetchSize = offsetStoreMaxFetchSize;
}
public void setOffsetStoreBatchBytes(int offsetStoreBatchBytes) {
this.offsetStoreBatchBytes = offsetStoreBatchBytes;
}
public void setOffsetStoreBatchTime(int offsetStoreBatchTime) {
this.offsetStoreBatchTime = offsetStoreBatchTime;
}
public void setOffsetUpdateTimeWindow(int offsetUpdateTimeWindow) {
this.offsetUpdateTimeWindow = offsetUpdateTimeWindow;
}
public void setOffsetUpdateCount(int offsetUpdateCount) {
this.offsetUpdateCount = offsetUpdateCount;
}
public void setOffsetUpdateShutdownTimeout(int offsetUpdateShutdownTimeout) {
this.offsetUpdateShutdownTimeout = offsetUpdateShutdownTimeout;
}
public void setCodec(Codec codec) {
this.codec = codec;
}
}

49
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring-cloud-stream/binder/kafka-binder.xml
View File

@@ -1,49 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<bean id="zookeeperConnect" class="org.springframework.integration.kafka.support.ZookeeperConnect">
<property name="zkConnect" value="${xd.messagebus.kafka.zkAddress}"/>
</bean>
<bean id="messageBus" class="org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder">
<constructor-arg ref="zookeeperConnect"/>
<constructor-arg value="${xd.messagebus.kafka.brokers}"/>
<constructor-arg value="${xd.messagebus.kafka.zkAddress}"/>
<constructor-arg ref="codec"/>
<constructor-arg value="#{new String[0]}"/>
<property name="mode" value="${xd.messagebus.kafka.mode}"/>
<!-- Producer properties -->
<property name="defaultBatchSize" value="${xd.messagebus.kafka.default.batchSize}"/>
<property name="defaultBatchTimeout" value="${xd.messagebus.kafka.default.batchTimeout}"/>
<property name="defaultRequiredAcks" value="${xd.messagebus.kafka.default.requiredAcks}"/>
<property name="defaultReplicationFactor" value="${xd.messagebus.kafka.default.replicationFactor}"/>
<property name="defaultConcurrency" value="${xd.messagebus.kafka.default.concurrency}"/>
<property name="defaultCompressionCodec" value="${xd.messagebus.kafka.default.compressionCodec}"/>
<!-- ConsumerProperties -->
<property name="defaultAutoCommitEnabled" value="${xd.messagebus.kafka.default.autoCommitEnabled}"/>
<property name="defaultFetchSize" value="${xd.messagebus.kafka.default.fetchSize}"/>
<property name="defaultMinPartitionCount" value="${xd.messagebus.kafka.default.minPartitionCount}"/>
<property name="defaultQueueSize" value="${xd.messagebus.kafka.default.queueSize}"/>
<!-- Offset Manager Properties-->
<property name="offsetStoreTopic" value="${xd.messagebus.kafka.offsetStoreTopic}"/>
<property name="offsetStoreSegmentSize" value="${xd.messagebus.kafka.offsetStoreSegmentSize}"/>
<property name="offsetStoreRetentionTime" value="${xd.messagebus.kafka.offsetStoreRetentionTime}"/>
<property name="offsetStoreRequiredAcks" value="${xd.messagebus.kafka.offsetStoreRequiredAcks}"/>
<property name="offsetStoreMaxFetchSize" value="${xd.messagebus.kafka.offsetStoreMaxFetchSize}"/>
<property name="offsetStoreBatchBytes" value="${xd.messagebus.kafka.offsetStoreBatchBytes}"/>
<property name="offsetStoreBatchTime" value="${xd.messagebus.kafka.offsetStoreBatchTime}"/>
<property name="offsetUpdateTimeWindow" value="${xd.messagebus.kafka.offsetUpdateTimeWindow}"/>
<property name="offsetUpdateCount" value="${xd.messagebus.kafka.offsetUpdateCount}"/>
<property name="offsetUpdateShutdownTimeout" value="${xd.messagebus.kafka.offsetUpdateShutdownTimeout}"/>
</bean>
</beans>

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

@@ -77,7 +77,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests {
}
protected KafkaTestBinder createKafkaTestBinder() {
return new KafkaTestBinder(kafkaTestSupport, getCodec(), KafkaMessageChannelBinder.Mode.embeddedHeaders);
return new KafkaTestBinder(kafkaTestSupport, KafkaMessageChannelBinder.Mode.embeddedHeaders);
}
@Override

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

@@ -16,11 +16,17 @@
package org.springframework.cloud.stream.binder.kafka;
import java.util.List;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Registration;
import org.springframework.cloud.stream.binder.AbstractTestBinder;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.codec.Codec;
import org.springframework.integration.codec.kryo.KryoRegistrar;
import org.springframework.integration.codec.kryo.PojoCodec;
import org.springframework.integration.kafka.support.ZookeeperConnect;
import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
import org.springframework.xd.dirt.integration.bus.serializer.kryo.PojoCodec;
import org.springframework.xd.tuple.serializer.kryo.TupleKryoRegistrar;
@@ -34,11 +40,11 @@ import org.springframework.xd.tuple.serializer.kryo.TupleKryoRegistrar;
public class KafkaTestBinder extends AbstractTestBinder<KafkaMessageChannelBinder> {
public KafkaTestBinder(KafkaTestSupport kafkaTestSupport) {
this(kafkaTestSupport, getCodec(), KafkaMessageChannelBinder.Mode.embeddedHeaders);
this(kafkaTestSupport, KafkaMessageChannelBinder.Mode.embeddedHeaders);
}
public KafkaTestBinder(KafkaTestSupport kafkaTestSupport, MultiTypeCodec<Object> codec,
public KafkaTestBinder(KafkaTestSupport kafkaTestSupport,
KafkaMessageChannelBinder.Mode mode) {
try {
@@ -46,7 +52,8 @@ public class KafkaTestBinder extends AbstractTestBinder<KafkaMessageChannelBinde
zookeeperConnect.setZkConnect(kafkaTestSupport.getZkConnectString());
KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(zookeeperConnect,
kafkaTestSupport.getBrokerAddress(),
kafkaTestSupport.getZkConnectString(), codec);
kafkaTestSupport.getZkConnectString());
binder.setCodec(getCodec());
binder.setDefaultBatchingEnabled(false);
binder.setMode(mode);
binder.afterPropertiesSet();
@@ -65,9 +72,23 @@ public class KafkaTestBinder extends AbstractTestBinder<KafkaMessageChannelBinde
// do nothing - the rule will take care of that
}
@SuppressWarnings({"unchecked", "rawtypes"})
private static MultiTypeCodec<Object> getCodec() {
return new PojoCodec(new TupleKryoRegistrar());
private static Codec getCodec() {
return new PojoCodec(new TupleRegistrar());
}
//TODO: temporary wrapper for compatibility with SI Codec types
private static class TupleRegistrar implements KryoRegistrar {
private TupleKryoRegistrar delegate = new TupleKryoRegistrar();
@Override
public void registerTypes(Kryo kryo) {
delegate.registerTypes(kryo);
}
@Override
public List<Registration> getRegistrations() {
return delegate.getRegistrations();
}
}
}

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

@@ -49,13 +49,14 @@ import org.springframework.cloud.stream.binder.TestUtils;
/**
* @author Marius Bogoevici
* @author David Turanski
*/
public class RawModeKafkaBinderTests extends KafkaBinderTests {
@Override
protected KafkaTestBinder createKafkaTestBinder() {
return new KafkaTestBinder(kafkaTestSupport, getCodec(), KafkaMessageChannelBinder.Mode.raw);
return new KafkaTestBinder(kafkaTestSupport, KafkaMessageChannelBinder.Mode.raw);
}
@Test

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

@@ -23,9 +23,15 @@
<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>

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

@@ -0,0 +1,65 @@
/*
* 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 executorCorePoolSize;
private int executorMaxPoolSize;
private int executorQueueSize = Integer.MAX_VALUE;
private int executorKeepAliveSeconds;
public int getExecutorCorePoolSize() {
return executorCorePoolSize;
}
public void setExecutorCorePoolSize(int executorCorePoolSize) {
this.executorCorePoolSize = executorCorePoolSize;
}
public int getExecutorMaxPoolSize() {
return executorMaxPoolSize;
}
public void setExecutorMaxPoolSize(int executorMaxPoolSize) {
this.executorMaxPoolSize = executorMaxPoolSize;
}
public int getExecutorQueueSize() {
return executorQueueSize;
}
public void setExecutorQueueSize(int executorQueueSize) {
this.executorQueueSize = executorQueueSize;
}
public int getExecutorKeepAliveSeconds() {
return executorKeepAliveSeconds;
}
public void setExecutorKeepAliveSeconds(int executorKeepAliveSeconds) {
this.executorKeepAliveSeconds = executorKeepAliveSeconds;
}
}

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

@@ -0,0 +1,70 @@
/*
* 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.getExecutorCorePoolSize());
localMessageChannelBinder.setExecutorKeepAliveSeconds(localExecutorConfigurationProperties.getExecutorKeepAliveSeconds());
localMessageChannelBinder.setExecutorMaxPoolSize(localExecutorConfigurationProperties.getExecutorMaxPoolSize());
localMessageChannelBinder.setExecutorQueueSize(localExecutorConfigurationProperties.getExecutorQueueSize());
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;
}
}

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

@@ -1,18 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<bean id="messageBus" class="org.springframework.cloud.stream.binder.local.LocalMessageChannelBinder">
<property name="queueSize"
value="${xd.messagebus.local.queueSize: #{T(Integer).MAX_VALUE}}" />
<property name="poller">
<int:poller fixed-rate="${xd.messagebus.local.polling}" />
</property>
<property name="executorCorePoolSize" value="${xd.messagebus.local.executor.corePoolSize}" />
<property name="executorMaxPoolSize" value="${xd.messagebus.local.executor.maxPoolSize}" />
<property name="executorQueueSize" value="${xd.messagebus.local.executor.queueSize: #{T(Integer).MAX_VALUE}}" />
<property name="executorKeepAliveSeconds" value="${xd.messagebus.local.executor.keepAliveSeconds}" />
</bean>
</beans>

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

@@ -1,40 +1,49 @@
<?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>
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-rabbit</artifactId>
<packaging>jar</packaging>
<name>spring-cloud-stream-binder-rabbit</name>
<description>RabbitMQ binder implementation</description>
<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
<packaging>jar</packaging>
<name>spring-cloud-stream-binder-rabbit</name>
<description>RabbitMQ binder implementation</description>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binders-parent</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<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>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-spi</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-test</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-amqp</artifactId>
<version>${spring-integration.version}</version>
</dependency>
</dependencies>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-spi</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-codec</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>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-amqp</artifactId>
<version>${spring-integration.version}</version>
</dependency>
</dependencies>
</project>

5
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -90,7 +90,6 @@ import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
import org.springframework.cloud.stream.binder.MessageValues;
import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
@@ -302,9 +301,8 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
private volatile boolean clustered;
public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, MultiTypeCodec<Object> codec) {
public RabbitMessageChannelBinder(ConnectionFactory connectionFactory) {
Assert.notNull(connectionFactory, "connectionFactory must not be null");
Assert.notNull(codec, "codec must not be null");
this.connectionFactory = connectionFactory;
this.rabbitTemplate.setConnectionFactory(connectionFactory);
this.rabbitTemplate.afterPropertiesSet();
@@ -312,7 +310,6 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
this.autoDeclareContext.refresh();
this.rabbitAdmin.setApplicationContext(this.autoDeclareContext);
this.rabbitAdmin.afterPropertiesSet();
this.setCodec(codec);
}
/**

267
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java Normal file
View File

@@ -0,0 +1,267 @@
/*
* 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.rabbit.config;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* @author David Turanski
*/
@ConfigurationProperties(prefix = "spring.cloud.stream.binder.rabbit.default")
class RabbitBinderConfigurationProperties {
private AcknowledgeMode acknowledgeMode;
private int backOffInitialInterval;
private int backOffMaxInterval;
private int backOffMultiplier;
private boolean transacted;
private boolean concurrency;
private MessageDeliveryMode defaultDeliveryMode;
private boolean defaultRequeueRejected;
private int maxAttempts;
private int maxConcurrency;
private int prefetchCount;
private String prefix;
private String[] replyHeaderPatterns;
private String[] requestHeaderPatterns;
private int txSize;
private boolean autoBindDLQ;
private boolean republishToDLQ;
private boolean batchingEnabled;
private int batchSize;
private int batchBufferLimit;
private int batchTimeout;
private boolean compress;
private int compressionLevel;
private boolean durableSubscription;
public AcknowledgeMode getAcknowledgeMode() {
return acknowledgeMode;
}
public void setAcknowledgeMode(AcknowledgeMode acknowledgeMode) {
this.acknowledgeMode = acknowledgeMode;
}
public int getBackOffInitialInterval() {
return backOffInitialInterval;
}
public void setBackOffInitialInterval(int backOffInitialInterval) {
this.backOffInitialInterval = backOffInitialInterval;
}
public int getBackOffMaxInterval() {
return backOffMaxInterval;
}
public void setBackOffMaxInterval(int backOffMaxInterval) {
this.backOffMaxInterval = backOffMaxInterval;
}
public int getBackOffMultiplier() {
return backOffMultiplier;
}
public void setBackOffMultiplier(int backOffMultiplier) {
this.backOffMultiplier = backOffMultiplier;
}
public boolean isTransacted() {
return transacted;
}
public void setTransacted(boolean transacted) {
this.transacted = transacted;
}
public boolean isConcurrency() {
return concurrency;
}
public void setConcurrency(boolean concurrency) {
this.concurrency = concurrency;
}
public MessageDeliveryMode getDefaultDeliveryMode() {
return defaultDeliveryMode;
}
public void setDefaultDeliveryMode(MessageDeliveryMode defaultDeliveryMode) {
this.defaultDeliveryMode = defaultDeliveryMode;
}
public boolean isDefaultRequeueRejected() {
return defaultRequeueRejected;
}
public void setDefaultRequeueRejected(boolean defaultRequeueRejected) {
this.defaultRequeueRejected = defaultRequeueRejected;
}
public int getMaxAttempts() {
return maxAttempts;
}
public void setMaxAttempts(int maxAttempts) {
this.maxAttempts = maxAttempts;
}
public int getMaxConcurrency() {
return maxConcurrency;
}
public void setMaxConcurrency(int maxConcurrency) {
this.maxConcurrency = maxConcurrency;
}
public int getPrefetchCount() {
return prefetchCount;
}
public void setPrefetchCount(int prefetchCount) {
this.prefetchCount = prefetchCount;
}
public String getPrefix() {
return prefix;
}
public void setPrefix(String prefix) {
this.prefix = prefix;
}
public String[] getReplyHeaderPatterns() {
return replyHeaderPatterns;
}
public void setReplyHeaderPatterns(String[] replyHeaderPatterns) {
this.replyHeaderPatterns = replyHeaderPatterns;
}
public String[] getRequestHeaderPatterns() {
return requestHeaderPatterns;
}
public void setRequestHeaderPatterns(String[] requestHeaderPatterns) {
this.requestHeaderPatterns = requestHeaderPatterns;
}
public int getTxSize() {
return txSize;
}
public void setTxSize(int txSize) {
this.txSize = txSize;
}
public boolean isAutoBindDLQ() {
return autoBindDLQ;
}
public void setAutoBindDLQ(boolean autoBindDLQ) {
this.autoBindDLQ = autoBindDLQ;
}
public boolean isRepublishToDLQ() {
return republishToDLQ;
}
public void setRepublishToDLQ(boolean republishToDLQ) {
this.republishToDLQ = republishToDLQ;
}
public boolean isBatchingEnabled() {
return batchingEnabled;
}
public void setBatchingEnabled(boolean batchingEnabled) {
this.batchingEnabled = batchingEnabled;
}
public int getBatchSize() {
return batchSize;
}
public void setBatchSize(int batchSize) {
this.batchSize = batchSize;
}
public int getBatchBufferLimit() {
return batchBufferLimit;
}
public void setBatchBufferLimit(int batchBufferLimit) {
this.batchBufferLimit = batchBufferLimit;
}
public int getBatchTimeout() {
return batchTimeout;
}
public void setBatchTimeout(int batchTimeout) {
this.batchTimeout = batchTimeout;
}
public boolean isCompress() {
return compress;
}
public void setCompress(boolean compress) {
this.compress = compress;
}
public int getCompressionLevel() {
return compressionLevel;
}
public void setCompressionLevel(int compressionLevel) {
this.compressionLevel = compressionLevel;
}
public boolean isDurableSubscription() {
return durableSubscription;
}
public void setDurableSubscription(boolean durableSubscription) {
this.durableSubscription = durableSubscription;
}
}

99
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java Normal file
View File

@@ -0,0 +1,99 @@
/*
* 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.rabbit.config;
/**
* @author David Turanski
*/
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.rabbit.ConnectionFactorySettings;
import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
import org.springframework.cloud.stream.config.codec.CodecConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.integration.codec.Codec;
@Configuration
@Import(CodecConfiguration.class)
@EnableConfigurationProperties({RabbitBinderConfigurationProperties.class, SpringRabbitMQProperties.class})
public class RabbitMessageChannelBinderConfiguration {
@Autowired
private Codec codec;
@Autowired
private ConnectionFactory rabbitConnectionFactory;
@Autowired
private RabbitBinderConfigurationProperties rabbitBinderConfigurationProperties;
@Autowired
private SpringRabbitMQProperties springRabbitMQProperties;
@Bean
RabbitMessageChannelBinder rabbitMessageChannelBinder() {
RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(rabbitConnectionFactory);
binder.setCodec(codec);
binder.setAddresses(springRabbitMQProperties.getAddresses());
binder.setAdminAddresses(springRabbitMQProperties.getAdminAdresses());
binder.setCompressingPostProcessor(gZipPostProcessor());
binder.setDecompressingPostProcessor(deCompressingPostProcessor());
binder.setDefaultAcknowledgeMode(rabbitBinderConfigurationProperties.getAcknowledgeMode());
binder.setDefaultAutoBindDLQ(rabbitBinderConfigurationProperties.isAutoBindDLQ());
binder.setDefaultChannelTransacted(rabbitBinderConfigurationProperties.isTransacted());
binder.setDefaultDefaultDeliveryMode(rabbitBinderConfigurationProperties.getDefaultDeliveryMode());
binder.setDefaultDefaultRequeueRejected(rabbitBinderConfigurationProperties.isDefaultRequeueRejected());
binder.setDefaultMaxConcurrency(rabbitBinderConfigurationProperties.getMaxConcurrency());
binder.setDefaultPrefetchCount(rabbitBinderConfigurationProperties.getPrefetchCount());
binder.setDefaultPrefix(rabbitBinderConfigurationProperties.getPrefix());
binder.setDefaultReplyHeaderPatterns(rabbitBinderConfigurationProperties.getReplyHeaderPatterns());
binder.setDefaultRepublishToDLQ(rabbitBinderConfigurationProperties.isRepublishToDLQ());
binder.setDefaultRequestHeaderPatterns(rabbitBinderConfigurationProperties.getRequestHeaderPatterns());
binder.setDefaultTxSize(rabbitBinderConfigurationProperties.getTxSize());
binder.setNodes(springRabbitMQProperties.getNodes());
binder.setPassword(springRabbitMQProperties.getPassword());
binder.setSslPropertiesLocation(springRabbitMQProperties.getSslPropertiesLocation());
binder.setUsername(springRabbitMQProperties.getUsername());
binder.setUseSSL(springRabbitMQProperties.isUseSSL());
binder.setVhost(springRabbitMQProperties.getVhost());
return binder;
}
@Bean
MessagePostProcessor deCompressingPostProcessor() {
return new DelegatingDecompressingPostProcessor();
}
@Bean
MessagePostProcessor gZipPostProcessor() {
GZipPostProcessor gZipPostProcessor = new GZipPostProcessor();
gZipPostProcessor.setLevel(rabbitBinderConfigurationProperties.getCompressionLevel());
return gZipPostProcessor;
}
@Bean
ConnectionFactorySettings rabbitConnectionFactorySettings() {
return new ConnectionFactorySettings();
}
}

106
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/SpringRabbitMQProperties.java Normal file
View File

@@ -0,0 +1,106 @@
/*
* 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.rabbit.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.core.io.Resource;
/**
* @author David Turanski
*/
@ConfigurationProperties(prefix = "spring.rabbitmq")
class SpringRabbitMQProperties {
private String[] addresses;
private String[] adminAdresses;
private String[] nodes;
private String username;
private String password;
private String vhost;
private boolean useSSL;
private Resource sslPropertiesLocation;
public String[] getAddresses() {
return addresses;
}
public void setAddresses(String[] addresses) {
this.addresses = addresses;
}
public String[] getAdminAdresses() {
return adminAdresses;
}
public void setAdminAdresses(String[] adminAdresses) {
this.adminAdresses = adminAdresses;
}
public String[] getNodes() {
return nodes;
}
public void setNodes(String[] nodes) {
this.nodes = nodes;
}
public String getUsername() {
return username;
}
public void setUsername(String username) {
this.username = username;
}
public String getPassword() {
return password;
}
public void setPassword(String password) {
this.password = password;
}
public String getVhost() {
return vhost;
}
public void setVhost(String vhost) {
this.vhost = vhost;
}
public boolean isUseSSL() {
return useSSL;
}
public void setUseSSL(boolean useSSL) {
this.useSSL = useSSL;
}
public Resource getSslPropertiesLocation() {
return sslPropertiesLocation;
}
public void setSslPropertiesLocation(Resource sslPropertiesLocation) {
this.sslPropertiesLocation = sslPropertiesLocation;
}
}

56
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring-cloud-stream/binder/rabbit-binder.xml
View File

@@ -1,56 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<bean class="org.springframework.cloud.stream.binder.rabbit.ConnectionFactorySettings" />
<bean id="messageBus" class="org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder">
<constructor-arg ref="rabbitConnectionFactory" />
<constructor-arg ref="codec"/>
<property name="defaultAcknowledgeMode" value="#{T(org.springframework.amqp.core.AcknowledgeMode).${xd.messagebus.rabbit.default.ackMode}}" />
<property name="defaultBackOffInitialInterval" value="${xd.messagebus.rabbit.default.backOffInitialInterval}" />
<property name="defaultBackOffMaxInterval" value="${xd.messagebus.rabbit.default.backOffMaxInterval}" />
<property name="defaultBackOffMultiplier" value="${xd.messagebus.rabbit.default.backOffMultiplier}" />
<property name="defaultChannelTransacted" value="${xd.messagebus.rabbit.default.transacted}" />
<property name="defaultConcurrency" value="${xd.messagebus.rabbit.default.concurrency}" />
<property name="defaultDefaultDeliveryMode" value="#{T(org.springframework.amqp.core.MessageDeliveryMode).${xd.messagebus.rabbit.default.deliveryMode}}" />
<property name="defaultDefaultRequeueRejected" value="${xd.messagebus.rabbit.default.requeue}" />
<property name="defaultMaxAttempts" value="${xd.messagebus.rabbit.default.maxAttempts}" />
<property name="defaultMaxConcurrency" value="${xd.messagebus.rabbit.default.maxConcurrency}" />
<property name="defaultPrefetchCount" value="${xd.messagebus.rabbit.default.prefetch}" />
<property name="defaultPrefix" value="${xd.messagebus.rabbit.default.prefix}" />
<property name="defaultReplyHeaderPatterns" value="${xd.messagebus.rabbit.default.replyHeaderPatterns}" />
<property name="defaultRequestHeaderPatterns" value="${xd.messagebus.rabbit.default.requestHeaderPatterns}" />
<property name="defaultTxSize" value="${xd.messagebus.rabbit.default.txSize}" />
<property name="defaultAutoBindDLQ" value="${xd.messagebus.rabbit.default.autoBindDLQ}" />
<property name="defaultRepublishToDLQ" value="${xd.messagebus.rabbit.default.republishToDLQ}" />
<property name="defaultBatchingEnabled" value="${xd.messagebus.rabbit.default.batchingEnabled}" />
<property name="defaultBatchSize" value="${xd.messagebus.rabbit.default.batchSize}" />
<property name="defaultBatchBufferLimit" value="${xd.messagebus.rabbit.default.batchBufferLimit}" />
<property name="defaultBatchTimeout" value="${xd.messagebus.rabbit.default.batchTimeout}" />
<property name="defaultCompress" value="${xd.messagebus.rabbit.default.compress}" />
<property name="compressingPostProcessor">
<bean class="org.springframework.amqp.support.postprocessor.GZipPostProcessor">
<property name="level" value="${xd.messagebus.rabbit.compressionLevel:#{T(java.util.zip.Deflater).BEST_SPEED}}" />
</bean>
</property>
<property name="decompressingPostProcessor">
<bean class="org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor">
<!-- set a map of decompressors here if using other than the default -->
</bean>
</property>
<property name="defaultDurableSubscription" value="${xd.messagebus.rabbit.default.durableSubscription}" />
<property name="addresses" value="${spring.rabbitmq.addresses:}" />
<property name="adminAddresses" value="${spring.rabbitmq.adminAddresses:}" />
<property name="nodes" value="${spring.rabbitmq.nodes:}" />
<property name="username" value="${spring.rabbitmq.username:}" />
<property name="password" value="${spring.rabbitmq.password:}" />
<property name="vhost" value="${spring.rabbitmq.virtual_host:}" />
<property name="useSSL" value="${spring.rabbitmq.useSSL:false}" />
<property name="sslPropertiesLocation" value="${spring.rabbitmq.sslProperties:}" />
</bean>
</beans>

3
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
View File

@@ -75,6 +75,7 @@ import static org.mockito.Mockito.when;
/**
* @author Mark Fisher
* @author Gary Russell
* @author David Turanski
*/
public class RabbitBinderTests extends PartitionCapableBinderTests {
@@ -88,7 +89,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
@Override
protected Binder getBinder() {
if (testBinder == null) {
testBinder = new RabbitTestBinder(rabbitAvailableRule.getResource(), getCodec());
testBinder = new RabbitTestBinder(rabbitAvailableRule.getResource());
}
return testBinder;
}

11
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java
View File

@@ -19,10 +19,11 @@ package org.springframework.cloud.stream.binder.rabbit;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.codec.Codec;
import org.springframework.integration.codec.kryo.PojoCodec;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.cloud.stream.binder.AbstractTestBinder;
import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
/**
@@ -30,6 +31,7 @@ import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
*
* @author Ilayaperumal Gopinathan
* @author Gary Russell
* @author David Turanski
*/
public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBinder> {
@@ -38,11 +40,7 @@ public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBin
private final RabbitAdmin rabbitAdmin;
public RabbitTestBinder(ConnectionFactory connectionFactory) {
this.rabbitAdmin = new RabbitAdmin(connectionFactory);
}
public RabbitTestBinder(ConnectionFactory connectionFactory, MultiTypeCodec<Object> codec) {
RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(connectionFactory, codec);
RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(connectionFactory);
GenericApplicationContext context = new GenericApplicationContext();
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.setPoolSize(1);
@@ -50,6 +48,7 @@ public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBin
context.getBeanFactory().registerSingleton(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, scheduler);
context.refresh();
binder.setApplicationContext(context);
binder.setCodec(new PojoCodec());
this.setBinder(binder);
this.rabbitAdmin = new RabbitAdmin(connectionFactory);
}

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

@@ -1,58 +1,66 @@
<?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-redis</artifactId>
<packaging>jar</packaging>
<name>spring-cloud-stream-binder-redis</name>
<description>Redis binder implementation</description>
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>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binders-parent</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<artifactId>spring-cloud-stream-binder-redis</artifactId>
<packaging>jar</packaging>
<name>spring-cloud-stream-binder-redis</name>
<description>Redis binder implementation</description>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binders-parent</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-spi</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-test</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-redis</artifactId>
<version>${spring-integration.version}</version>
</dependency>
<dependency>
<groupId>joda-time</groupId>
<artifactId>joda-time</artifactId>
<version>2.5</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-tuple</artifactId>
<version>${spring-xd.version}</version>
<exclusions>
<exclusion>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-codec</artifactId>
</exclusion>
</exclusions>
<scope>test</scope>
</dependency>
</dependencies>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-spi</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-codec</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-codec</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>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-redis</artifactId>
<version>${spring-integration.version}</version>
</dependency>
<dependency>
<groupId>joda-time</groupId>
<artifactId>joda-time</artifactId>
<version>2.5</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-tuple</artifactId>
<version>${spring-xd.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>

9
spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java
View File

@@ -53,7 +53,6 @@ import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
import org.springframework.cloud.stream.binder.MessageValues;
import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
/**
* A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by Redis.
@@ -140,16 +139,14 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple
private final RedisQueueOutboundChannelAdapter errorAdapter;
public RedisMessageChannelBinder(RedisConnectionFactory connectionFactory, MultiTypeCodec<Object> codec) {
this(connectionFactory, codec, new String[0]);
public RedisMessageChannelBinder(RedisConnectionFactory connectionFactory) {
this(connectionFactory, new String[0]);
}
public RedisMessageChannelBinder(RedisConnectionFactory connectionFactory, MultiTypeCodec<Object> codec,
public RedisMessageChannelBinder(RedisConnectionFactory connectionFactory,
String... headersToMap) {
Assert.notNull(connectionFactory, "connectionFactory must not be null");
Assert.notNull(codec, "codec must not be null");
this.connectionFactory = connectionFactory;
setCodec(codec);
this.errorAdapter = new RedisQueueOutboundChannelAdapter(
parser.parseExpression("headers['" + ERROR_HEADER + "']"), connectionFactory);
if (headersToMap != null && headersToMap.length > 0) {

71
spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisBinderConfigurationProperties.java Normal file
View File

@@ -0,0 +1,71 @@
/*
* 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.redis.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* @author David Turanski
*/
@ConfigurationProperties(prefix = "spring.cloud.stream.binder.redis.default")
class RedisBinderConfigurationProperties {
private int backOffInitialInterval;
private int backOffMaxInterval;
private double backOffMultiplier;
private int concurrency;
private int maxAttempts;
public int getBackOffInitialInterval() {
return backOffInitialInterval;
}
public void setBackOffInitialInterval(int backOffInitialInterval) {
this.backOffInitialInterval = backOffInitialInterval;
}
public int getBackOffMaxInterval() {
return backOffMaxInterval;
}
public void setBackOffMaxInterval(int backOffMaxInterval) {
this.backOffMaxInterval = backOffMaxInterval;
}
public double getBackOffMultiplier() {
return backOffMultiplier;
}
public void setBackOffMultiplier(double backOffMultiplier) {
this.backOffMultiplier = backOffMultiplier;
}
public int getConcurrency() {
return concurrency;
}
public void setConcurrency(int concurrency) {
this.concurrency = concurrency;
}
public int getMaxAttempts() {
return maxAttempts;
}
public void setMaxAttempts(int maxAttempts) {
this.maxAttempts = maxAttempts;
}
}

66
spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisMessageChannelBinderConfiguration.java Normal file
View File

@@ -0,0 +1,66 @@
/*
* 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.redis.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.redis.RedisMessageChannelBinder;
import org.springframework.cloud.stream.config.codec.CodecConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.integration.codec.Codec;
/**
* @author David Turanski
*/
@Configuration
@Import(CodecConfiguration.class)
@EnableConfigurationProperties(RedisBinderConfigurationProperties.class)
@ConfigurationProperties(prefix = "spring.cloud.stream.binder.redis")
public class RedisMessageChannelBinderConfiguration {
private String[] headers;
@Autowired
private Codec codec;
@Autowired
private RedisBinderConfigurationProperties redisBinderConfigurationProperties;
@Autowired
private RedisConnectionFactory redisConnectionFactory;
@Bean
public RedisMessageChannelBinder redisMessageChannelBinder() {
RedisMessageChannelBinder redisMessageChannelBinder = new RedisMessageChannelBinder(redisConnectionFactory,
headers);
redisMessageChannelBinder.setCodec(codec);
redisMessageChannelBinder.setDefaultBackOffInitialInterval(redisBinderConfigurationProperties.getBackOffInitialInterval());
redisMessageChannelBinder.setDefaultBackOffMaxInterval(redisBinderConfigurationProperties.getBackOffMaxInterval());
redisMessageChannelBinder.setDefaultBackOffMultiplier(redisBinderConfigurationProperties.getBackOffMultiplier());
redisMessageChannelBinder.setDefaultConcurrency(redisBinderConfigurationProperties.getConcurrency());
redisMessageChannelBinder.setDefaultMaxAttempts(redisBinderConfigurationProperties.getMaxAttempts());
return redisMessageChannelBinder;
}
public void setHeaders(String[] headers) {
this.headers = headers;
}
}

19
spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring-cloud-stream/binder/redis-binder.xml
View File

@@ -1,19 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<bean id="messageBus" class="org.springframework.cloud.stream.binder.redis.RedisMessageChannelBinder">
<constructor-arg ref="redisConnectionFactory" />
<constructor-arg ref="codec"/>
<constructor-arg value="${xd.messagebus.redis.headers:}" />
<property name="defaultBackOffInitialInterval" value="${xd.messagebus.redis.default.backOffInitialInterval}" />
<property name="defaultBackOffMaxInterval" value="${xd.messagebus.redis.default.backOffMaxInterval}" />
<property name="defaultBackOffMultiplier" value="${xd.messagebus.redis.default.backOffMultiplier}" />
<property name="defaultConcurrency" value="${xd.messagebus.redis.default.concurrency}" />
<property name="defaultMaxAttempts" value="${xd.messagebus.redis.default.maxAttempts}" />
</bean>
</beans>

5
spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java
View File

@@ -56,6 +56,7 @@ import org.springframework.cloud.stream.binder.Spy;
/**
* @author Gary Russell
* @author David Turanski
*/
public class RedisBinderTests extends PartitionCapableBinderTests {
@@ -72,7 +73,7 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
@Override
protected Binder getBinder() {
if (testBinder == null) {
testBinder = new RedisTestBinder(redisAvailableRule.getResource(), getCodec());
testBinder = new RedisTestBinder(redisAvailableRule.getResource());
}
return testBinder;
}
@@ -339,7 +340,7 @@ public class RedisBinderTests extends PartitionCapableBinderTests {
@Test
public void testMoreHeaders() {
RedisMessageChannelBinder binder = new RedisMessageChannelBinder(mock(RedisConnectionFactory.class), getCodec(), "foo", "bar");
RedisMessageChannelBinder binder = new RedisMessageChannelBinder(mock(RedisConnectionFactory.class), "foo", "bar");
Collection<String> headers = Arrays.asList(TestUtils.getPropertyValue(binder, "headersToMap", String[].class));
assertEquals(10, headers.size());
assertTrue(headers.contains("foo"));

11
spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisTestBinder.java
View File

@@ -20,12 +20,13 @@ import org.springframework.context.support.GenericApplicationContext;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.integration.channel.DefaultHeaderChannelRegistry;
import org.springframework.integration.codec.Codec;
import org.springframework.integration.codec.kryo.PojoCodec;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.cloud.stream.binder.AbstractTestBinder;
import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
/**
@@ -33,17 +34,14 @@ import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
*
* @author Ilayaperumal Gopinathan
* @author Gary Russell
* @author David Turanski
*/
public class RedisTestBinder extends AbstractTestBinder<RedisMessageChannelBinder> {
private StringRedisTemplate template;
public RedisTestBinder(RedisConnectionFactory connectionFactory) {
template = new StringRedisTemplate(connectionFactory);
}
public RedisTestBinder(RedisConnectionFactory connectionFactory, MultiTypeCodec<Object> codec) {
RedisMessageChannelBinder binder = new RedisMessageChannelBinder(connectionFactory, codec);
RedisMessageChannelBinder binder = new RedisMessageChannelBinder(connectionFactory);
GenericApplicationContext context = new GenericApplicationContext();
context.getBeanFactory().registerSingleton(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
new DefaultMessageBuilderFactory());
@@ -57,6 +55,7 @@ public class RedisTestBinder extends AbstractTestBinder<RedisMessageChannelBinde
channelRegistry);
context.refresh();
binder.setApplicationContext(context);
binder.setCodec(new PojoCodec());
setBinder(binder);
template = new StringRedisTemplate(connectionFactory);
}

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

@@ -19,15 +19,15 @@
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-codec</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>
<version>${spring-framework.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.retry</groupId>
<artifactId>spring-retry</artifactId>

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

@@ -46,6 +46,7 @@ import org.springframework.core.serializer.support.SerializationFailedException;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.codec.Codec;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.messaging.Message;
@@ -63,7 +64,6 @@ import org.springframework.util.ClassUtils;
import org.springframework.util.IdGenerator;
import org.springframework.util.MimeType;
import org.springframework.util.StringUtils;
import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
import static org.springframework.util.MimeTypeUtils.ALL;
import static org.springframework.util.MimeTypeUtils.APPLICATION_OCTET_STREAM;
@@ -91,7 +91,7 @@ public abstract class MessageChannelBinderSupport
private volatile AbstractApplicationContext applicationContext;
private volatile MultiTypeCodec<Object> codec;
private volatile Codec codec;
private final StringConvertingContentTypeResolver contentTypeResolver = new StringConvertingContentTypeResolver();
@@ -256,7 +256,7 @@ public abstract class MessageChannelBinderSupport
return this.applicationContext.getBeanFactory();
}
public void setCodec(MultiTypeCodec<Object> codec) {
public void setCodec(Codec codec) {
this.codec = codec;
}
@@ -584,7 +584,7 @@ public abstract class MessageChannelBinderSupport
if (originalPayload instanceof String) {
return ((String) originalPayload).getBytes("UTF-8");
}
this.codec.serialize(originalPayload, bos);
this.codec.encode(originalPayload, bos);
return bos.toByteArray();
}
catch (IOException e) {
@@ -644,7 +644,7 @@ public abstract class MessageChannelBinderSupport
targetType = ClassUtils.forName(className, null);
payloadTypeCache.put(className, targetType);
}
return codec.deserialize(bytes, targetType);
return codec.decode(bytes, targetType);
}
catch (ClassNotFoundException e) {
throw new SerializationFailedException("unable to deserialize [" + className + "]. Class not found.",

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

@@ -42,21 +42,11 @@
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-tuple</artifactId>
<version>${spring-xd.version}</version>
<exclusions>
<exclusion>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-codec</artifactId>
</exclusion>
</exclusions>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-spi</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-codec</artifactId>
</dependency>
</dependencies>
</project>

9
spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java
View File

@@ -30,12 +30,12 @@ import org.springframework.http.MediaType;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.interceptor.WireTap;
import org.springframework.integration.codec.Codec;
import org.springframework.integration.codec.kryo.PojoCodec;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.cloud.stream.binder.Binder.Capability;
import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
import org.springframework.xd.dirt.integration.bus.serializer.kryo.PojoCodec;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -287,11 +287,6 @@ public abstract class AbstractBinderTests {
return (List<?>) accessor.getPropertyValue("bindings");
}
@SuppressWarnings({"unchecked", "rawtypes"})
protected MultiTypeCodec<Object> getCodec() {
return new PojoCodec();
}
protected abstract Binder getBinder() throws Exception;
@After

24
spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/test/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupportTests.java
View File

@@ -18,12 +18,17 @@ package org.springframework.cloud.stream.binder;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Registration;
import org.junit.Before;
import org.junit.Test;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport.JavaClassMimeTypeConversion;
import org.springframework.integration.codec.kryo.KryoRegistrar;
import org.springframework.integration.codec.kryo.PojoCodec;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -32,7 +37,6 @@ import org.springframework.messaging.converter.ContentTypeResolver;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import org.springframework.xd.dirt.integration.bus.serializer.kryo.PojoCodec;
import org.springframework.xd.tuple.DefaultTuple;
import org.springframework.xd.tuple.Tuple;
import org.springframework.xd.tuple.TupleBuilder;
@@ -52,10 +56,9 @@ public class MessageChannelBinderSupportTests {
private final TestMessageChannelBinder binder = new TestMessageChannelBinder();
@SuppressWarnings({"unchecked", "rawtypes"})
@Before
public void setUp() {
binder.setCodec(new PojoCodec(new TupleKryoRegistrar()));
binder.setCodec(new PojoCodec(new TupleRegistrar()));
}
@Test
@@ -293,4 +296,19 @@ public class MessageChannelBinderSupportTests {
}
}
//TODO: temporary wrapper for compatibility with SI Codec types
private static class TupleRegistrar implements KryoRegistrar {
private TupleKryoRegistrar delegate = new TupleKryoRegistrar();
@Override
public void registerTypes(Kryo kryo) {
delegate.registerTypes(kryo);
}
@Override
public List<Registration> getRegistrations() {
return delegate.getRegistrations();
}
}
}