SolaceIO: map message user properties #40108
+164
−1
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -21,12 +21,17 @@ | |
| import com.solacesystems.jcsmp.BytesMessage; | ||
| import com.solacesystems.jcsmp.BytesXMLMessage; | ||
| import com.solacesystems.jcsmp.JCSMPFactory; | ||
| import com.solacesystems.jcsmp.SDTException; | ||
| import com.solacesystems.jcsmp.SDTMap; | ||
| import com.solacesystems.jcsmp.TextMessage; | ||
| import java.nio.ByteBuffer; | ||
| import java.nio.charset.CharacterCodingException; | ||
| import java.nio.charset.CodingErrorAction; | ||
| import java.nio.charset.StandardCharsets; | ||
| import java.util.Arrays; | ||
| import java.util.Collections; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import org.apache.beam.sdk.schemas.AutoValueSchema; | ||
| import org.apache.beam.sdk.schemas.annotations.DefaultSchema; | ||
| import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber; | ||
|
|
@@ -276,6 +281,16 @@ public enum PayloadType { | |
| @SchemaFieldNumber("13") | ||
| public abstract PayloadType getPayloadType(); | ||
|
|
||
| /** | ||
| * Gets the user properties of the message as a string map. | ||
| * | ||
| * <p>Mapped from {@link BytesXMLMessage#getProperties()}. Values are stringified. | ||
| * | ||
| * @return The user properties, or an empty map if the message carries none. | ||
| */ | ||
| @SchemaFieldNumber("14") | ||
| public abstract Map<String, String> getUserProperties(); | ||
|
|
||
| /** Gets the payload decoded as UTF-8 when this record has type {@link PayloadType#TEXT}. */ | ||
| public final String getText() { | ||
| if (getPayloadType() != PayloadType.TEXT) { | ||
|
|
@@ -292,7 +307,8 @@ public static Builder builder() { | |
| .setRedelivered(false) | ||
| .setTimeToLive(0) | ||
| .setAttachmentBytes(new byte[0]) | ||
| .setPayloadType(PayloadType.BYTES_XML) | ||
| .setUserProperties(Collections.emptyMap()); | ||
| } | ||
|
|
||
| @AutoValue.Builder | ||
|
|
@@ -332,6 +348,8 @@ public abstract Builder setReplicationGroupMessageId( | |
|
|
||
| public abstract Builder setAttachmentBytes(byte[] attachmentBytes); | ||
|
|
||
| public abstract Builder setUserProperties(Map<String, String> userProperties); | ||
|
|
||
| public abstract Record build(); | ||
| } | ||
|
|
||
|
|
@@ -456,6 +474,7 @@ public static class SolaceRecordMapper { | |
|
|
||
| Destination replyTo = getDestination(msg.getCorrelationId(), msg.getReplyTo()); | ||
| Destination destination = getDestination(msg.getCorrelationId(), msg.getDestination()); | ||
| Map<String, String> userProperties = getUserProperties(msg.getProperties()); | ||
|
|
||
| Record.Builder recordBuilder = decodePayload(msg); | ||
| return recordBuilder | ||
|
|
@@ -473,6 +492,7 @@ public static class SolaceRecordMapper { | |
| msg.getReplicationGroupMessageId() != null | ||
| ? msg.getReplicationGroupMessageId().toString() | ||
| : null) | ||
| .setUserProperties(userProperties) | ||
| .build(); | ||
| } | ||
|
|
||
|
|
@@ -519,6 +539,10 @@ public static BytesXMLMessage toMessage(Record record) { | |
| msg.setSenderTimestamp(senderTimestamp); | ||
| msg.setApplicationMessageId(record.getMessageId()); | ||
|
|
||
| if (!record.getUserProperties().isEmpty()) { | ||
| msg.setProperties(createUserProperties(record.getUserProperties())); | ||
| } | ||
|
|
||
| return msg; | ||
| } | ||
|
|
||
|
|
@@ -599,5 +623,47 @@ private static byte[] readAttachment(BytesXMLMessage msg) { | |
| buffer.get(attachment); | ||
| return attachment; | ||
| } | ||
|
|
||
| private static Map<String, String> getUserProperties(@Nullable SDTMap properties) { | ||
| if (properties == null || properties.isEmpty()) { | ||
| return Collections.emptyMap(); | ||
| } | ||
|
|
||
| Map<String, String> userProperties = new HashMap<>(); | ||
| for (String key : properties.keySet()) { | ||
| String value = stringifyUserProperty(properties, key); | ||
| if (value == null) { | ||
| LOG.warn("User property '{}' has a null value, skipping.", key); | ||
| continue; | ||
| } | ||
| userProperties.put(key, value); | ||
| } | ||
| return Collections.unmodifiableMap(userProperties); | ||
| } | ||
|
|
||
| private static @Nullable String stringifyUserProperty(SDTMap properties, String key) { | ||
| try { | ||
| Object value = properties.get(key); | ||
| if (value == null) { | ||
| return null; | ||
| } | ||
| return String.valueOf(value); | ||
|
Contributor
There was a problem hiding this comment. this will invoke toString but some types (destination, stream, byte array) that are part of SDTMap don't have it and this will run poorly for those, some specialized approach should be used for those like
Contributor
Author
There was a problem hiding this comment. good catch I didn't check all the types. Maybe translating all types into string is not the good choice at the end because with this design we won't be able to reverse to Solace types. @stankiewicz what do you think if instead having a Map<String, String>, having a Map<String, UserPropertyValue> where UserPropertyValue will be a beam schema compatible model that supports all the type kinds : @AutoValue
@DefaultSchema(AutoValueSchema.class)
public abstract static class UserPropertyValue {
public enum Kind {
BOOLEAN,
BYTE,
SHORT,
INTEGER,
LONG,
FLOAT,
DOUBLE,
CHARACTER,
STRING,
BYTES,
TOPIC,
QUEUE,
MAP,
STREAM
}
public abstract Kind getKind();
public abstract @Nullable Map<String, UserPropertyValue> getMapValue();
public abstract @Nullable List<UserPropertyValue> getStreamValue();
...
} |
||
| } catch (SDTException e) { | ||
| LOG.error("Could not read user property '{}'.", key, e); | ||
|
Contributor
There was a problem hiding this comment. @ngibanel this will cause metadata loss as message will be acked. Maybe rethrowing will be better?
Contributor
Author
There was a problem hiding this comment. ok, that means the consumer must enable the dead message queue in Solace (which is a best practice) to avoid losing messages if metadata cannot be deserialized. Otherwise the broker will redeliver the message until the max retry count is reached and then message will be discarded and lost. |
||
| return null; | ||
| } | ||
| } | ||
|
|
||
| private static SDTMap createUserProperties(Map<String, String> userProperties) { | ||
| SDTMap properties = JCSMPFactory.onlyInstance().createMap(); | ||
| for (Map.Entry<String, String> entry : userProperties.entrySet()) { | ||
| try { | ||
| properties.putString(entry.getKey(), entry.getValue()); | ||
| } catch (SDTException e) { | ||
| LOG.error("Could not write user property '{}'.", entry.getKey(), e); | ||
| } | ||
| } | ||
| return properties; | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
this may be excessive, would skip this log.