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

GH-1894 Added support for named destination in test binder

Browse Source 
Deprecated indexed operations
Modified several tests to use new operators

Resolves #1894
This commit is contained in:
Oleg Zhurakousky
2020-01-24 19:16:21 +01:00
parent e1a3b72027
commit a94e533a50
5 changed files with 47 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
docs/src/main/asciidoc/spring-cloud-stream.adoc
View File

@@ -873,13 +873,13 @@ public void testMultipleFunctions() {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
Message<byte[]> inputMessage = MessageBuilder.withPayload("Hello".getBytes()).build();
inputDestination.send(inputMessage, 0);
inputDestination.send(inputMessage, 1);
inputDestination.send(inputMessage, "uppercase-in-0");
inputDestination.send(inputMessage, "reverse-in-0");
Message<byte[]> outputMessage = outputDestination.receive(0, 0);
Message<byte[]> outputMessage = outputDestination.receive(0, "uppercase-out-0");
assertThat(outputMessage.getPayload()).isEqualTo("HELLO".getBytes());
outputMessage = outputDestination.receive(0, 1);
outputMessage = outputDestination.receive(0, "uppercase-out-1");
assertThat(outputMessage.getPayload()).isEqualTo("olleH".getBytes());
}
}

2
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/AbstractDestination.java
View File

@@ -44,7 +44,7 @@ abstract class AbstractDestination {
}
SubscribableChannel getChannelByName(String name) {
//name = name + ".destination";
name = name.endsWith(".destination") ? name : name + ".destination";
for (AbstractSubscribableChannel subscribableChannel : channels) {
if (subscribableChannel.getBeanName().equals(name)) {
return subscribableChannel;

16
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java
View File

@@ -29,18 +29,30 @@ import org.springframework.messaging.Message;
public class InputDestination extends AbstractDestination {
/**
* Allows the {@link Message} to be sent to a Binder to be delegated to binder's input
* destination (e.g., Processor.INPUT).
* Allows the {@link Message} to be sent to a Binder to be delegated to a default binding
* destination (e.g., "function-in-0" for cases where you only have a single function with the name 'function').
* @param message message to send
*/
public void send(Message<?> message) {
this.getChannel(0).send(message);
}
/**
* @param message message to send
* @param inputIndex input index
* @deprecated since 3.0.2 in favor of {@link #receive(long, String)} where you should use the actual binding name (e.g., "foo-in-0")
*/
@Deprecated
public void send(Message<?> message, int inputIndex) {
this.getChannel(inputIndex).send(message);
}
/**
* Allows the {@link Message} to be sent to a Binder to be delegated to a named binding
* destination (e.g., "function-in-0" for cases where you want to send input to function with the name 'function').
* @param message message to send
* @param bindingName binding name
*/
public void send(Message<?> message, String bindingName) {
this.getChannelByName(bindingName).send(message);
}

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/OutputDestination.java
View File

@@ -18,7 +18,6 @@ package org.springframework.cloud.stream.binder.test;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedTransferQueue;
@@ -40,6 +39,7 @@ public class OutputDestination extends AbstractDestination {
public Message<byte[]> receive(long timeout, String bindingName) {
try {
bindingName = bindingName.endsWith(".destination") ? bindingName : bindingName + ".destination";
return this.messageQueues.get(bindingName).poll(timeout, TimeUnit.MILLISECONDS);
}
catch (InterruptedException e) {
@@ -51,7 +51,9 @@ public class OutputDestination extends AbstractDestination {
* Allows to access {@link Message}s received by this {@link OutputDestination}.
* @param timeout how long to wait before giving up
* @return received message
* @deprecated since 3.0.2 in favor of {@link #receive(long, String)} where you should use the actual binding name (e.g., "foo-in-0")
*/
@Deprecated
public Message<byte[]> receive(long timeout, int bindingIndex) {
try {
BlockingQueue<Message<byte[]>> destinationQueue = (new ArrayList<>(this.messageQueues.values())).get(bindingIndex);

48
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/MultipleInputOutputFunctionTests.java
View File

@@ -141,12 +141,12 @@ public class MultipleInputOutputFunctionTests {
Message<byte[]> stringInputMessage = MessageBuilder.withPayload("one".getBytes()).build();
Message<byte[]> integerInputMessage = MessageBuilder.withPayload("1".getBytes()).build();
inputDestination.send(stringInputMessage, 0);
inputDestination.send(integerInputMessage, 1);
inputDestination.send(stringInputMessage, "multiInputSingleOutput-in-0");
inputDestination.send(integerInputMessage, "multiInputSingleOutput-in-1");
Message<byte[]> outputMessage = outputDestination.receive();
assertThat(outputMessage.getPayload()).isEqualTo("one".getBytes());
outputMessage = outputDestination.receive();
outputMessage = outputDestination.receive(0, "multiInputSingleOutput-out-0");
assertThat(outputMessage.getPayload()).isEqualTo("1".getBytes());
}
}
@@ -165,14 +165,14 @@ public class MultipleInputOutputFunctionTests {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
for (int i = 0; i < 10; i++) {
inputDestination.send(MessageBuilder.withPayload(String.valueOf(i).getBytes()).build());
inputDestination.send(MessageBuilder.withPayload(String.valueOf(i).getBytes()).build(), "singleInputMultipleOutputs-in-0");
}
int counter = 0;
for (int i = 0; i < 5; i++) {
Message<byte[]> even = outputDestination.receive(0, 0);
Message<byte[]> even = outputDestination.receive(0, "singleInputMultipleOutputs-out-0");
assertThat(even.getPayload()).isEqualTo(("EVEN: " + String.valueOf(counter++)).getBytes());
Message<byte[]> odd = outputDestination.receive(0, 1);
Message<byte[]> odd = outputDestination.receive(0, "singleInputMultipleOutputs-out-1");
assertThat(odd.getPayload()).isEqualTo(("ODD: " + String.valueOf(counter++)).getBytes());
}
}
@@ -192,13 +192,13 @@ public class MultipleInputOutputFunctionTests {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
Message<byte[]> inputMessage = MessageBuilder.withPayload("Hello".getBytes()).build();
inputDestination.send(inputMessage, 0);
inputDestination.send(inputMessage, 1);
inputDestination.send(inputMessage, "uppercase-in-0");
inputDestination.send(inputMessage, "reverse-in-0");
Message<byte[]> outputMessage = outputDestination.receive(0, 0);
Message<byte[]> outputMessage = outputDestination.receive(0, "uppercase-out-0");
assertThat(outputMessage.getPayload()).isEqualTo("HELLO".getBytes());
outputMessage = outputDestination.receive(0, 1);
outputMessage = outputDestination.receive(0, "reverse-out-0");
assertThat(outputMessage.getPayload()).isEqualTo("olleH".getBytes());
}
}
@@ -217,13 +217,13 @@ public class MultipleInputOutputFunctionTests {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
Message<byte[]> inputMessage = MessageBuilder.withPayload("Hello".getBytes()).build();
inputDestination.send(inputMessage, 0);
inputDestination.send(inputMessage, 1);
inputDestination.send(inputMessage, "uppercasereverse-in-0");
inputDestination.send(inputMessage, "reverseuppercase-in-0");
Message<byte[]> outputMessage = outputDestination.receive(0, 0);
Message<byte[]> outputMessage = outputDestination.receive(0, "uppercasereverse-out-0");
assertThat(outputMessage.getPayload()).isEqualTo("OLLEH".getBytes());
outputMessage = outputDestination.receive(0, 1);
outputMessage = outputDestination.receive(0, "reverseuppercase-out-0");
assertThat(outputMessage.getPayload()).isEqualTo("OLLEH".getBytes());
}
}
@@ -245,12 +245,12 @@ public class MultipleInputOutputFunctionTests {
Message<byte[]> stringInputMessage = MessageBuilder.withPayload("ricky".getBytes()).build();
Message<byte[]> integerInputMessage = MessageBuilder.withPayload("bobby".getBytes()).build();
inputDestination.send(stringInputMessage, 0);
inputDestination.send(integerInputMessage, 1);
inputDestination.send(stringInputMessage, "multiInputSingleOutput-in-0");
inputDestination.send(integerInputMessage, "multiInputSingleOutput-in-1");
Message<byte[]> outputMessage = outputDestination.receive();
Message<byte[]> outputMessage = outputDestination.receive(1000, "multiInputSingleOutput-out-0");
assertThat(outputMessage.getPayload()).isEqualTo("RICKY".getBytes());
outputMessage = outputDestination.receive();
outputMessage = outputDestination.receive(1000, "multiInputSingleOutput-out-0");
assertThat(outputMessage.getPayload()).isEqualTo("BOBBY".getBytes());
}
}
@@ -273,13 +273,11 @@ public class MultipleInputOutputFunctionTests {
Message<byte[]> stringInputMessage = MessageBuilder.withPayload("ricky".getBytes()).build();
Message<byte[]> integerInputMessage = MessageBuilder.withPayload("bobby".getBytes()).build();
inputDestination.send(stringInputMessage, 0);
inputDestination.send(integerInputMessage, 1);
inputDestination.send(stringInputMessage, "multiInputSingleOutput2-in-0");
inputDestination.send(integerInputMessage, "multiInputSingleOutput2-in-1");
Message<byte[]> outputMessage = outputDestination.receive(2500);
Message<byte[]> outputMessage = outputDestination.receive(1000, "multiInputSingleOutput2-out-0");
assertThat(outputMessage.getPayload()).isEqualTo("rickybobby".getBytes());
// outputMessage = outputDestination.receive();
// assertThat(outputMessage.getPayload()).isEqualTo("BOBBY".getBytes());
}
}
@@ -384,6 +382,7 @@ public class MultipleInputOutputFunctionTests {
return Person.class.isAssignableFrom(clazz);
}
@Override
@Nullable
protected Object convertFromInternal(
Message<?> message, Class<?> targetClass, @Nullable Object conversionHint) {
@@ -392,6 +391,7 @@ public class MultipleInputOutputFunctionTests {
return person;
}
@Override
@Nullable
protected Object convertToInternal(
Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) {
@@ -410,6 +410,7 @@ public class MultipleInputOutputFunctionTests {
return Employee.class.isAssignableFrom(clazz);
}
@Override
@Nullable
protected Object convertFromInternal(
Message<?> message, Class<?> targetClass, @Nullable Object conversionHint) {
@@ -418,6 +419,7 @@ public class MultipleInputOutputFunctionTests {
return person;
}
@Override
@Nullable
protected Object convertToInternal(
Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) {