This is an automated email from the ASF dual-hosted git repository.
stankiewicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 73bb356a408 SolaceIO: map message user properties (#40108)
73bb356a408 is described below
commit 73bb356a40861f9e9ee17afce1b13659e3ecec9c
Author: Nicolas Gibanel <[email protected]>
AuthorDate: Fri Sep 18 10:13:48 2026 +0200
SolaceIO: map message user properties (#40108)
* SolaceIO: map message user properties
* fixup: rather than trying stringify user properties, support all types
except protocol specific ones (SDTMap and SDTStream)
* fix: log.warning instead of log.info
---------
Co-authored-by: Nicolas Gibanel <[email protected]>
---
CHANGES.md | 1 +
.../solace/broker/SolaceUserPropertiesMapper.java | 200 +++++++++++++++++++++
.../org/apache/beam/sdk/io/solace/data/Solace.java | 153 +++++++++++++++-
.../sdk/io/solace/data/SolaceRecordMapperTest.java | 126 +++++++++++++
4 files changed, 478 insertions(+), 2 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 47da5f81fb0..cd4ea53eb76 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -99,6 +99,7 @@
* BigQueryIO now supports reading BigQuery Lakehouse runtime catalog (BigLake
metastore) Iceberg tables with the Storage Read API, using 4-part
`project.catalog.namespace.table` identifiers (or a `TableReference` with a
composite `catalog.namespace` dataset id). Previously such references were
silently mis-parsed (Java)
([#39597](https://github.com/apache/beam/issues/39597)) .
* SolaceIO now supports reading and writing binary and text content data
payload (Java) ([#39875](https://github.com/apache/beam/issues/39875)).
* ClickHouseIO: support writing `Decimal(P, S)` / `Decimal32/64/128/256`
columns (Java) ([#39840](https://github.com/apache/beam/issues/39840)).
+* SolaceIO now supports reading and writing user properties (message metadata)
(Java) ([#40099](https://github.com/apache/beam/issues/40099)).
* [IcebergIO] AddFiles (`IcebergAddFiles` in YAML) can evolve the table schema
before registering files, with `schema_evolution_options`, `required_columns`,
`incompatible_schema_handling` and `unverifiable_file_handling` (Java/YAML,
batch only) ([#40144](https://github.com/apache/beam/issues/40144)).
## New Features / Improvements
diff --git
a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/SolaceUserPropertiesMapper.java
b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/SolaceUserPropertiesMapper.java
new file mode 100644
index 00000000000..45e5f60e985
--- /dev/null
+++
b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/SolaceUserPropertiesMapper.java
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.solace.broker;
+
+import com.solacesystems.common.util.ByteArray;
+import com.solacesystems.jcsmp.Destination;
+import com.solacesystems.jcsmp.JCSMPFactory;
+import com.solacesystems.jcsmp.SDTException;
+import com.solacesystems.jcsmp.SDTMap;
+import com.solacesystems.jcsmp.Topic;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.beam.sdk.io.solace.data.Solace;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.Bytes;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public final class SolaceUserPropertiesMapper {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(SolaceUserPropertiesMapper.class);
+
+ private SolaceUserPropertiesMapper() {}
+
+ public static Map<String, Solace.UserPropertyValue> toUserPropertyValueMap(
+ @Nullable SDTMap properties) {
+ if (properties == null || properties.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ Map<String, Solace.UserPropertyValue> userProperties = new HashMap<>();
+ for (String key : properties.keySet()) {
+ try {
+ Object value = properties.get(key);
+ if (value == null) {
+ continue;
+ }
+
+ Solace.UserPropertyValue userPropertyValue =
toUserPropertyValue(value);
+ if (userPropertyValue.getKind() == Solace.UserPropertyValue.Kind.NONE)
{
+ LOG.warn("Unsupported user property type: {}. ", value.getClass());
+ continue;
+ }
+
+ userProperties.put(key, userPropertyValue);
+ } catch (SDTException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ return Collections.unmodifiableMap(userProperties);
+ }
+
+ public static SDTMap toSDTMap(@Nullable Map<String,
Solace.UserPropertyValue> properties) {
+ SDTMap sdtMap = JCSMPFactory.onlyInstance().createMap();
+
+ if (properties == null) {
+ return sdtMap;
+ }
+
+ for (Map.Entry<String, Solace.UserPropertyValue> entry :
properties.entrySet()) {
+ try {
+ putUserProperty(sdtMap, entry.getKey(), entry.getValue());
+ } catch (SDTException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ return sdtMap;
+ }
+
+ private static Solace.UserPropertyValue toUserPropertyValue(@NonNull Object
value) {
+ if (value instanceof Boolean) {
+ return Solace.UserPropertyValue.of((Boolean) value);
+ }
+ if (value instanceof Byte) {
+ return Solace.UserPropertyValue.of((Byte) value);
+ }
+ if (value instanceof Short) {
+ return Solace.UserPropertyValue.of((Short) value);
+ }
+ if (value instanceof Integer) {
+ return Solace.UserPropertyValue.of((Integer) value);
+ }
+ if (value instanceof Long) {
+ return Solace.UserPropertyValue.of((Long) value);
+ }
+ if (value instanceof Float) {
+ return Solace.UserPropertyValue.of((Float) value);
+ }
+ if (value instanceof Double) {
+ return Solace.UserPropertyValue.of((Double) value);
+ }
+ if (value instanceof Character) {
+ return Solace.UserPropertyValue.of((Character) value);
+ }
+ if (value instanceof String) {
+ return Solace.UserPropertyValue.of((String) value);
+ }
+ if (value instanceof byte[]) {
+ return Solace.UserPropertyValue.of(Bytes.asList((byte[]) value));
+ }
+ if (value instanceof ByteArray) {
+ return Solace.UserPropertyValue.of(Bytes.asList(((ByteArray)
value).asBytes()));
+ }
+ if (value instanceof Destination) {
+ return Solace.UserPropertyValue.of(toSolaceDestination((Destination)
value));
+ }
+ return Solace.UserPropertyValue.of();
+ }
+
+ private static Solace.Destination toSolaceDestination(Destination
destination) {
+ return Solace.Destination.builder()
+ .setType(
+ destination instanceof Topic
+ ? Solace.DestinationType.TOPIC
+ : Solace.DestinationType.QUEUE)
+ .setName(destination.getName())
+ .build();
+ }
+
+ private static Destination toDestination(Solace.Destination destination) {
+ if (destination.getType() == Solace.DestinationType.QUEUE) {
+ return JCSMPFactory.onlyInstance().createQueue(destination.getName());
+ }
+ return JCSMPFactory.onlyInstance().createTopic(destination.getName());
+ }
+
+ private static void putUserProperty(
+ SDTMap map, String key, Solace.UserPropertyValue propertyValue)
+ throws UnsupportedOperationException, SDTException {
+ if (propertyValue == null) {
+ return;
+ }
+ switch (propertyValue.getKind()) {
+ case BOOLEAN:
+ ifNotNull(propertyValue.getBoolean(), v -> map.putBoolean(key, v));
+ return;
+ case BYTE:
+ ifNotNull(propertyValue.getByte(), v -> map.putByte(key, v));
+ return;
+ case SHORT:
+ ifNotNull(propertyValue.getShort(), v -> map.putShort(key, v));
+ return;
+ case INTEGER:
+ ifNotNull(propertyValue.getInteger(), v -> map.putInteger(key, v));
+ return;
+ case LONG:
+ ifNotNull(propertyValue.getLong(), v -> map.putLong(key, v));
+ return;
+ case FLOAT:
+ ifNotNull(propertyValue.getFloat(), v -> map.putFloat(key, v));
+ return;
+ case DOUBLE:
+ ifNotNull(propertyValue.getDouble(), v -> map.putDouble(key, v));
+ return;
+ case CHARACTER:
+ ifNotNull(propertyValue.getCharacter(), v -> map.putCharacter(key, v));
+ return;
+ case STRING:
+ ifNotNull(propertyValue.getString(), v -> map.putString(key, v));
+ return;
+ case BYTES:
+ ifNotNull(propertyValue.getBytes(), v -> map.putBytes(key,
Bytes.toArray(v)));
+ return;
+ case DESTINATION:
+ ifNotNull(propertyValue.getDestination(), v -> map.putDestination(key,
toDestination(v)));
+ return;
+ case NONE:
+ default:
+ }
+ }
+
+ @FunctionalInterface
+ private interface SDTConsumer<T> {
+ void accept(@NonNull T value) throws SDTException;
+ }
+
+ private static <T> void ifNotNull(@Nullable T value, SDTConsumer<T> consumer)
+ throws SDTException {
+ if (value != null) {
+ consumer.accept(value);
+ }
+ }
+}
diff --git
a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/data/Solace.java
b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/data/Solace.java
index 15fe06103fb..b3345dad6e0 100644
---
a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/data/Solace.java
+++
b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/data/Solace.java
@@ -22,11 +22,16 @@ import com.solacesystems.jcsmp.BytesMessage;
import com.solacesystems.jcsmp.BytesXMLMessage;
import com.solacesystems.jcsmp.JCSMPFactory;
import com.solacesystems.jcsmp.TextMessage;
+import java.io.Serializable;
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.List;
+import java.util.Map;
+import org.apache.beam.sdk.io.solace.broker.SolaceUserPropertiesMapper;
import org.apache.beam.sdk.schemas.AutoValueSchema;
import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber;
@@ -83,10 +88,135 @@ public class Solace {
UNKNOWN
}
+ /** An immutable, typed value carried by a user-property map. */
+ @AutoValue
+ @DefaultSchema(AutoValueSchema.class)
+ public abstract static class UserPropertyValue {
+ public enum Kind {
+ NONE,
+ BOOLEAN,
+ BYTE,
+ SHORT,
+ INTEGER,
+ LONG,
+ FLOAT,
+ DOUBLE,
+ CHARACTER,
+ STRING,
+ BYTES,
+ DESTINATION
+ }
+
+ public abstract Kind getKind();
+
+ public abstract @Nullable Boolean getBoolean();
+
+ public abstract @Nullable Byte getByte();
+
+ public abstract @Nullable Short getShort();
+
+ public abstract @Nullable Integer getInteger();
+
+ public abstract @Nullable Long getLong();
+
+ public abstract @Nullable Float getFloat();
+
+ public abstract @Nullable Double getDouble();
+
+ public abstract @Nullable Character getCharacter();
+
+ public abstract @Nullable String getString();
+
+ public abstract @Nullable List<Byte> getBytes();
+
+ public abstract @Nullable Destination getDestination();
+
+ public static UserPropertyValue of() {
+ return builder(Kind.NONE).build();
+ }
+
+ public static UserPropertyValue of(Boolean value) {
+ return builder(Kind.BOOLEAN).setBoolean(value).build();
+ }
+
+ public static UserPropertyValue of(Byte value) {
+ return builder(Kind.BYTE).setByte(value).build();
+ }
+
+ public static UserPropertyValue of(Short value) {
+ return builder(Kind.SHORT).setShort(value).build();
+ }
+
+ public static UserPropertyValue of(Integer value) {
+ return builder(Kind.INTEGER).setInteger(value).build();
+ }
+
+ public static UserPropertyValue of(Long value) {
+ return builder(Kind.LONG).setLong(value).build();
+ }
+
+ public static UserPropertyValue of(Float value) {
+ return builder(Kind.FLOAT).setFloat(value).build();
+ }
+
+ public static UserPropertyValue of(Double value) {
+ return builder(Kind.DOUBLE).setDouble(value).build();
+ }
+
+ public static UserPropertyValue of(Character value) {
+ return builder(Kind.CHARACTER).setCharacter(value).build();
+ }
+
+ public static UserPropertyValue of(String value) {
+ return builder(Kind.STRING).setString(value).build();
+ }
+
+ public static UserPropertyValue of(List<Byte> value) {
+ return builder(Kind.BYTES).setBytes(List.copyOf(value)).build();
+ }
+
+ public static UserPropertyValue of(Destination destination) {
+ return builder(Kind.DESTINATION).setDestination(destination).build();
+ }
+
+ private static Builder builder(Kind kind) {
+ return new AutoValue_Solace_UserPropertyValue.Builder().setKind(kind);
+ }
+
+ @AutoValue.Builder
+ abstract static class Builder {
+ abstract Builder setKind(Kind value);
+
+ abstract Builder setBoolean(@Nullable Boolean value);
+
+ abstract Builder setByte(@Nullable Byte value);
+
+ abstract Builder setShort(@Nullable Short value);
+
+ abstract Builder setInteger(@Nullable Integer value);
+
+ abstract Builder setLong(@Nullable Long value);
+
+ abstract Builder setFloat(@Nullable Float value);
+
+ abstract Builder setDouble(@Nullable Double value);
+
+ abstract Builder setCharacter(@Nullable Character value);
+
+ abstract Builder setString(@Nullable String value);
+
+ abstract Builder setBytes(@Nullable List<Byte> value);
+
+ abstract Builder setDestination(@Nullable Destination value);
+
+ abstract UserPropertyValue build();
+ }
+ }
+
/** Represents a Solace message destination (either a Topic or a Queue). */
@AutoValue
@DefaultSchema(AutoValueSchema.class)
- public abstract static class Destination {
+ public abstract static class Destination implements Serializable {
/**
* Gets the name of the destination.
*
@@ -276,6 +406,15 @@ public class Solace {
@SchemaFieldNumber("13")
public abstract PayloadType getPayloadType();
+ /**
+ * Gets the typed, SDK-independent user properties of the message. Non
beam-schema compatible
+ * types are not supported (SDTMap and SDTStream)
+ *
+ * @return The user properties, or an empty map if the message carries
none.
+ */
+ @SchemaFieldNumber("14")
+ public abstract Map<String, UserPropertyValue> 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 +431,8 @@ public class Solace {
.setRedelivered(false)
.setTimeToLive(0)
.setAttachmentBytes(new byte[0])
- .setPayloadType(PayloadType.BYTES_XML);
+ .setPayloadType(PayloadType.BYTES_XML)
+ .setUserProperties(Collections.emptyMap());
}
@AutoValue.Builder
@@ -332,6 +472,8 @@ public class Solace {
public abstract Builder setAttachmentBytes(byte[] attachmentBytes);
+ public abstract Builder setUserProperties(Map<String, UserPropertyValue>
userProperties);
+
public abstract Record build();
}
@@ -456,6 +598,8 @@ public class Solace {
Destination replyTo = getDestination(msg.getCorrelationId(),
msg.getReplyTo());
Destination destination = getDestination(msg.getCorrelationId(),
msg.getDestination());
+ Map<String, UserPropertyValue> userProperties =
+
SolaceUserPropertiesMapper.toUserPropertyValueMap(msg.getProperties());
Record.Builder recordBuilder = decodePayload(msg);
return recordBuilder
@@ -473,6 +617,7 @@ public class Solace {
msg.getReplicationGroupMessageId() != null
? msg.getReplicationGroupMessageId().toString()
: null)
+ .setUserProperties(userProperties)
.build();
}
@@ -519,6 +664,10 @@ public class Solace {
msg.setSenderTimestamp(senderTimestamp);
msg.setApplicationMessageId(record.getMessageId());
+ if (!record.getUserProperties().isEmpty()) {
+
msg.setProperties(SolaceUserPropertiesMapper.toSDTMap(record.getUserProperties()));
+ }
+
return msg;
}
diff --git
a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/data/SolaceRecordMapperTest.java
b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/data/SolaceRecordMapperTest.java
index cc6567b4c88..99deac638ea 100644
---
a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/data/SolaceRecordMapperTest.java
+++
b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/data/SolaceRecordMapperTest.java
@@ -22,16 +22,22 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
+import com.solacesystems.common.util.ByteArray;
import com.solacesystems.jcsmp.BytesMessage;
import com.solacesystems.jcsmp.BytesXMLMessage;
import com.solacesystems.jcsmp.DeliveryMode;
import com.solacesystems.jcsmp.JCSMPFactory;
+import com.solacesystems.jcsmp.SDTException;
+import com.solacesystems.jcsmp.SDTMap;
import com.solacesystems.jcsmp.TextMessage;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
import org.apache.beam.sdk.io.solace.broker.MessageProducerUtils;
import org.apache.beam.sdk.io.solace.data.Solace.Record;
import org.apache.beam.sdk.io.solace.data.Solace.Record.PayloadType;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.Bytes;
import org.junit.Test;
public class SolaceRecordMapperTest {
@@ -144,6 +150,48 @@ public class SolaceRecordMapperTest {
assertEquals(789L, record.getTimeToLive());
}
+ @Test
+ public void testMapMessageUserProperties() throws SDTException {
+ BytesXMLMessage message =
JCSMPFactory.onlyInstance().createBytesXMLMessage();
+ message.setApplicationMessageId("id");
+ SDTMap properties = JCSMPFactory.onlyInstance().createMap();
+ properties.putString("string", "value");
+ properties.putBoolean("boolean", true);
+ properties.putDouble("double", 1.23);
+ properties.putFloat("float", 4.56f);
+ properties.putLong("long", 123456789L);
+ properties.putInteger("integer", 3);
+ properties.putShort("short", (short) 123);
+ properties.putCharacter("character", 'c');
+ properties.putByte("byte", (byte) 1);
+ properties.putBytes("bytes", new byte[] {1, 2, 3});
+ properties.putByteArray("byteArray", new ByteArray(new byte[] {4, 5, 6}));
+ properties.putDestination("topic",
JCSMPFactory.onlyInstance().createTopic("topic"));
+ properties.putDestination("queue",
JCSMPFactory.onlyInstance().createQueue("queue"));
+
+ // unsupported types will be ignored
+ properties.putMap("unsupported-map",
JCSMPFactory.onlyInstance().createMap());
+ properties.putStream("unsupported-stream",
JCSMPFactory.onlyInstance().createStream());
+
+ message.setProperties(properties);
+
+ Record record = Solace.SolaceRecordMapper.toRecord(message);
+
+ properties.remove("unsupported-map");
+ properties.remove("unsupported-stream");
+ assertEquals(properties.keySet(), record.getUserProperties().keySet());
+ }
+
+ @Test
+ public void testMapWithEmptyMessageUserProperties() {
+ BytesXMLMessage message =
JCSMPFactory.onlyInstance().createBytesXMLMessage();
+ message.setApplicationMessageId("id");
+
+ Record record = Solace.SolaceRecordMapper.toRecord(message);
+
+ assertTrue(record.getUserProperties().isEmpty());
+ }
+
@Test
public void testMapTextRecord() {
Record record =
@@ -257,6 +305,70 @@ public class SolaceRecordMapperTest {
assertNull(msg.getCorrelationKey());
}
+ @Test
+ public void testMapRecordUserProperties() throws Exception {
+ Map<String, Solace.UserPropertyValue> userProperties = new HashMap<>();
+ userProperties.put("string", Solace.UserPropertyValue.of("value"));
+ userProperties.put("boolean", Solace.UserPropertyValue.of(true));
+ userProperties.put("double", Solace.UserPropertyValue.of(1.23));
+ userProperties.put("float", Solace.UserPropertyValue.of(4.56f));
+ userProperties.put("long", Solace.UserPropertyValue.of(123456789L));
+ userProperties.put("integer", Solace.UserPropertyValue.of(3));
+ userProperties.put("short", Solace.UserPropertyValue.of((short) 123));
+ userProperties.put("character", Solace.UserPropertyValue.of('c'));
+ userProperties.put("byte", Solace.UserPropertyValue.of((byte) 1));
+ userProperties.put("bytes", Solace.UserPropertyValue.of(Bytes.asList(new
byte[] {1, 2, 3})));
+ userProperties.put(
+ "topic",
+ Solace.UserPropertyValue.of(
+ Solace.Destination.builder()
+ .setType(Solace.DestinationType.TOPIC)
+ .setName("topic")
+ .build()));
+ userProperties.put(
+ "queue",
+ Solace.UserPropertyValue.of(
+ Solace.Destination.builder()
+ .setType(Solace.DestinationType.QUEUE)
+ .setName("queue")
+ .build()));
+
+ Record record =
+ Record.builder()
+ .setMessageId("id")
+ .setText("test")
+ .setUserProperties(userProperties)
+ .build();
+
+ BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(record);
+
+ SDTMap expected = JCSMPFactory.onlyInstance().createMap();
+ expected.putString("string", "value");
+ expected.putBoolean("boolean", true);
+ expected.putDouble("double", 1.23);
+ expected.putFloat("float", 4.56f);
+ expected.putLong("long", 123456789L);
+ expected.putInteger("integer", 3);
+ expected.putShort("short", (short) 123);
+ expected.putCharacter("character", 'c');
+ expected.putByte("byte", (byte) 1);
+ expected.putBytes("bytes", new byte[] {1, 2, 3});
+ expected.putDestination("topic",
JCSMPFactory.onlyInstance().createTopic("topic"));
+ expected.putDestination("queue",
JCSMPFactory.onlyInstance().createQueue("queue"));
+
+ assertEquals(expected, msg.getProperties());
+ }
+
+ @Test
+ public void testMapWithEmptyRecordUserProperties() {
+ Record record =
+
Record.builder().setMessageId("id").setText("hello").setSenderTimestamp(1L).build();
+
+ BytesXMLMessage msg = Solace.SolaceRecordMapper.toMessage(record);
+
+ assertNull(msg.getProperties());
+ }
+
//
---------------------------------------------------------------------------
// round-trip
//
---------------------------------------------------------------------------
@@ -310,4 +422,18 @@ public class SolaceRecordMapperTest {
assertArrayEquals(new byte[] {1, 2}, Arrays.copyOf(decoded.getPayload(),
2));
assertArrayEquals(new byte[] {3, 4}, decoded.getAttachmentBytes());
}
+
+ private static void assertUserPropertiesEqual(
+ SDTMap expected, Map<String, Solace.UserPropertyValue> actual) throws
SDTException {
+ assertEquals(expected.keySet(), actual.keySet());
+ for (String key : expected.keySet()) {
+ Object expectedValue = expected.get(key);
+ Object actualValue = actual.get(key);
+ if (expectedValue instanceof byte[]) {
+ assertArrayEquals((byte[]) expectedValue, (byte[]) actualValue);
+ } else {
+ assertEquals(expectedValue, actualValue);
+ }
+ }
+ }
}
