diff --git a/multibinder-differentsystems/pom.xml b/multibinder-differentsystems/pom.xml
index 8aec847..5ec4d97 100644
--- a/multibinder-differentsystems/pom.xml
+++ b/multibinder-differentsystems/pom.xml
@@ -45,19 +45,6 @@
spring-cloud-stream-binder-kafka-test-support
test
-
- org.apache.kafka
- kafka_2.11
- test
- 0.9.0.1
- test
-
-
- org.slf4j
- slf4j-log4j12
-
-
-
org.apache.curator
curator-test
diff --git a/multibinder-differentsystems/src/main/resources/application.yml b/multibinder-differentsystems/src/main/resources/application.yml
index 85cbda7..9844b4f 100644
--- a/multibinder-differentsystems/src/main/resources/application.yml
+++ b/multibinder-differentsystems/src/main/resources/application.yml
@@ -7,6 +7,7 @@ spring:
input:
destination: dataIn
binder: kafka1
+ group: testGroup
output:
destination: dataOut
binder: kafka2
diff --git a/multibinder/src/test/java/multibinder/RabbitAndKafkaBinderApplicationTests.java b/multibinder/src/test/java/multibinder/RabbitAndKafkaBinderApplicationTests.java
index 7ecfe01..38621fc 100644
--- a/multibinder/src/test/java/multibinder/RabbitAndKafkaBinderApplicationTests.java
+++ b/multibinder/src/test/java/multibinder/RabbitAndKafkaBinderApplicationTests.java
@@ -1,106 +1,107 @@
-/*
- * Copyright 2015-2017 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 multibinder;
-
-import java.util.UUID;
-
-import org.hamcrest.CoreMatchers;
-import org.hamcrest.Matchers;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.ClassRule;
-import org.junit.Test;
-
-import org.springframework.amqp.rabbit.core.RabbitAdmin;
-import org.springframework.boot.SpringApplication;
-import org.springframework.cloud.stream.binder.BinderFactory;
-import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
-import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
-import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder;
-import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
-import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
-import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
-import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport;
-import org.springframework.context.ConfigurableApplicationContext;
-import org.springframework.integration.channel.DirectChannel;
-import org.springframework.integration.channel.QueueChannel;
-import org.springframework.kafka.test.rule.KafkaEmbedded;
-import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageChannel;
-import org.springframework.messaging.support.MessageBuilder;
-import org.springframework.test.annotation.DirtiesContext;
-
-/**
- * @author Marius Bogoevici
- * @author Gary Russell
- */
-@DirtiesContext
-public class RabbitAndKafkaBinderApplicationTests {
-
- @ClassRule
- public static RabbitTestSupport rabbitTestSupport = new RabbitTestSupport();
-
- @ClassRule
- public static KafkaEmbedded kafkaEmbedded = new KafkaEmbedded(1, true, "test");
-
- private final String randomGroup = UUID.randomUUID().toString();
-
- @After
- public void cleanUp() {
- RabbitAdmin admin = new RabbitAdmin(rabbitTestSupport.getResource());
- admin.deleteQueue("binder.dataOut.default");
- admin.deleteQueue("binder.dataOut." + this.randomGroup);
- admin.deleteExchange("binder.dataOut");
- }
-
- @Test
- public void contextLoads() throws Exception {
- // passing connection arguments arguments to the embedded Kafka instance
- ConfigurableApplicationContext context = SpringApplication.run(MultibinderApplication.class,
- "--spring.cloud.stream.kafka.binder.brokers=" + kafkaEmbedded.getBrokersAsString(),
- "--spring.cloud.stream.kafka.binder.zkNodes=" + kafkaEmbedded.getZookeeperConnectionString());
- context.close();
- }
-
- @Test
- public void messagingWorks() throws Exception {
- // passing connection arguments arguments to the embedded Kafka instance
- ConfigurableApplicationContext context = SpringApplication.run(MultibinderApplication.class,
- "--spring.cloud.stream.kafka.binder.brokers=" + kafkaEmbedded.getBrokersAsString(),
- "--spring.cloud.stream.kafka.binder.zkNodes=" + kafkaEmbedded.getZookeeperConnectionString(),
- "--spring.cloud.stream.bindings.output.producer.requiredGroups=" + this.randomGroup);
- DirectChannel dataProducer = new DirectChannel();
- BinderFactory binderFactory = context.getBean(BinderFactory.class);
-
- QueueChannel dataConsumer = new QueueChannel();
-
- ((RabbitMessageChannelBinder) binderFactory.getBinder("rabbit", MessageChannel.class)).bindConsumer("dataOut", this.randomGroup,
- dataConsumer, new ExtendedConsumerProperties<>(new RabbitConsumerProperties()));
-
- ((KafkaMessageChannelBinder) binderFactory.getBinder("kafka", MessageChannel.class))
- .bindProducer("dataIn", dataProducer, new ExtendedProducerProperties<>(new KafkaProducerProperties()));
-
- String testPayload = "testFoo" + this.randomGroup;
- dataProducer.send(MessageBuilder.withPayload(testPayload).build());
-
- Message receive = dataConsumer.receive(60_000);
- Assert.assertThat(receive, Matchers.notNullValue());
- Assert.assertThat(receive.getPayload(), CoreMatchers.equalTo(testPayload));
- context.close();
- }
-
-}
+/*
+ * Copyright 2015-2017 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 multibinder;
+
+import java.util.UUID;
+
+import org.hamcrest.CoreMatchers;
+import org.hamcrest.Matchers;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.boot.SpringApplication;
+import org.springframework.cloud.stream.binder.BinderFactory;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
+import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
+import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
+import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.kafka.test.rule.KafkaEmbedded;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.test.annotation.DirtiesContext;
+
+/**
+ * @author Marius Bogoevici
+ * @author Gary Russell
+ */
+@DirtiesContext
+public class RabbitAndKafkaBinderApplicationTests {
+
+ @ClassRule
+ public static RabbitTestSupport rabbitTestSupport = new RabbitTestSupport();
+
+ @ClassRule
+ public static KafkaEmbedded kafkaEmbedded = new KafkaEmbedded(1, true, "test");
+
+ private final String randomGroup = UUID.randomUUID().toString();
+
+ @After
+ public void cleanUp() {
+ RabbitAdmin admin = new RabbitAdmin(rabbitTestSupport.getResource());
+ admin.deleteQueue("binder.dataOut.default");
+ admin.deleteQueue("binder.dataOut." + this.randomGroup);
+ admin.deleteExchange("binder.dataOut");
+ }
+
+ @Test
+ public void contextLoads() throws Exception {
+ // passing connection arguments arguments to the embedded Kafka instance
+ ConfigurableApplicationContext context = SpringApplication.run(MultibinderApplication.class,
+ "--spring.cloud.stream.kafka.binder.brokers=" + kafkaEmbedded.getBrokersAsString(),
+ "--spring.cloud.stream.kafka.binder.zkNodes=" + kafkaEmbedded.getZookeeperConnectionString());
+ context.close();
+ }
+
+ @Test
+ public void messagingWorks() throws Exception {
+ // passing connection arguments arguments to the embedded Kafka instance
+ ConfigurableApplicationContext context = SpringApplication.run(MultibinderApplication.class,
+ "--spring.cloud.stream.kafka.binder.brokers=" + kafkaEmbedded.getBrokersAsString(),
+ "--spring.cloud.stream.kafka.binder.zkNodes=" + kafkaEmbedded.getZookeeperConnectionString(),
+ "--spring.cloud.stream.bindings.input.group=testGroup",
+ "--spring.cloud.stream.bindings.output.producer.requiredGroups=" + this.randomGroup);
+ DirectChannel dataProducer = new DirectChannel();
+ BinderFactory binderFactory = context.getBean(BinderFactory.class);
+
+ QueueChannel dataConsumer = new QueueChannel();
+
+ ((RabbitMessageChannelBinder) binderFactory.getBinder("rabbit", MessageChannel.class)).bindConsumer("dataOut", this.randomGroup,
+ dataConsumer, new ExtendedConsumerProperties<>(new RabbitConsumerProperties()));
+
+ ((KafkaMessageChannelBinder) binderFactory.getBinder("kafka", MessageChannel.class))
+ .bindProducer("dataIn", dataProducer, new ExtendedProducerProperties<>(new KafkaProducerProperties()));
+
+ String testPayload = "testFoo" + this.randomGroup;
+ dataProducer.send(MessageBuilder.withPayload(testPayload).build());
+
+ Message receive = dataConsumer.receive(60_000);
+ Assert.assertThat(receive, Matchers.notNullValue());
+ Assert.assertThat(receive.getPayload(), CoreMatchers.equalTo(testPayload));
+ context.close();
+ }
+
+}
diff --git a/pom.xml b/pom.xml
index 67246f6..7604dc7 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1,139 +1,139 @@
-
-
- 4.0.0
- org.springframework.cloud
- spring-cloud-stream-samples
- 1.2.0.BUILD-SNAPSHOT
- pom
- https://github.com/spring-cloud/spring-cloud-stream-samples
-
- Pivotal Software, Inc.
- http://www.spring.io
-
-
- org.springframework.cloud
- spring-cloud-build
- 1.3.1.RELEASE
-
-
- UTF-8
- Chelsea.RELEASE
- 1.8
-
-
- source
- dynamic-source
- sink
- transform
- double
- non-self-contained-aggregate-app
- multibinder
- multibinder-differentsystems
- rxjava-processor
- multi-io
- stream-listener
- reactive-processor-kafka
-
-
-
-
- org.springframework.cloud
- spring-cloud-stream-dependencies
- ${spring-cloud-stream.version}
- pom
- import
-
-
- org.springframework.cloud
- spring-cloud-stream-sample-source
- 1.2.0.BUILD-SNAPSHOT
-
-
- org.springframework.cloud
- spring-cloud-stream-sample-sink
- 1.2.0.BUILD-SNAPSHOT
-
-
- org.springframework.cloud
- spring-cloud-stream-sample-transform
- 1.2.0.BUILD-SNAPSHOT
-
-
-
-
-
-
-
-
- maven-deploy-plugin
-
- true
-
-
-
-
-
-
-
- spring
-
-
- spring-snapshots
- Spring Snapshots
- http://repo.spring.io/libs-snapshot-local
-
- true
-
-
- false
-
-
-
- spring-milestones
- Spring Milestones
- http://repo.spring.io/libs-milestone-local
-
- false
-
-
-
- spring-releases
- Spring Releases
- http://repo.spring.io/release
-
- false
-
-
-
-
-
- spring-snapshots
- Spring Snapshots
- http://repo.spring.io/libs-snapshot-local
-
- true
-
-
- false
-
-
-
- spring-milestones
- Spring Milestones
- http://repo.spring.io/libs-milestone-local
-
- false
-
-
-
- spring-releases
- Spring Releases
- http://repo.spring.io/libs-release-local
-
- false
-
-
-
-
-
-
+
+
+ 4.0.0
+ org.springframework.cloud
+ spring-cloud-stream-samples
+ 1.2.0.BUILD-SNAPSHOT
+ pom
+ https://github.com/spring-cloud/spring-cloud-stream-samples
+
+ Pivotal Software, Inc.
+ http://www.spring.io
+
+
+ org.springframework.cloud
+ spring-cloud-build
+ 1.3.1.RELEASE
+
+
+ UTF-8
+ Chelsea.BUILD-SNAPSHOT
+ 1.8
+
+
+ source
+ dynamic-source
+ sink
+ transform
+ double
+ non-self-contained-aggregate-app
+ multibinder
+ multibinder-differentsystems
+ rxjava-processor
+ multi-io
+ stream-listener
+ reactive-processor-kafka
+
+
+
+
+ org.springframework.cloud
+ spring-cloud-stream-dependencies
+ ${spring-cloud-stream.version}
+ pom
+ import
+
+
+ org.springframework.cloud
+ spring-cloud-stream-sample-source
+ 1.2.0.BUILD-SNAPSHOT
+
+
+ org.springframework.cloud
+ spring-cloud-stream-sample-sink
+ 1.2.0.BUILD-SNAPSHOT
+
+
+ org.springframework.cloud
+ spring-cloud-stream-sample-transform
+ 1.2.0.BUILD-SNAPSHOT
+
+
+
+
+
+
+
+
+ maven-deploy-plugin
+
+ true
+
+
+
+
+
+
+
+ spring
+
+
+ spring-snapshots
+ Spring Snapshots
+ http://repo.spring.io/libs-snapshot-local
+
+ true
+
+
+ false
+
+
+
+ spring-milestones
+ Spring Milestones
+ http://repo.spring.io/libs-milestone-local
+
+ false
+
+
+
+ spring-releases
+ Spring Releases
+ http://repo.spring.io/release
+
+ false
+
+
+
+
+
+ spring-snapshots
+ Spring Snapshots
+ http://repo.spring.io/libs-snapshot-local
+
+ true
+
+
+ false
+
+
+
+ spring-milestones
+ Spring Milestones
+ http://repo.spring.io/libs-milestone-local
+
+ false
+
+
+
+ spring-releases
+ Spring Releases
+ http://repo.spring.io/libs-release-local
+
+ false
+
+
+
+
+
+