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

Fix concurrency issue in StreamBridge

Browse Source 
Resolves #2300
This commit is contained in:
Oleg Zhurakousky
2022-03-25 14:26:25 +01:00
parent 9f28bcde3d
commit f58dcf371d
2 changed files with 16 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
View File

@@ -208,20 +208,26 @@ public final class StreamBridge implements SmartInitializingSingleton {
@SuppressWarnings({ "unchecked"})
public boolean send(String bindingName, @Nullable String binderName, Object data, MimeType outputContentType) {
if (data instanceof Message) {
data = MessageBuilder.fromMessage((Message) data).setHeader(MessageUtils.TARGET_PROTOCOL, "streamBridge").build();
}
else {
data = new GenericMessage<>(data, Collections.singletonMap(MessageUtils.TARGET_PROTOCOL, "streamBridge"));
}
ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName);
MessageChannel messageChannel = this.resolveDestination(bindingName, producerProperties, binderName);
Function functionToInvoke = this.getStreamBridgeFunction(outputContentType.toString(), producerProperties);
if (producerProperties != null && producerProperties.isPartitioned()) {
functionToInvoke = new PartitionAwareFunctionWrapper(functionToInvoke, this.applicationContext, producerProperties);
}
Message<byte[]> resultMessage = (Message<byte[]>) functionToInvoke.apply(data);
Message<?> messageToSend = data instanceof Message
? MessageBuilder.fromMessage((Message) data).setHeader(MessageUtils.TARGET_PROTOCOL, "streamBridge").build()
: new GenericMessage<>(data, Collections.singletonMap(MessageUtils.TARGET_PROTOCOL, "streamBridge"));
Message<?> resultMessage;
synchronized (this) {
resultMessage = (Message<byte[]>) functionToInvoke.apply(messageToSend);
}
return messageChannel.send(resultMessage);
}

5
core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java
View File

@@ -267,10 +267,11 @@ public class ContentTypeTckTests {
.setHeader("contentType", new MimeType("text", "plain")).build());
Message<byte[]> outputMessage = target.receive();
System.out.println(new String(outputMessage.getPayload()));
assertThat(outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE))
.isEqualTo(MimeTypeUtils.TEXT_PLAIN_VALUE);
assertThat(new String(outputMessage.getPayload(), StandardCharsets.UTF_8))
.isEqualTo(jsonPayload);
assertThat(outputMessage.getPayload())
.isEqualTo(jsonPayload.getBytes());
}
@Test