spring/spring-framework
1
0
Fork 0
mirror of https://github.com/spring-projects/spring-framework.git synced 2026-09-17 08:24:13 +00:00
Code Issues Projects Releases Wiki Activity

Ensure Payload release on early error in createHeaders

Browse Source 
Closes gh-37026
This commit is contained in:
rstoyanchev
2026-08-14 09:19:58 +02:00
committed by Brian Clozel
parent a69fe71630
commit d5dbc9b310
1 changed files with 31 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/MessagingRSocket.java
+31 -22
View File
@@ -103,8 +103,9 @@ class MessagingRSocket implements RSocket {
* @return completion handle for success or error
*/
public Mono<Void> handleConnectionSetupPayload(ConnectionSetupPayload payload) {
// frameDecoder does not apply to connectionSetupPayload
// so retain here since handle expects it.
// RSocket Java releases its own ConnectionSetupPayload reference
// For other frames, PayloadDecoder increases the refCount, and handles must releases it;
// retain here to match the release in handle (in retainDataAndReleasePayload)
payload.retain();
return handle(payload, FrameType.SETUP);
}
@@ -187,32 +188,40 @@ class MessagingRSocket implements RSocket {
private MessageHeaders createHeaders(
Payload payload, FrameType frameType, @Nullable AtomicReference<Flux<Payload>> responseRef) {
MessageHeaderAccessor headers = new MessageHeaderAccessor();
headers.setLeaveMutable(true);
try {
MessageHeaderAccessor headers = new MessageHeaderAccessor();
headers.setLeaveMutable(true);
Map<String, Object> metadataValues = this.metadataExtractor.extract(payload, this.metadataMimeType);
Map<String, Object> metadataValues = this.metadataExtractor.extract(payload, this.metadataMimeType);
metadataValues.putIfAbsent(MetadataExtractor.ROUTE_KEY, "");
for (Map.Entry<String, Object> entry : metadataValues.entrySet()) {
if (entry.getKey().equals(MetadataExtractor.ROUTE_KEY)) {
RouteMatcher.Route route = this.routeMatcher.parseRoute((String) entry.getValue());
headers.setHeader(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER, route);
metadataValues.putIfAbsent(MetadataExtractor.ROUTE_KEY, "");
for (Map.Entry<String, Object> entry : metadataValues.entrySet()) {
if (entry.getKey().equals(MetadataExtractor.ROUTE_KEY)) {
RouteMatcher.Route route = this.routeMatcher.parseRoute((String) entry.getValue());
headers.setHeader(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER, route);
}
else {
headers.setHeader(entry.getKey(), entry.getValue());
}
}
else {
headers.setHeader(entry.getKey(), entry.getValue());
headers.setContentType(this.dataMimeType);
headers.setHeader(RSocketFrameTypeMessageCondition.FRAME_TYPE_HEADER, frameType);
headers.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, this.requester);
if (responseRef != null) {
headers.setHeader(RSocketPayloadReturnValueHandler.RESPONSE_HEADER, responseRef);
}
}
headers.setHeader(HandlerMethodReturnValueHandler.DATA_BUFFER_FACTORY_HEADER,
this.strategies.dataBufferFactory());
headers.setContentType(this.dataMimeType);
headers.setHeader(RSocketFrameTypeMessageCondition.FRAME_TYPE_HEADER, frameType);
headers.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, this.requester);
if (responseRef != null) {
headers.setHeader(RSocketPayloadReturnValueHandler.RESPONSE_HEADER, responseRef);
return headers.getMessageHeaders();
}
catch (Throwable ex) {
if (payload.refCnt() > 0) {
payload.release();
}
throw ex;
}
headers.setHeader(HandlerMethodReturnValueHandler.DATA_BUFFER_FACTORY_HEADER,
this.strategies.dataBufferFactory());
return headers.getMessageHeaders();
}
}