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

GH-2196 Enhance test case to demonstrate StreamBridge with multiple partitioned producers

Browse Source 
Resolves #2196
This commit is contained in:
Oleg Zhurakousky
2021-07-16 15:25:26 +02:00
parent 70feca33ec
commit aa3b0675c7
1 changed files with 36 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java
View File

@@ -228,19 +228,52 @@ public class StreamBridgeTests {
.getCompleteConfiguration(EmptyConfiguration.class))
.web(WebApplicationType.NONE).run("--spring.cloud.stream.source=foo;bar",
"--spring.cloud.stream.bindings.foo-out-0.producer.partitionKeyExpression=payload",
"--spring.cloud.stream.bindings.foo-out-0.producer.partitionCount=2",
"--spring.cloud.stream.bindings.foo-out-0.producer.partitionCount=5",
"--spring.cloud.stream.bindings.bar-out-0.producer.partitionKeyExpression=payload",
"--spring.cloud.stream.bindings.bar-out-0.producer.partitionCount=1",
"--spring.jmx.enabled=false")) {
StreamBridge bridge = context.getBean(StreamBridge.class);
bridge.send("foo-out-0", "a");
bridge.send("bar-out-0", "b");
bridge.send("foo-out-0", "c");
bridge.send("foo-out-0", "d");
bridge.send("bar-out-0", "e");
bridge.send("foo-out-0", "f");
bridge.send("bar-out-0", "g");
OutputDestination outputDestination = context.getBean(OutputDestination.class);
Message<byte[]> message = outputDestination.receive(100, "foo-out-0");
assertThat(new String(message.getPayload())).isEqualTo("a");
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(1);
assertThat(new String(outputDestination.receive(100, "bar-out-0").getPayload())).isEqualTo("b");
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(2);
message = outputDestination.receive(100, "foo-out-0");
assertThat(new String(message.getPayload())).isEqualTo("c");
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(4);
message = outputDestination.receive(100, "foo-out-0");
assertThat(new String(message.getPayload())).isEqualTo("d");
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(0);
message = outputDestination.receive(100, "bar-out-0");
assertThat(new String(message.getPayload())).isEqualTo("b");
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(0);
message = outputDestination.receive(100, "bar-out-0");
assertThat(new String(message.getPayload())).isEqualTo("e");
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(0);
message = outputDestination.receive(100, "bar-out-0");
assertThat(new String(message.getPayload())).isEqualTo("g");
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(0);
message = outputDestination.receive(100, "foo-out-0");
assertThat(new String(message.getPayload())).isEqualTo("f");
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(2);
//assertThat(new String(outputDestination.receive(100, "bar-out-0").getPayload())).isEqualTo("b");
}
}