From 61240fd0fa3ae03e152df7a48101e53afd7b8015 Mon Sep 17 00:00:00 2001 From: Jacob Severson Date: Wed, 1 Nov 2017 21:12:40 -0500 Subject: [PATCH] Adding hook for changing message target delimeter Resolves #1120 --- .../stream/binder/AbstractBinderTests.java | 129 ++++++++++-------- .../binder/PartitionCapableBinderTests.java | 42 +++--- 2 files changed, 98 insertions(+), 73 deletions(-) diff --git a/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java b/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java index 150ba703c..0d52fba60 100644 --- a/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java +++ b/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java @@ -72,6 +72,7 @@ import static org.junit.Assert.assertTrue; * @author Mark Fisher * @author Marius Bogoevici * @author Oleg Zhurakousky + * @author Jacob Severson */ @SuppressWarnings("unchecked") public abstract class AbstractBinderTests, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties> { @@ -114,16 +115,21 @@ public abstract class AbstractBinderTests foo0ProducerBinding = binder.bindProducer("foo.0", - this.createBindableChannel("output", new BindingProperties()), createProducerProperties()); - Binding foo0ConsumerBinding = binder.bindConsumer("foo.0", - "testClean", this.createBindableChannel("input", new BindingProperties()), createConsumerProperties()); - Binding foo1ProducerBinding = binder.bindProducer("foo.1", - this.createBindableChannel("output", new BindingProperties()), createProducerProperties()); - Binding foo1ConsumerBinding = binder.bindConsumer("foo.1", - "testClean", this.createBindableChannel("input", new BindingProperties()), createConsumerProperties()); - Binding foo2ProducerBinding = binder.bindProducer("foo.2", - this.createBindableChannel("output", new BindingProperties()), createProducerProperties()); + Binding foo0ProducerBinding = binder.bindProducer(String.format("foo%s0", + getDestinationNameDelimiter()), this.createBindableChannel("output", new BindingProperties()), + createProducerProperties()); + Binding foo0ConsumerBinding = binder.bindConsumer(String.format("foo%s0", + getDestinationNameDelimiter()), "testClean", this.createBindableChannel("input", + new BindingProperties()), createConsumerProperties()); + Binding foo1ProducerBinding = binder.bindProducer(String.format("foo%s1", + getDestinationNameDelimiter()), this.createBindableChannel("output", new BindingProperties()), + createProducerProperties()); + Binding foo1ConsumerBinding = binder.bindConsumer(String.format("foo%s1", + getDestinationNameDelimiter()), "testClean", this.createBindableChannel("input", + new BindingProperties()), createConsumerProperties()); + Binding foo2ProducerBinding = binder.bindProducer(String.format("foo%s2", + getDestinationNameDelimiter()), this.createBindableChannel("output", new BindingProperties()), + createProducerProperties()); foo0ProducerBinding.unbind(); assertThat(TestUtils .getPropertyValue(foo0ProducerBinding, "lifecycle", Lifecycle.class) @@ -156,10 +162,10 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer("foo.0", moduleOutputChannel, - outputBindingProperties.getProducer()); - Binding consumerBinding = binder.bindConsumer("foo.0", "testSendAndReceive", moduleInputChannel, - createConsumerProperties()); + Binding producerBinding = binder.bindProducer(String.format("foo%s0", + getDestinationNameDelimiter()), moduleOutputChannel, outputBindingProperties.getProducer()); + Binding consumerBinding = binder.bindConsumer(String.format("foo%s0", + getDestinationNameDelimiter()), "testSendAndReceive", moduleInputChannel, createConsumerProperties()); Message message = MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar") .build(); // Let the consumer actually bind to the producer before sending a msg @@ -213,10 +219,10 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer("foo.0x", moduleOutputChannel, - outputBindingProperties.getProducer()); - Binding consumerBinding = binder.bindConsumer("foo.0x", "testSendAndReceiveKryo", moduleInputChannel, - createConsumerProperties()); + Binding producerBinding = binder.bindProducer(String.format("foo%s0x", + getDestinationNameDelimiter()), moduleOutputChannel, outputBindingProperties.getProducer()); + Binding consumerBinding = binder.bindConsumer(String.format("foo%s0x", + getDestinationNameDelimiter()), "testSendAndReceiveKryo", moduleInputChannel, createConsumerProperties()); Foo foo = new Foo(); foo.setName("Bill"); Message message = MessageBuilder.withPayload(foo).setHeader(MessageHeaders.CONTENT_TYPE, MessageConverterUtils.X_JAVA_OBJECT) @@ -259,9 +265,10 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer("foo.0y", moduleOutputChannel, - outputBindingProperties.getProducer()); - Binding consumerBinding = binder.bindConsumer("foo.0y", "testSendAndReceiveJavaSerialization", moduleInputChannel, + Binding producerBinding = binder.bindProducer(String.format("foo%s0y", + getDestinationNameDelimiter()), moduleOutputChannel, outputBindingProperties.getProducer()); + Binding consumerBinding = binder.bindConsumer(String.format("foo%s0y", + getDestinationNameDelimiter()), "testSendAndReceiveJavaSerialization", moduleInputChannel, createConsumerProperties()); SerializableFoo foo = new SerializableFoo(); Message message = MessageBuilder.withPayload(foo).setHeader(MessageHeaders.CONTENT_TYPE, MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT) @@ -306,16 +313,16 @@ public abstract class AbstractBinderTests producerBinding1 = binder.bindProducer("foo.xy", - moduleOutputChannel1, createProducerProperties()); - Binding producerBinding2 = binder.bindProducer("foo.yz", - moduleOutputChannel2, createProducerProperties()); + Binding producerBinding1 = binder.bindProducer(String.format("foo%sxy", + getDestinationNameDelimiter()), moduleOutputChannel1, createProducerProperties()); + Binding producerBinding2 = binder.bindProducer(String.format("foo%syz", + getDestinationNameDelimiter()), moduleOutputChannel2, createProducerProperties()); - Binding consumerBinding1 = binder.bindConsumer("foo.xy", - "testSendAndReceiveMultipleTopics", moduleInputChannel, + Binding consumerBinding1 = binder.bindConsumer(String.format("foo%sxy", + getDestinationNameDelimiter()), "testSendAndReceiveMultipleTopics", moduleInputChannel, createConsumerProperties()); - Binding consumerBinding2 = binder.bindConsumer("foo.yz", - "testSendAndReceiveMultipleTopics", moduleInputChannel, + Binding consumerBinding2 = binder.bindConsumer(String.format("foo%syz", + getDestinationNameDelimiter()), "testSendAndReceiveMultipleTopics", moduleInputChannel, createConsumerProperties()); String testPayload1 = "foo" + UUID.randomUUID().toString(); @@ -361,10 +368,10 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer("bar.0", - moduleOutputChannel, producerBindingProperties.getProducer()); - Binding consumerBinding = binder.bindConsumer("bar.0", - "testSendAndReceiveNoOriginalContentType", moduleInputChannel, + Binding producerBinding = binder.bindProducer(String.format("bar%s0", + getDestinationNameDelimiter()), moduleOutputChannel, producerBindingProperties.getProducer()); + Binding consumerBinding = binder.bindConsumer(String.format("bar%s0", + getDestinationNameDelimiter()), "testSendAndReceiveNoOriginalContentType", moduleInputChannel, createConsumerProperties()); binderBindUnbindLatency(); @@ -468,6 +475,14 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer("bad.0a", - moduleOutputChannel, createProducerProperties()); - Binding consumerBinding = binder.bindConsumer("bad.0a", "test-1", - moduleInputChannel, createConsumerProperties()); + Binding producerBinding = binder.bindProducer(String.format("bad%s0a", + getDestinationNameDelimiter()), moduleOutputChannel, createProducerProperties()); + Binding consumerBinding = binder.bindConsumer(String.format("bad%s0a", + getDestinationNameDelimiter()), "test-1", moduleInputChannel, createConsumerProperties()); Station station = new Station(); Message message = MessageBuilder.withPayload(station).build(); @@ -510,10 +525,10 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer("bad.0b", - moduleOutputChannel, createProducerProperties()); - Binding consumerBinding = binder.bindConsumer("bad.0b", "test-2", - moduleInputChannel, createConsumerProperties()); + Binding producerBinding = binder.bindProducer(String.format("bad%s0b", + getDestinationNameDelimiter()), moduleOutputChannel, createProducerProperties()); + Binding consumerBinding = binder.bindConsumer(String.format("bad%s0b", + getDestinationNameDelimiter()), "test-2", moduleInputChannel, createConsumerProperties()); Station station = new Station(); Message message = MessageBuilder.withPayload(station).setHeader( @@ -537,10 +552,10 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer("bad.0c", - moduleOutputChannel, createProducerProperties()); - Binding consumerBinding = binder.bindConsumer("bad.0c", "test-3", - moduleInputChannel, createConsumerProperties()); + Binding producerBinding = binder.bindProducer(String.format("bad%s0c", + getDestinationNameDelimiter()), moduleOutputChannel, createProducerProperties()); + Binding consumerBinding = binder.bindConsumer(String.format("bad%s0c", + getDestinationNameDelimiter()), "test-3", moduleInputChannel, createConsumerProperties()); try { Station station = new Station(); Message message = MessageBuilder.withPayload(station) @@ -565,10 +580,10 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer("bad.0d", - moduleOutputChannel, createProducerProperties()); - Binding consumerBinding = binder.bindConsumer("bad.0d", "test-4", - moduleInputChannel, createConsumerProperties()); + Binding producerBinding = binder.bindProducer(String.format("bad%s0d", + getDestinationNameDelimiter()), moduleOutputChannel, createProducerProperties()); + Binding consumerBinding = binder.bindConsumer(String.format("bad%s0d", + getDestinationNameDelimiter()), "test-4", moduleInputChannel, createConsumerProperties()); String value = "{\"readings\":[{\"stationid\":\"fgh\"," + "\"customerid\":\"12345\",\"timestamp\":null},{\"stationid\":\"hjk\",\"customerid\":\"222\",\"timestamp\":null}]}"; @@ -599,10 +614,10 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer("bad.0e", - moduleOutputChannel, createProducerProperties()); - Binding consumerBinding = binder.bindConsumer("bad.0e", "test-5", - moduleInputChannel, createConsumerProperties()); + Binding producerBinding = binder.bindProducer(String.format("bad%s0e", + getDestinationNameDelimiter()), moduleOutputChannel, createProducerProperties()); + Binding consumerBinding = binder.bindConsumer(String.format("bad%s0e", + getDestinationNameDelimiter()), "test-5", moduleInputChannel, createConsumerProperties()); String value = "{\"readings\":[{\"stationid\":\"fgh\"," + "\"customerid\":\"12345\",\"timestamp\":null},{\"stationid\":\"hjk\",\"customerid\":\"222\",\"timestamp\":null}]}"; @@ -633,10 +648,10 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer("bad.0f", - moduleOutputChannel, createProducerProperties()); - Binding consumerBinding = binder.bindConsumer("bad.0f", "test-6", - moduleInputChannel, createConsumerProperties()); + Binding producerBinding = binder.bindProducer(String.format("bad%s0f", + getDestinationNameDelimiter()), moduleOutputChannel, createProducerProperties()); + Binding consumerBinding = binder.bindConsumer(String.format("bad%s0f", + getDestinationNameDelimiter()), "test-6", moduleInputChannel, createConsumerProperties()); Readings r1 = new Readings(); r1.setCustomerid("123"); diff --git a/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java b/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java index 136b3a848..9cc90767f 100644 --- a/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java +++ b/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java @@ -57,16 +57,16 @@ public abstract class PartitionCapableBinderTests producerBinding = binder.bindProducer("defaultGroup.0", output, - (PP) producerBindingProperties.getProducer()); + Binding producerBinding = binder.bindProducer(String.format("defaultGroup%s0", + getDestinationNameDelimiter()), output, (PP) producerBindingProperties.getProducer()); QueueChannel input1 = new QueueChannel(); - Binding binding1 = binder.bindConsumer("defaultGroup.0", null, input1, - createConsumerProperties()); + Binding binding1 = binder.bindConsumer(String.format("defaultGroup%s0", + getDestinationNameDelimiter()), null, input1, createConsumerProperties()); QueueChannel input2 = new QueueChannel(); - Binding binding2 = binder.bindConsumer("defaultGroup.0", null, input2, - createConsumerProperties()); + Binding binding2 = binder.bindConsumer(String.format("defaultGroup%s0", + getDestinationNameDelimiter()), null, input2, createConsumerProperties()); String testPayload1 = "foo-" + UUID.randomUUID().toString(); output.send(MessageBuilder.withPayload(testPayload1).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build()); @@ -84,7 +84,8 @@ public abstract class PartitionCapableBinderTests input0Binding = binder.bindConsumer("part.0", "testPartitionedModuleSpEL", input0, consumerProperties); + Binding input0Binding = binder.bindConsumer(String.format("part%s0", + getDestinationNameDelimiter()), "testPartitionedModuleSpEL", input0, consumerProperties); consumerProperties.setInstanceIndex(1); QueueChannel input1 = new QueueChannel(); input1.setBeanName("test.input1S"); - Binding input1Binding = binder.bindConsumer("part.0", "testPartitionedModuleSpEL", input1, consumerProperties); + Binding input1Binding = binder.bindConsumer(String.format("part%s0", + getDestinationNameDelimiter()), "testPartitionedModuleSpEL", input1, consumerProperties); consumerProperties.setInstanceIndex(2); QueueChannel input2 = new QueueChannel(); input2.setBeanName("test.input2S"); - Binding input2Binding = binder.bindConsumer("part.0", "testPartitionedModuleSpEL", input2, consumerProperties); + Binding input2Binding = binder.bindConsumer(String.format("part%s0", + getDestinationNameDelimiter()), "testPartitionedModuleSpEL", input2, consumerProperties); PP producerProperties = createProducerProperties(); producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload")); @@ -192,7 +196,8 @@ public abstract class PartitionCapableBinderTests outputBinding = binder.bindProducer("part.0", output, producerProperties); + Binding outputBinding = binder.bindProducer(String.format("part%s0", + getDestinationNameDelimiter()), output, producerProperties); try { Object endpoint = extractEndpoint(outputBinding); checkRkExpressionForPartitionedModuleSpEL(endpoint); @@ -252,7 +257,8 @@ public abstract class PartitionCapableBinderTests input0Binding = binder.bindConsumer("partJ.0", "testPartitionedModuleJava", input0, consumerProperties); + Binding input0Binding = binder.bindConsumer(String.format("partJ%s0", + getDestinationNameDelimiter()), "testPartitionedModuleJava", input0, consumerProperties); consumerProperties.setInstanceIndex(1); QueueChannel input1 = new QueueChannel(); input1.setBeanName("test.input1J"); - Binding input1Binding = binder.bindConsumer("partJ.0", "testPartitionedModuleJava", input1, consumerProperties); + Binding input1Binding = binder.bindConsumer(String.format("partJ%s0", + getDestinationNameDelimiter()), "testPartitionedModuleJava", input1, consumerProperties); consumerProperties.setInstanceIndex(2); QueueChannel input2 = new QueueChannel(); input2.setBeanName("test.input2J"); - Binding input2Binding = binder.bindConsumer("partJ.0", "testPartitionedModuleJava", input2, consumerProperties); + Binding input2Binding = binder.bindConsumer(String.format("partJ%s0", + getDestinationNameDelimiter()), "testPartitionedModuleJava", input2, consumerProperties); PP producerProperties = createProducerProperties(); producerProperties.setPartitionKeyExtractorClass(PartitionTestSupport.class); @@ -285,7 +294,8 @@ public abstract class PartitionCapableBinderTests outputBinding = binder.bindProducer("partJ.0", output, producerProperties); if (usesExplicitRouting()) { Object endpoint = extractEndpoint(outputBinding); - assertThat(getEndpointRouting(endpoint)).contains(getExpectedRoutingBaseDestination("partJ.0", "testPartitionedModuleJava") + assertThat(getEndpointRouting(endpoint)).contains(getExpectedRoutingBaseDestination( + String.format("partJ%s0", getDestinationNameDelimiter()), "testPartitionedModuleJava") + "-' + headers['" + BinderHeaders.PARTITION_HEADER + "']"); }