diff --git a/pom.xml b/pom.xml index aa7c7378c..ad4deff54 100644 --- a/pom.xml +++ b/pom.xml @@ -23,7 +23,7 @@ Brixton.BUILD-SNAPSHOT 1.2.1.BUILD-SNAPSHOT 4.2.0.RC2 - 4.2.0.M2 + 4.2.0.BUILD-SNAPSHOT 1.2.0.BUILD-SNAPSHOT 1.1.0.BUILD-SNAPSHOT diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/pom.xml b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/pom.xml index 70637f041..7b6dbb771 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/pom.xml +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/pom.xml @@ -26,6 +26,15 @@ org.springframework.cloud spring-cloud-stream-binder-spi + + org.springframework.cloud + spring-cloud-stream-codec + + + org.springframework.boot + spring-boot-autoconfigure + true + org.springframework.cloud spring-cloud-stream-binder-test @@ -63,12 +72,6 @@ org.springframework.xd spring-xd-tuple ${spring-xd.version} - - - org.springframework.xd - spring-xd-codec - - test diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index ad2ff9bde..0aa7ba118 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -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 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 diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java new file mode 100644 index 000000000..beaddc929 --- /dev/null +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java @@ -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; + } +} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaMessageChannelBinderConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaMessageChannelBinderConfiguration.java new file mode 100644 index 000000000..3c609219e --- /dev/null +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaMessageChannelBinderConfiguration.java @@ -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; + } + +} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring-cloud-stream/binder/kafka-binder.xml b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring-cloud-stream/binder/kafka-binder.xml deleted file mode 100644 index e7f23bf24..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring-cloud-stream/binder/kafka-binder.xml +++ /dev/null @@ -1,49 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index c5203e727..58fcf8e45 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -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 diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java index 311d10226..759c94404 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTestBinder.java @@ -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 { public KafkaTestBinder(KafkaTestSupport kafkaTestSupport) { - this(kafkaTestSupport, getCodec(), KafkaMessageChannelBinder.Mode.embeddedHeaders); + this(kafkaTestSupport, KafkaMessageChannelBinder.Mode.embeddedHeaders); } - public KafkaTestBinder(KafkaTestSupport kafkaTestSupport, MultiTypeCodec codec, + public KafkaTestBinder(KafkaTestSupport kafkaTestSupport, KafkaMessageChannelBinder.Mode mode) { try { @@ -46,7 +52,8 @@ public class KafkaTestBinder extends AbstractTestBinder 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 getRegistrations() { + return delegate.getRegistrations(); + } } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java index 176b89dd6..79acf60a3 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java @@ -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 diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-local/pom.xml b/spring-cloud-stream-binders/spring-cloud-stream-binder-local/pom.xml index c4ede9961..3566e897c 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-local/pom.xml +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-local/pom.xml @@ -23,9 +23,15 @@ org.springframework.cloud spring-cloud-stream-binder-spi + + org.springframework.boot + spring-boot-autoconfigure + true + org.springframework.cloud spring-cloud-stream-binder-test + test diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalExecutorConfigurationProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalExecutorConfigurationProperties.java new file mode 100644 index 000000000..d10582345 --- /dev/null +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalExecutorConfigurationProperties.java @@ -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; + } +} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalMessageChannelBinderConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalMessageChannelBinderConfiguration.java new file mode 100644 index 000000000..cee477b01 --- /dev/null +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalMessageChannelBinderConfiguration.java @@ -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; + } + +} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/resources/META-INF/spring-cloud-stream/binder/local-binder.xml b/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/resources/META-INF/spring-cloud-stream/binder/local-binder.xml deleted file mode 100644 index 0a8a9930a..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/resources/META-INF/spring-cloud-stream/binder/local-binder.xml +++ /dev/null @@ -1,18 +0,0 @@ - - - - - - - - - - - - - - diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/pom.xml b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/pom.xml index 9f074fc81..062050a59 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/pom.xml +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/pom.xml @@ -1,40 +1,49 @@ - 4.0.0 + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> + 4.0.0 - spring-cloud-stream-binder-rabbit - jar - spring-cloud-stream-binder-rabbit - RabbitMQ binder implementation + spring-cloud-stream-binder-rabbit + jar + spring-cloud-stream-binder-rabbit + RabbitMQ binder implementation - - org.springframework.cloud - spring-cloud-stream-binders-parent - 1.0.0.BUILD-SNAPSHOT - + + org.springframework.cloud + spring-cloud-stream-binders-parent + 1.0.0.BUILD-SNAPSHOT + - - UTF-8 - + + UTF-8 + - - - org.springframework.cloud - spring-cloud-stream-binder-spi - - - org.springframework.cloud - spring-cloud-stream-binder-test - - - org.springframework.boot - spring-boot-starter-amqp - - - org.springframework.integration - spring-integration-amqp - ${spring-integration.version} - - + + + org.springframework.cloud + spring-cloud-stream-binder-spi + + + org.springframework.cloud + spring-cloud-stream-codec + + + org.springframework.boot + spring-boot-autoconfigure + true + + + org.springframework.cloud + spring-cloud-stream-binder-test + + + org.springframework.boot + spring-boot-starter-amqp + + + org.springframework.integration + spring-integration-amqp + ${spring-integration.version} + + diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 9d756ab92..9d4da7aa6 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -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 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); } /** diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java new file mode 100644 index 000000000..b8379283d --- /dev/null +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java @@ -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; + } +} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java new file mode 100644 index 000000000..868750f36 --- /dev/null +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java @@ -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(); + } +} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/SpringRabbitMQProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/SpringRabbitMQProperties.java new file mode 100644 index 000000000..8d9651b34 --- /dev/null +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/SpringRabbitMQProperties.java @@ -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; + } +} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring-cloud-stream/binder/rabbit-binder.xml b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring-cloud-stream/binder/rabbit-binder.xml deleted file mode 100644 index 43aefda86..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring-cloud-stream/binder/rabbit-binder.xml +++ /dev/null @@ -1,56 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index a6c8cf427..66afadc96 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -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; } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java index a8b9b714e..d7ef2c621 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java @@ -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 { @@ -38,11 +40,7 @@ public class RabbitTestBinder extends AbstractTestBinder 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 - 4.0.0 - - spring-cloud-stream-binder-redis - jar - spring-cloud-stream-binder-redis - Redis binder implementation + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> + 4.0.0 - - org.springframework.cloud - spring-cloud-stream-binders-parent - 1.0.0.BUILD-SNAPSHOT - + spring-cloud-stream-binder-redis + jar + spring-cloud-stream-binder-redis + Redis binder implementation - - UTF-8 - + + org.springframework.cloud + spring-cloud-stream-binders-parent + 1.0.0.BUILD-SNAPSHOT + - - - org.springframework.cloud - spring-cloud-stream-binder-spi - - - org.springframework.cloud - spring-cloud-stream-binder-test - - - org.springframework.boot - spring-boot-starter-redis - - - org.springframework.integration - spring-integration-redis - ${spring-integration.version} - - - joda-time - joda-time - 2.5 - test - - - org.springframework.xd - spring-xd-tuple - ${spring-xd.version} - - - org.springframework.xd - spring-xd-codec - - - test - - + + UTF-8 + + + + + org.springframework.cloud + spring-cloud-stream-binder-spi + + + org.springframework.cloud + spring-cloud-stream-codec + true + + + org.springframework.cloud + spring-cloud-stream-codec + + + org.springframework.boot + spring-boot-autoconfigure + true + + + org.springframework.cloud + spring-cloud-stream-binder-test + + + org.springframework.boot + spring-boot-starter-redis + + + org.springframework.integration + spring-integration-redis + ${spring-integration.version} + + + joda-time + joda-time + 2.5 + test + + + org.springframework.xd + spring-xd-tuple + ${spring-xd.version} + test + + diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java index 2cd082aab..2bd3b1fa2 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java @@ -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 codec) { - this(connectionFactory, codec, new String[0]); + public RedisMessageChannelBinder(RedisConnectionFactory connectionFactory) { + this(connectionFactory, new String[0]); } - public RedisMessageChannelBinder(RedisConnectionFactory connectionFactory, MultiTypeCodec 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) { diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisBinderConfigurationProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisBinderConfigurationProperties.java new file mode 100644 index 000000000..c1e8063c1 --- /dev/null +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisBinderConfigurationProperties.java @@ -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; + } +} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisMessageChannelBinderConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisMessageChannelBinderConfiguration.java new file mode 100644 index 000000000..624132448 --- /dev/null +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/config/RedisMessageChannelBinderConfiguration.java @@ -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; + } +} diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring-cloud-stream/binder/redis-binder.xml b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring-cloud-stream/binder/redis-binder.xml deleted file mode 100644 index 7eb9b2d76..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring-cloud-stream/binder/redis-binder.xml +++ /dev/null @@ -1,19 +0,0 @@ - - - - - - - - - - - - - - - diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java index 42f2753c4..ac16192d1 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java @@ -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 headers = Arrays.asList(TestUtils.getPropertyValue(binder, "headersToMap", String[].class)); assertEquals(10, headers.size()); assertTrue(headers.contains("foo")); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisTestBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisTestBinder.java index 27f746f77..30b469bf7 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisTestBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisTestBinder.java @@ -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 { private StringRedisTemplate template; public RedisTestBinder(RedisConnectionFactory connectionFactory) { - template = new StringRedisTemplate(connectionFactory); - } - - public RedisTestBinder(RedisConnectionFactory connectionFactory, MultiTypeCodec 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 - - org.springframework.cloud - spring-cloud-stream-codec - org.springframework spring-web ${spring-framework.version} + + org.springframework.integration + spring-integration-core + org.springframework.retry spring-retry diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java index 0cf3f85e2..9605c5c3b 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java @@ -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 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 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.", diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/pom.xml b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/pom.xml index 2f492f18d..e7517b249 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/pom.xml +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/pom.xml @@ -42,21 +42,11 @@ org.springframework.xd spring-xd-tuple ${spring-xd.version} - - - org.springframework.xd - spring-xd-codec - - test org.springframework.cloud spring-cloud-stream-binder-spi - - org.springframework.cloud - spring-cloud-stream-codec - diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java index ae193646b..36354842a 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java @@ -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 getCodec() { - return new PojoCodec(); - } - protected abstract Binder getBinder() throws Exception; @After diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/test/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupportTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/test/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupportTests.java index 8ab651ac7..168081c29 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/test/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupportTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/test/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupportTests.java @@ -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 getRegistrations() { + return delegate.getRegistrations(); + } + } + } diff --git a/spring-cloud-stream-codec/pom.xml b/spring-cloud-stream-codec/pom.xml index db2e23b3d..2cf3782f8 100644 --- a/spring-cloud-stream-codec/pom.xml +++ b/spring-cloud-stream-codec/pom.xml @@ -39,15 +39,11 @@ org.springframework.boot spring-boot-autoconfigure + true com.fasterxml.jackson.core jackson-annotations - - org.springframework.boot - spring-boot-starter-test - test - diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/codec/CodecConfiguration.java b/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/codec/CodecConfiguration.java new file mode 100644 index 000000000..03cdf7231 --- /dev/null +++ b/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/codec/CodecConfiguration.java @@ -0,0 +1,34 @@ +/* + * 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.config.codec; + +import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; + +/** + * Generic {@link org.springframework.integration.codec.Codec} configuration imported by the + * Binder configuration. This is the default configuration for + * Kryo. To provide an alternate Codec implementation, + * remove this configuration from the classpath and replace this class to resolve the Binder configuration + * reference. + * + * @author David Turanski + */ +@Configuration +@Import(KryoCodecAutoConfiguration.class) +public class CodecConfiguration { +} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/CodecConfiguration.java b/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/codec/kryo/KryoCodecAutoConfiguration.java similarity index 61% rename from spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/CodecConfiguration.java rename to spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/codec/kryo/KryoCodecAutoConfiguration.java index 94af0efea..0d8777c99 100644 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/CodecConfiguration.java +++ b/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/codec/kryo/KryoCodecAutoConfiguration.java @@ -13,43 +13,46 @@ * limitations under the License. */ -package org.springframework.cloud.stream.config; +package org.springframework.cloud.stream.config.codec.kryo; import java.util.ArrayList; import java.util.Map; +import com.esotericsoftware.kryo.Kryo; + import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec; -import org.springframework.xd.dirt.integration.bus.serializer.kryo.FileKryoRegistrar; -import org.springframework.xd.dirt.integration.bus.serializer.kryo.KryoRegistrar; -import org.springframework.xd.dirt.integration.bus.serializer.kryo.PojoCodec; +import org.springframework.integration.codec.Codec; +import org.springframework.integration.codec.kryo.FileKryoRegistrar; +import org.springframework.integration.codec.kryo.KryoRegistrar; +import org.springframework.integration.codec.kryo.PojoCodec; + /** + * Auto configures {@link PojoCodec} if Kryo is on the class path. * @author David Turanski */ @Configuration -public class CodecConfiguration { - +@ConditionalOnClass(Kryo.class) +@EnableConfigurationProperties(KryoCodecProperties.class) +@ConditionalOnMissingBean(Codec.class) +public class KryoCodecAutoConfiguration { @Autowired ApplicationContext applicationContext; - @ConditionalOnMissingBean(KryoCodecProperties.class) - @Bean(name = "spring.cloud.streams.codec.kryo.CONFIGURATION_PROPERTIES") - public KryoCodecProperties kryoCodecProperties() { - return new KryoCodecProperties(); - } + @Autowired KryoCodecProperties kryoCodecProperties; @Bean - @ConditionalOnMissingBean(name = "codec") - public MultiTypeCodec codec() { + public PojoCodec codec() { Map kryoRegistrarMap = applicationContext.getBeansOfType(KryoRegistrar .class); - return new PojoCodec(new ArrayList<>(kryoRegistrarMap.values()), kryoCodecProperties().isReferences()); + return new PojoCodec(new ArrayList<>(kryoRegistrarMap.values()), kryoCodecProperties.isReferences()); } @Bean diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/KryoCodecProperties.java b/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/codec/kryo/KryoCodecProperties.java similarity index 94% rename from spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/KryoCodecProperties.java rename to spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/codec/kryo/KryoCodecProperties.java index bd5d415ed..fa168ad6f 100644 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/KryoCodecProperties.java +++ b/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/codec/kryo/KryoCodecProperties.java @@ -13,7 +13,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.config; +package org.springframework.cloud.stream.config.codec.kryo; import com.fasterxml.jackson.annotation.JsonInclude; diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/AbstractCodec.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/AbstractCodec.java deleted file mode 100644 index 5a356ba33..000000000 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/AbstractCodec.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * Copyright 2013 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.xd.dirt.integration.bus.serializer; - -import java.io.ByteArrayInputStream; -import java.io.IOException; - -import org.springframework.core.serializer.Deserializer; -import org.springframework.core.serializer.Serializer; - - -/** - * Support class providing convenience methods for codecs. - * - * @author David Turanski - */ -public abstract class AbstractCodec implements Serializer, Deserializer { - - /** - * Deserialize a byte array. - * - * @param bytes - * @throws IOException - */ - public T deserialize(byte[] bytes) throws IOException { - return deserialize(new ByteArrayInputStream(bytes)); - } -} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/CompositeCodec.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/CompositeCodec.java deleted file mode 100644 index d05e82e91..000000000 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/CompositeCodec.java +++ /dev/null @@ -1,89 +0,0 @@ -/* - * Copyright 2013 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.xd.dirt.integration.bus.serializer; - -import java.io.ByteArrayInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.util.Map; - -import org.springframework.integration.util.ClassUtils; -import org.springframework.util.Assert; - - -/** - * A codec that can delegate to one out of many codecs, depending on the type of the object to serialize/deserialize. - * - * @author David Turanski - */ -public class CompositeCodec

implements MultiTypeCodec { - - private final MultiTypeCodec

defaultCodec; - - private final Map, AbstractCodec

> delegates; - - public CompositeCodec(Map, AbstractCodec

> delegates, MultiTypeCodec

defaultCodec) - { - Assert.notNull(defaultCodec, "'defaultCodec' cannot be null"); - this.defaultCodec = defaultCodec; - this.delegates = delegates; - } - - public CompositeCodec(MultiTypeCodec

defaultCodec) { - this(null, defaultCodec); - } - - @SuppressWarnings("unchecked") - @Override - public void serialize(Object object, OutputStream outputStream) throws IOException { - Assert.notNull(object, "cannot serialize a null object"); - AbstractCodec

codec = findDelegate(object.getClass()); - if (codec != null) { - codec.serialize((P) object, outputStream); - } - else { - defaultCodec.serialize((P) object, outputStream); - } - } - - @SuppressWarnings("unchecked") - @Override - public Object deserialize(InputStream inputStream, Class type) throws IOException { - AbstractCodec

codec = findDelegate(type); - if (codec != null) { - return codec.deserialize(inputStream); - } - else { - return defaultCodec.deserialize(inputStream, (Class

) type); - } - } - - @Override - public Object deserialize(byte[] bytes, Class type) throws IOException { - return deserialize(new ByteArrayInputStream(bytes), type); - } - - private AbstractCodec

findDelegate(Class type) { - if (delegates == null) { - return null; - } - - Class clazz = ClassUtils.findClosestMatch(type, delegates.keySet(), false); - return delegates.get(clazz); - } -} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/MultiTypeCodec.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/MultiTypeCodec.java deleted file mode 100644 index a4d8028e2..000000000 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/MultiTypeCodec.java +++ /dev/null @@ -1,48 +0,0 @@ -/* - * Copyright 2013 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.xd.dirt.integration.bus.serializer; - -import java.io.IOException; -import java.io.InputStream; - -import org.springframework.core.serializer.Serializer; - - -/** - * Interface for classes that perform both serialization and deserialization. - * @author David Turanski - */ -public interface MultiTypeCodec extends Serializer { - - /** - * Deserialize an object of a given type - * @param inputStream the input stream containing the serialized object - * @param type the object's class - * @return the object - * @throws IOException - */ - public abstract T deserialize(InputStream inputStream, Class type) throws IOException; - - /** - * Deserialize an object of a given type - * @param bytes the byte array containing the serialized object - * @param type the object's class - * @return the object - * @throws IOException - */ - public abstract T deserialize(byte[] bytes, Class type) throws IOException; -} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoCodec.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoCodec.java deleted file mode 100644 index a967f22b6..000000000 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoCodec.java +++ /dev/null @@ -1,117 +0,0 @@ -/* - * Copyright 2013 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.xd.dirt.integration.bus.serializer.kryo; - -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; - -import com.esotericsoftware.kryo.Kryo; -import com.esotericsoftware.kryo.io.Input; -import com.esotericsoftware.kryo.io.Output; -import com.esotericsoftware.kryo.pool.KryoCallback; -import com.esotericsoftware.kryo.pool.KryoFactory; -import com.esotericsoftware.kryo.pool.KryoPool; - -import org.springframework.util.Assert; -import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec; - -/** - * Base class for Codecs using {@link com.esotericsoftware.kryo.Kryo} - * @author David Turanski - */ -public abstract class AbstractKryoCodec implements MultiTypeCodec { - - protected final KryoPool pool; - - protected AbstractKryoCodec() { - KryoFactory factory = new KryoFactory() { - public Kryo create() { - Kryo kryo = new Kryo(); - // configure kryo instance, customize settings - configureKryoInstance(kryo); - return kryo; - } - }; - // Build pool with SoftReferences enabled (optional) - pool = new KryoPool.Builder(factory).softReferences().build(); - } - - /** - * Serialize an object using an existing output stream - * @param object the object to be serialized - * @param outputStream the output stream, e.g. a FileOutputStream - * @throws IOException - */ - - public void serialize(final Object object, OutputStream outputStream) throws IOException { - Assert.notNull(outputStream, "\'outputSteam\' cannot be null"); - final Output output = (outputStream instanceof Output ? (Output) outputStream : new Output(outputStream)); - this.pool.run(new KryoCallback() { - @SuppressWarnings("unchecked") - public Object execute(Kryo kryo) { - doSerialize(kryo, object, output); - return Void.class; - } - }); - output.close(); - } - - protected abstract void doSerialize(Kryo kryo, Object object, Output output); - - protected abstract Object doDeserialize(Kryo kryo, Input input, Class type); - - protected abstract void configureKryoInstance(Kryo kryo); - - /** - * Deserialize an object of a given type given a byte array - * @param bytes the byte array containing the serialized object - * @param type the object's class - * @return the object - * @throws IOException - */ - @Override - public Object deserialize(byte[] bytes, Class type) throws IOException { - final Input input = new Input(bytes); - try { - return deserialize(input, type); - } - finally { - input.close(); - } - } - - @Override - public Object deserialize(InputStream inputStream, final Class type) throws IOException { - Assert.notNull(inputStream, "\'inputStream\' cannot be null"); - final Input input = (inputStream instanceof Input ? (Input) inputStream : new Input(inputStream)); - Object result = null; - try { - result = this.pool.run(new KryoCallback() { - @SuppressWarnings("unchecked") - public Object execute(Kryo kryo) { - return doDeserialize(kryo, input, type); - } - }); - } - finally { - input.close(); - } - return result; - } - -} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoRegistrar.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoRegistrar.java deleted file mode 100644 index 1436a0bd5..000000000 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoRegistrar.java +++ /dev/null @@ -1,58 +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.xd.dirt.integration.bus.serializer.kryo; - -import java.util.List; - -import com.esotericsoftware.kryo.Kryo; -import com.esotericsoftware.kryo.Registration; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.springframework.core.serializer.support.SerializationFailedException; - -/** - * @author David Turanski - */ -public abstract class AbstractKryoRegistrar implements KryoRegistrar { - protected final Logger log = LoggerFactory.getLogger(this.getClass()); - - protected final static Kryo kryo = new Kryo(); - - @Override - public void registerTypes(Kryo kryo) { - for (Registration registration : getRegistrations()) { - register(kryo, registration); - } - } - - public abstract List getRegistrations(); - - protected void register(Kryo kryo, Registration registration) { - int id = registration.getId(); - - Registration existing = kryo.getRegistration(id); - - if (existing != null) { - throw new SerializationFailedException(String.format("registration already exists %s", existing)); - } - - log.info("registering {} with serializer {}", registration, registration.getSerializer().getClass() - .getName()); - - kryo.register(registration); - } -} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/CompositeKryoRegistrar.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/CompositeKryoRegistrar.java deleted file mode 100644 index e7d1d5d18..000000000 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/CompositeKryoRegistrar.java +++ /dev/null @@ -1,80 +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.xd.dirt.integration.bus.serializer.kryo; - -import java.util.ArrayList; -import java.util.List; - -import com.esotericsoftware.kryo.Registration; - -import org.springframework.core.serializer.support.SerializationFailedException; -import org.springframework.util.Assert; -import org.springframework.util.CollectionUtils; - -/** - * A {@link KryoRegistrar} that delegates and validates - * registrations across all components. - * @author David Turanski - * @since 1.2 - */ -public class CompositeKryoRegistrar extends AbstractKryoRegistrar { - - private final List delegates; - - public CompositeKryoRegistrar(List delegates) { - super(); - this.delegates = delegates; - - if (!CollectionUtils.isEmpty(this.delegates)) { - validateRegistrations(); - } - } - - @Override - public List getRegistrations() { - List registrations = new ArrayList<>(); - for (KryoRegistrar registrar : delegates) { - registrations.addAll(registrar.getRegistrations()); - } - return registrations; - } - - private void validateRegistrations() { - List ids = new ArrayList<>(); - List> types = new ArrayList<>(); - - for (Registration registration : getRegistrations()) { - Assert.isTrue(registration.getId() >= MIN_REGISTRATION_VALUE, "registration ID must be >= " + - MIN_REGISTRATION_VALUE); - if (ids.contains(registration.getId())) { - throw new SerializationFailedException(String.format("Duplicate registration ID found: %d", - registration.getId())); - } - ids.add(registration.getId()); - - if (types.contains(registration.getType())) { - throw new SerializationFailedException(String.format("Duplicate registration found for type: %s", - registration.getType())); - } - types.add(registration.getType()); - - log.info("configured Kryo registration {} with serializer {}", registration, - registration.getSerializer().getClass().getName()); - - } - - } -} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/FileKryoRegistrar.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/FileKryoRegistrar.java deleted file mode 100644 index 1f483540b..000000000 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/FileKryoRegistrar.java +++ /dev/null @@ -1,38 +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.xd.dirt.integration.bus.serializer.kryo; - -import java.util.Collections; -import java.util.List; - -import com.esotericsoftware.kryo.Registration; - -/** - * A {@link KryoRegistrar} used to register a File serializer. - * @author David Turanski - * @since 1.2 - */ -public class FileKryoRegistrar extends AbstractKryoRegistrar { - - private final static int FILE_REGISTRATION_ID = 40; - - private final FileSerializer fileSerializer = new FileSerializer(); - - @Override - public List getRegistrations() { - return Collections.singletonList(new Registration(java.io.File.class, fileSerializer, FILE_REGISTRATION_ID)); - } -} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/FileSerializer.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/FileSerializer.java deleted file mode 100644 index 607af6444..000000000 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/FileSerializer.java +++ /dev/null @@ -1,41 +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.xd.dirt.integration.bus.serializer.kryo; - -import java.io.File; - -import com.esotericsoftware.kryo.Kryo; -import com.esotericsoftware.kryo.Serializer; -import com.esotericsoftware.kryo.io.Input; -import com.esotericsoftware.kryo.io.Output; - -/** - * @author David Turanski - * @since 1.2 - */ -public class FileSerializer extends Serializer { - - @Override - public void write(Kryo kryo, Output output, File file) { - output.writeString(file.getPath()); - } - - @Override - public File read(Kryo kryo, Input input, Class type) { - String path = input.readString(); - return new File(path); - } -} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoClassListRegistrar.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoClassListRegistrar.java deleted file mode 100644 index a90ddd1c6..000000000 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoClassListRegistrar.java +++ /dev/null @@ -1,72 +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.xd.dirt.integration.bus.serializer.kryo; - -import java.util.ArrayList; -import java.util.List; - -import com.esotericsoftware.kryo.Registration; - - -import org.springframework.util.Assert; -import org.springframework.util.CollectionUtils; - -/** - * A {@link KryoRegistrar} used to register a - * list of Java classes. This assigns a sequential registration ID starting with an initial value (50 by default), but - * may be configured. This is easiest to set up but requires that every server node be configured with the identical - * list in the same order. - * @author David Turanski - * @since 1.1 - */ -public class KryoClassListRegistrar extends AbstractKryoRegistrar { - - private final List registeredClasses; - - private int initialValue = 50; - - /** - * @param classes the list of classes to register - */ - public KryoClassListRegistrar(List classes) { - this.registeredClasses = classes; - } - - /** - * Set the inital ID value. Classes in the list will be sequentially assigned an ID starting with this value - * (default is 50). - * @param initialValue the initial value - */ - public void setInitialValue(int initialValue) { - Assert.isTrue(initialValue >= MIN_REGISTRATION_VALUE, "'initialValue' must be >= " + - MIN_REGISTRATION_VALUE); - this.initialValue = initialValue; - } - - - @Override - public List getRegistrations() { - List registrations = new ArrayList<>(); - if (!CollectionUtils.isEmpty(registeredClasses)) { - for (int i = 0; i < registeredClasses.size(); i++) { - registrations.add(new Registration(registeredClasses.get(i), kryo.getSerializer(registeredClasses.get - (i)), i + initialValue)); - } - } - return registrations; - } - -} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoClassMapRegistrar.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoClassMapRegistrar.java deleted file mode 100644 index a26e24e64..000000000 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoClassMapRegistrar.java +++ /dev/null @@ -1,52 +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.xd.dirt.integration.bus.serializer.kryo; - -import java.util.ArrayList; -import java.util.List; -import java.util.Map; - -import com.esotericsoftware.kryo.Registration; - -import org.springframework.util.CollectionUtils; - - -/** - * A {@link KryoRegistrar} implementation backed by a Map - * used to explicitly set the registration ID for each class. - * @author David Turanski - * @since 1.1 - */ -public class KryoClassMapRegistrar extends AbstractKryoRegistrar { - - final private Map> registeredClasses; - - public KryoClassMapRegistrar(Map> kryoRegisteredClasses) { - this.registeredClasses = kryoRegisteredClasses; - } - - - @Override - public List getRegistrations() { - List registrations = new ArrayList<>(); - if (!CollectionUtils.isEmpty(registeredClasses)) { - for (Map.Entry> entry : registeredClasses.entrySet()) { - registrations.add(new Registration(entry.getValue(), kryo.getSerializer(entry.getValue()), entry.getKey())); - } - } - return registrations; - } -} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoRegistrar.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoRegistrar.java deleted file mode 100644 index 15af67a48..000000000 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoRegistrar.java +++ /dev/null @@ -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.xd.dirt.integration.bus.serializer.kryo; - -import java.util.List; - -import com.esotericsoftware.kryo.Kryo; -import com.esotericsoftware.kryo.Registration; - -/** - * Strategy interface used by {@link PojoCodec} to register - * classes consistently across {@link Kryo} instances. An XD user may register an instance of this type in the Spring XD - * Application Context to enable kryo class registration which results in efficiency gains if you know the types your - * application needs in advance. Note that Kryo serialization only applies to types used as message payloads in XD - * streams. - * By default, user defined types are not registered to Kryo. Registration allows a unique ID (small positive integer is - * ideal) to represent the type in the byte stream. In a distributed environment, all Kryo instances must maintain the - * same registration state in order to properly take advantage of this feature. - * This is can result in better performance in demanding situations, but requires some care to maintain. Only use this - * if you really need it. Otherwise, it is a great example of premature optimization. - * This interface applies a strategy to register a statically configured, one-to-one mapping of a Java type to an - * integer. Basic implementations are provided backed by a Map> or a List>. These are simple - * and require the user to manually configure a bean in each XD server and ensure that the configuration is always - * consistent.* - * The container looks in classpath*:META-INF/spring-xd/xd/bus/ext/*.xml for an instance of this type named - * "kryoRegistrar". The KryoRegistrar provides the registration mapping and the strategy to apply the mapping to every - * Kryo instance. Note that statically declared Java types must also be present in the XD class path (xd/lib) else the - * container will fail to initialize. Only one instance may be registered and identically configured across all - * containers. - * - * @author David Turanski - * @since 1.1 - */ -public interface KryoRegistrar { - - static final int MIN_REGISTRATION_VALUE = 10; - - /** - * This method is invoked by the {@link PojoCodec} and - * applied to the {@link Kryo} instance whenever one is provided. This is currently done using an object pool so it - * is inevitable that this method will be invoked repeatedly on the same instance. Kryo registration is idempotent, - * but this could become inefficient if registering a large amount of types. - * - * @param kryo the provided instance - */ - void registerTypes(Kryo kryo); - - /** - * - * @return the list of {@link com.esotericsoftware.kryo.Registration} provided - */ - List getRegistrations(); -} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoRegistrationRegistrar.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoRegistrationRegistrar.java deleted file mode 100644 index 9e69f3e3c..000000000 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoRegistrationRegistrar.java +++ /dev/null @@ -1,39 +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.xd.dirt.integration.bus.serializer.kryo; - -import java.util.ArrayList; -import java.util.List; - -import com.esotericsoftware.kryo.Registration; - -/** - * A {@link KryoRegistrar } implementation backed by a List of {@link com.esotericsoftware.kryo.Registration}. - * @author David Turanski - * @since 1.2 - */ -public class KryoRegistrationRegistrar extends AbstractKryoRegistrar { - private final List registrations; - - public KryoRegistrationRegistrar(List registrations) { - this.registrations = registrations != null ? registrations : new ArrayList(); - } - - @Override - public List getRegistrations() { - return registrations; - } -} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/PojoCodec.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/PojoCodec.java deleted file mode 100644 index a532ba8e3..000000000 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/PojoCodec.java +++ /dev/null @@ -1,104 +0,0 @@ -/* - * Copyright 2013 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.xd.dirt.integration.bus.serializer.kryo; - - -import java.util.Collections; -import java.util.List; - -import com.esotericsoftware.kryo.Kryo; -import com.esotericsoftware.kryo.io.Input; -import com.esotericsoftware.kryo.io.Output; - -import org.springframework.util.CollectionUtils; - -/** - * Kryo Codec that can serialize and deserialize arbitrary types. Classes and associated - * {@link com.esotericsoftware.kryo.Serializer}s may be registered via - * {@link org.springframework.xd.dirt.integration.bus.serializer.kryo.KryoRegistrar}s. - * @author David Turanski - * @since 1.0 - */ -public class PojoCodec extends AbstractKryoCodec { - private final CompositeKryoRegistrar kryoRegistrar; - - private final boolean useReferences; - - public PojoCodec() { - this.kryoRegistrar = null; - this.useReferences = true; - } - - /** - * Create an instance with a single KryoRegistrar. - * @param kryoRegistrar the registrar. - */ - public PojoCodec(KryoRegistrar kryoRegistrar) { - this(kryoRegistrar != null ? Collections.singletonList(kryoRegistrar) : null, true); - } - - /** - * Create an instance with zero to many KryoRegistrars. - * @param kryoRegistrars a list KryoRegistrars. - */ - public PojoCodec(List kryoRegistrars) { - this.kryoRegistrar = CollectionUtils.isEmpty(kryoRegistrars) ? null : - new CompositeKryoRegistrar(kryoRegistrars); - this.useReferences = true; - } - - /** - * Create an instance with a single KryoRegistrar. - * @param kryoRegistrar the registrar. - * @param useReferences set to false if references are not required (if the object graph is known to be acyclical). - * The default is 'true' which is less performant but more flexible. - */ - public PojoCodec(KryoRegistrar kryoRegistrar, boolean useReferences) { - this(kryoRegistrar != null ? Collections.singletonList(kryoRegistrar) : null, useReferences); - } - - /** - * Create an instance with zero to many KryoRegistrars. - * @param kryoRegistrars a list KryoRegistrars. - * @param useReferences set to false if references are not required (if the object graph is known to be acyclical). - * The default is 'true' which is less performant but more flexible. - */ - public PojoCodec(List kryoRegistrars, boolean useReferences) { - kryoRegistrar = CollectionUtils.isEmpty(kryoRegistrars) ? null : - new CompositeKryoRegistrar(kryoRegistrars); - this.useReferences = useReferences; - } - - @Override - protected void doSerialize(Kryo kryo, Object object, Output output) { - kryo.writeObject(output, object); - } - - - @Override - protected Object doDeserialize(Kryo kryo, Input input, Class type) { - return kryo.readObject(input, type); - } - - @Override - protected void configureKryoInstance(Kryo kryo) { - if (kryoRegistrar != null) { - kryoRegistrar.registerTypes(kryo); - } - kryo.setReferences(useReferences); - } -} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/package-info.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/package-info.java deleted file mode 100644 index a80e77575..000000000 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/package-info.java +++ /dev/null @@ -1,5 +0,0 @@ -/** - * Contains classes that provide kryo serialization support to/from {@link org.springframework.xd.dirt.integration.bus.MessageBus}. - */ - -package org.springframework.xd.dirt.integration.bus.serializer.kryo; diff --git a/spring-cloud-stream-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/CompositeCodecTests.java b/spring-cloud-stream-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/CompositeCodecTests.java deleted file mode 100644 index 789ad1729..000000000 --- a/spring-cloud-stream-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/CompositeCodecTests.java +++ /dev/null @@ -1,90 +0,0 @@ -/* - * Copyright 2013 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.xd.dirt.integration.bus.serializer.kryo; - -import static org.junit.Assert.assertEquals; - -import java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.util.HashMap; -import java.util.Map; - -import org.junit.Before; -import org.junit.Test; - -import org.springframework.xd.dirt.integration.bus.serializer.AbstractCodec; -import org.springframework.xd.dirt.integration.bus.serializer.CompositeCodec; -import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec; - -/** - * @author David Turanski - */ -public class CompositeCodecTests { - - private MultiTypeCodec codec; - - - @SuppressWarnings({ "unchecked", "rawtypes" }) - @Before - public void setup() { - Map, AbstractCodec> codecs = new HashMap<>(); - codec = new CompositeCodec(codecs, new PojoCodec()); - } - - @Test - public void testPojoSerialization() throws IOException { - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - SomeClassWithNoDefaultConstructors foo = new SomeClassWithNoDefaultConstructors("hello", 123); - codec.serialize(foo, bos); - SomeClassWithNoDefaultConstructors foo2 = (SomeClassWithNoDefaultConstructors) codec.deserialize( - bos.toByteArray(), - SomeClassWithNoDefaultConstructors.class); - assertEquals(foo, foo2); - } - - static class SomeClassWithNoDefaultConstructors { - - private String val1; - - private int val2; - - public SomeClassWithNoDefaultConstructors(String val1) { - this.val1 = val1; - } - - public SomeClassWithNoDefaultConstructors(String val1, int val2) { - this.val1 = val1; - this.val2 = val2; - } - - @Override - public boolean equals(Object other) { - if (!(other instanceof SomeClassWithNoDefaultConstructors)) { - return false; - } - SomeClassWithNoDefaultConstructors that = (SomeClassWithNoDefaultConstructors) other; - return (this.val1.equals(that.val1) && val2 == that.val2); - } - - @Override - public int hashCode() { - int result = this.val1.hashCode(); - result = 31 * result + val2; - return result; - } - } -} diff --git a/spring-cloud-stream-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoCodecTests.java b/spring-cloud-stream-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoCodecTests.java deleted file mode 100644 index 94601713d..000000000 --- a/spring-cloud-stream-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoCodecTests.java +++ /dev/null @@ -1,173 +0,0 @@ -/* - * Copyright 2013 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.xd.dirt.integration.bus.serializer.kryo; - -import static org.junit.Assert.assertEquals; - -import java.io.ByteArrayOutputStream; -import java.io.File; -import java.io.FileInputStream; -import java.io.FileOutputStream; -import java.io.IOException; -import java.util.HashMap; -import java.util.Map; - -import org.junit.Test; - -/** - * @author David Turanski - * @since 1.0 - */ -public class KryoCodecTests { - - @Test - public void testStringSerialization() throws IOException { - String str = "hello"; - PojoCodec serializer = new PojoCodec(); - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - - serializer.serialize(str, bos); - - String s2 = (String)serializer.deserialize(bos.toByteArray(), String.class); - assertEquals(str, s2); - } - - @Test - public void testSerializationWithStreams() throws IOException { - String str = "hello"; - File file = new File("test.ser"); - PojoCodec serializer = new PojoCodec(); - FileOutputStream fos = new FileOutputStream(file); - serializer.serialize(str, fos); - fos.close(); - - FileInputStream fis = new FileInputStream(file); - String s2 = (String) serializer.deserialize(fis, String.class); - file.delete(); - assertEquals(str, s2); - } - - @Test - public void testPojoSerialization() throws IOException { - PojoCodec serializer = new PojoCodec(); - SomeClassWithNoDefaultConstructors foo = new SomeClassWithNoDefaultConstructors("foo", 123); - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - serializer.serialize(foo, bos); - Object foo2 = serializer.deserialize(bos.toByteArray(), SomeClassWithNoDefaultConstructors.class); - assertEquals(foo, foo2); - } - - static class SomeClassWithNoDefaultConstructors { - - private String val1; - - private int val2; - - public SomeClassWithNoDefaultConstructors(String val1, int val2) { - this.val1 = val1; - this.val2 = val2; - } - - @Override - public boolean equals(Object other) { - if (!(other instanceof SomeClassWithNoDefaultConstructors)) { - return false; - } - SomeClassWithNoDefaultConstructors that = (SomeClassWithNoDefaultConstructors) other; - return (this.val1.equals(that.val1) && val2 == that.val2); - } - - @Override - public int hashCode() { - int result = val1.hashCode(); - result = 31 * result + val2; - return result; - } - } - - @Test - public void testPrimitiveSerialization() throws IOException { - PojoCodec serializer = new PojoCodec(); - - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - serializer.serialize(true, bos); - boolean b = (Boolean) serializer.deserialize(bos.toByteArray(), Boolean.class); - assertEquals(true, b); - b = (Boolean) serializer.deserialize(bos.toByteArray(), boolean.class); - assertEquals(true, b); - - bos = new ByteArrayOutputStream(); - serializer.serialize(3.14159, bos); - - double d = (Double) serializer.deserialize(bos.toByteArray(), double.class); - assertEquals(3.14159, d, 0.00001); - - bos = new ByteArrayOutputStream(); - serializer.serialize(new Double(3.14159), bos); - - d = (Double) serializer.deserialize(bos.toByteArray(), Double.class); - assertEquals(3.14159, d, 0.00001); - - } - - @Test - public void testMapSerialization() throws IOException { - PojoCodec serializer = new PojoCodec(); - Map map = new HashMap(); - map.put("one", 1); - map.put("two", 2); - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - serializer.serialize(map, bos); - Map m2 = (Map) serializer.deserialize(bos.toByteArray(), HashMap.class); - assertEquals(2, m2.size()); - assertEquals(1, m2.get("one")); - assertEquals(2, m2.get("two")); - } - - @Test - public void testComplexObjectSerialization() throws IOException { - PojoCodec serializer = new PojoCodec(); - Foo foo = new Foo(); - foo.put("one", 1); - foo.put("two", 2); - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - serializer.serialize(foo, bos); - - Foo foo2 = (Foo) serializer.deserialize(bos.toByteArray(), Foo.class); - assertEquals(1, foo2.get("one")); - assertEquals(2, foo2.get("two")); - } - - - - static class Foo { - - private Map map; - - public Foo() { - map = new HashMap(); - } - - public void put(Object key, Object value) { - map.put(key, value); - } - - public Object get(Object key) { - return map.get(key); - } - } -} diff --git a/spring-cloud-stream-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoFileCodecTests.java b/spring-cloud-stream-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoFileCodecTests.java deleted file mode 100644 index 6bd38eb88..000000000 --- a/spring-cloud-stream-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoFileCodecTests.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * Copyright 2013 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.xd.dirt.integration.bus.serializer.kryo; - -import java.io.ByteArrayOutputStream; -import java.io.File; -import java.io.IOException; - -import org.junit.Test; - -import static org.junit.Assert.assertEquals; - - -/** - * @author David Turanski - */ -public class KryoFileCodecTests { - - @Test - public void test() throws IOException { - - PojoCodec pc = new PojoCodec(new FileKryoRegistrar()); - File file = new File("/foo/bar"); - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - pc.serialize(file, bos); - File file2 = (File) pc.deserialize(bos.toByteArray(), File.class); - assertEquals(file, file2); - } -} diff --git a/spring-cloud-stream/pom.xml b/spring-cloud-stream/pom.xml index 5085ff967..6b1fb51eb 100644 --- a/spring-cloud-stream/pom.xml +++ b/spring-cloud-stream/pom.xml @@ -43,10 +43,6 @@ org.springframework.cloud spring-cloud-lattice-connector - - org.springframework.cloud - spring-cloud-stream-codec - org.springframework.cloud spring-cloud-stream-binder-local diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java index b5af638c1..db18b30e1 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java @@ -25,7 +25,6 @@ import java.lang.annotation.Target; import org.springframework.cloud.stream.config.AggregateBuilderConfiguration; import org.springframework.cloud.stream.config.ChannelBindingAdapterConfiguration; -import org.springframework.cloud.stream.config.CodecConfiguration; import org.springframework.cloud.stream.config.ModuleRegistrar; import org.springframework.cloud.stream.config.ChannelBindingAdapterRunner; import org.springframework.cloud.stream.config.RabbitServiceConfiguration; @@ -47,7 +46,7 @@ import org.springframework.integration.annotation.MessageEndpoint; @Configuration @MessageEndpoint @Import({RedisServiceConfiguration.class, RabbitServiceConfiguration.class, - ChannelBindingAdapterConfiguration.class, CodecConfiguration.class, ChannelBindingAdapterRunner.class, + ChannelBindingAdapterConfiguration.class, ChannelBindingAdapterRunner.class, AggregateBuilderConfiguration.class, ModuleRegistrar.class}) public @interface EnableModule { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/RabbitServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/RabbitServiceConfiguration.java index 85024c0d1..315f35ead 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/RabbitServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/RabbitServiceConfiguration.java @@ -21,12 +21,13 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.cloud.Cloud; import org.springframework.cloud.CloudFactory; +import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder; +import org.springframework.cloud.stream.binder.rabbit.config.RabbitMessageChannelBinderConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.context.annotation.ImportResource; +import org.springframework.context.annotation.Import; import org.springframework.context.annotation.Profile; import org.springframework.context.annotation.PropertySource; -import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder; /** * Bind to services, either locally or in a Lattice environment. @@ -34,11 +35,12 @@ import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder * @author Mark Fisher * @author Dave Syer * @author Glenn Renfro + * @author David Turanski */ @Configuration @ConditionalOnClass(RabbitMessageChannelBinder.class) @ConditionalOnMissingBean(RabbitMessageChannelBinder.class) -@ImportResource("classpath*:/META-INF/spring-cloud-stream/binder/rabbit-binder.xml") +@Import(RabbitMessageChannelBinderConfiguration.class) @PropertySource("classpath:/META-INF/spring-cloud-stream/rabbit-binder.properties") public class RabbitServiceConfiguration { @Configuration @@ -48,6 +50,7 @@ public class RabbitServiceConfiguration { public Cloud cloud() { return new CloudFactory().getCloud(); } + @Bean ConnectionFactory rabbitConnectionFactory(Cloud cloud) { return cloud.getSingletonServiceConnector(ConnectionFactory.class, null); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/RedisServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/RedisServiceConfiguration.java index b0f6066eb..174d1db47 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/RedisServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/RedisServiceConfiguration.java @@ -21,8 +21,10 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean import org.springframework.cloud.Cloud; import org.springframework.cloud.CloudFactory; import org.springframework.cloud.stream.binder.redis.RedisMessageChannelBinder; +import org.springframework.cloud.stream.binder.redis.config.RedisMessageChannelBinderConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; import org.springframework.context.annotation.ImportResource; import org.springframework.context.annotation.Profile; import org.springframework.context.annotation.PropertySource; @@ -33,12 +35,13 @@ import org.springframework.data.redis.connection.RedisConnectionFactory; * * @author Mark Fisher * @author Dave Syer + * @author David Turanski */ @Configuration @ConditionalOnClass(RedisMessageChannelBinder.class) @ConditionalOnMissingBean(RedisMessageChannelBinder.class) -@ImportResource({ "classpath*:/META-INF/spring-cloud-stream/binder/redis-binder.xml", -"classpath*:/META-INF/spring-xd/analytics/redis-analytics.xml" }) +@Import(RedisMessageChannelBinderConfiguration.class) +@ImportResource("classpath*:/META-INF/spring-xd/analytics/redis-analytics.xml") @PropertySource("classpath:/META-INF/spring-cloud-stream/redis-binder.properties") public class RedisServiceConfiguration { diff --git a/spring-cloud-stream/src/main/resources/META-INF/spring-cloud-stream/rabbit-binder.properties b/spring-cloud-stream/src/main/resources/META-INF/spring-cloud-stream/rabbit-binder.properties index 18cdb037b..d7f95153d 100644 --- a/spring-cloud-stream/src/main/resources/META-INF/spring-cloud-stream/rabbit-binder.properties +++ b/spring-cloud-stream/src/main/resources/META-INF/spring-cloud-stream/rabbit-binder.properties @@ -1,24 +1,23 @@ -#TODO: New naming convention -xd.messagebus.rabbit.default.ackMode: AUTO -xd.messagebus.rabbit.default.autoBindDLQ: false -xd.messagebus.rabbit.default.backOffInitialInterval: 1000 -xd.messagebus.rabbit.default.backOffMaxInterval: 10000 -xd.messagebus.rabbit.default.backOffMultiplier: 2.0 -xd.messagebus.rabbit.default.batchBufferLimit: 10000 -xd.messagebus.rabbit.default.batchingEnabled: false -xd.messagebus.rabbit.default.batchSize: 100 -xd.messagebus.rabbit.default.batchTimeout: 5000 -xd.messagebus.rabbit.default.compress: false -xd.messagebus.rabbit.default.concurrency: 1 -xd.messagebus.rabbit.default.deliveryMode: PERSISTENT -xd.messagebus.rabbit.default.durableSubscription: false -xd.messagebus.rabbit.default.maxAttempts: 3 -xd.messagebus.rabbit.default.maxConcurrency: 1 -xd.messagebus.rabbit.default.prefix: xdbus. -xd.messagebus.rabbit.default.prefetch: 1 -xd.messagebus.rabbit.default.replyHeaderPatterns: STANDARD_REPLY_HEADERS,* -xd.messagebus.rabbit.default.republishToDLQ: false -xd.messagebus.rabbit.default.requestHeaderPatterns: STANDARD_REQUEST_HEADERS,* -xd.messagebus.rabbit.default.requeue: true -xd.messagebus.rabbit.default.transacted:false -xd.messagebus.rabbit.default.txSize: 1 +spring.cloud.stream.binder.rabbit.default.acknowledgeMode: AUTO +spring.cloud.stream.binder.rabbit.default.autoBindDLQ: false +spring.cloud.stream.binder.rabbit.default.backOffInitialInterval: 1000 +spring.cloud.stream.binder.rabbit.default.backOffMaxInterval: 10000 +spring.cloud.stream.binder.rabbit.default.backOffMultiplier: 2.0 +spring.cloud.stream.binder.rabbit.default.batchBufferLimit: 10000 +spring.cloud.stream.binder.rabbit.default.batchingEnabled: false +spring.cloud.stream.binder.rabbit.default.batchSize: 100 +spring.cloud.stream.binder.rabbit.default.batchTimeout: 5000 +spring.cloud.stream.binder.rabbit.default.compress: false +spring.cloud.stream.binder.rabbit.default.concurrency: 1 +spring.cloud.stream.binder.rabbit.default.deliveryMode: PERSISTENT +spring.cloud.stream.binder.rabbit.default.durableSubscription: false +spring.cloud.stream.binder.rabbit.default.maxAttempts: 3 +spring.cloud.stream.binder.rabbit.default.maxConcurrency: 1 +spring.cloud.stream.binder.rabbit.default.prefix: xdbus. +spring.cloud.stream.binder.rabbit.default.prefetch: 1 +spring.cloud.stream.binder.rabbit.default.replyHeaderPatterns: STANDARD_REPLY_HEADERS,* +spring.cloud.stream.binder.rabbit.default.republishToDLQ: false +spring.cloud.stream.binder.rabbit.default.requestHeaderPatterns: STANDARD_REQUEST_HEADERS,* +spring.cloud.stream.binder.rabbit.default.requeue: true +spring.cloud.stream.binder.rabbit.default.transacted:false +spring.cloud.stream.binder.rabbit.default.txSize: 1 diff --git a/spring-cloud-stream/src/main/resources/META-INF/spring-cloud-stream/redis-binder.properties b/spring-cloud-stream/src/main/resources/META-INF/spring-cloud-stream/redis-binder.properties index 578bb5712..dd78f8e81 100644 --- a/spring-cloud-stream/src/main/resources/META-INF/spring-cloud-stream/redis-binder.properties +++ b/spring-cloud-stream/src/main/resources/META-INF/spring-cloud-stream/redis-binder.properties @@ -1,6 +1,5 @@ -#TODO: New naming convention -xd.messagebus.redis.default.backOffInitialInterval: 1000 -xd.messagebus.redis.default.backOffMaxInterval: 10000 -xd.messagebus.redis.default.backOffMultiplier: 2.0 -xd.messagebus.redis.default.concurrency: 1 -xd.messagebus.redis.default.maxAttempts: 3 +spring.cloud.stream.binder.redis.default.backOffInitialInterval: 1000 +spring.cloud.stream.binder.redis.default.backOffMaxInterval: 10000 +spring.cloud.stream.binder.redis.default.backOffMultiplier: 2.0 +spring.cloud.stream.binder.redis.default.concurrency: 1 +spring.cloud.stream.binder.redis.default.maxAttempts: 3