kurrent-io · w1am · Aug 4, 2025 · Aug 1, 2025
diff --git a/src/main/java/io/kurrent/dbclient/DynamicValueMapper.java b/src/main/java/io/kurrent/dbclient/DynamicValueMapper.java
@@ -0,0 +1,120 @@
+package io.kurrent.dbclient;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.Timestamp;
+import com.google.protobuf.Duration;
+import io.kurrentdb.protocol.DynamicValue;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZonedDateTime;
+import java.util.Collections;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Utility class for converting Java objects to DynamicValue protobuf messages.
+ */
+public class DynamicValueMapper {
+    private static final JsonMapper objectMapper = new JsonMapper();
+
+    /**
+     * Converts JSON byte array metadata to a Map of DynamicValue objects.
+     *
+     * @param jsonMetadata the source metadata as JSON bytes
+     * @return a map with DynamicValue objects
+     */
+    public static Map<String, DynamicValue> mapJsonToDynamicValueMap(byte[] jsonMetadata) {
+        if (jsonMetadata == null || jsonMetadata.length == 0)
+            return Collections.emptyMap();
+
+        try {
+            Map<String, Object> metadata = objectMapper.readValue(jsonMetadata, new TypeReference<Map<String, Object>>() {
+            });
+            return mapToDynamicValueMap(metadata);
+        } catch (Exception e) {
+            return Collections.emptyMap();
+        }
+    }
+
+    /**
+     * Converts a Map of metadata to a Map of DynamicValue objects.
+     *
+     * @param metadata the source metadata map
+     * @return a map with DynamicValue objects
+     */
+    public static Map<String, DynamicValue> mapToDynamicValueMap(Map<String, Object> metadata) {
+        if (metadata == null) {
+            return Collections.emptyMap();
+        }
+
+        return metadata.entrySet().stream()
+                .collect(Collectors.toMap(
+                        Map.Entry::getKey,
+                        entry -> mapToDynamicValue(entry.getValue())
+                ));
+    }
+
+    /**
+     * Converts a Java object to a DynamicValue protobuf message.
+     *
+     * @param source the source object
+     * @return the corresponding DynamicValue
+     */
+    public static DynamicValue mapToDynamicValue(Object source) {
+        if (source == null) {
+            return DynamicValue.newBuilder()
+                    .setNullValue(com.google.protobuf.NullValue.NULL_VALUE)
+                    .build();
+        }
+
+        DynamicValue.Builder builder = DynamicValue.newBuilder();
+
+        if (source instanceof String) {
+            return builder.setStringValue((String) source).build();
+        } else if (source instanceof Boolean) {
+            return builder.setBooleanValue((Boolean) source).build();
+        } else if (source instanceof Integer) {
+            return builder.setInt32Value((Integer) source).build();
+        } else if (source instanceof Long) {
+            return builder.setInt64Value((Long) source).build();
+        } else if (source instanceof Float) {
+            return builder.setFloatValue((Float) source).build();
+        } else if (source instanceof Double) {
+            return builder.setDoubleValue((Double) source).build();
+        } else if (source instanceof Instant) {
+            Instant instant = (Instant) source;
+            return builder.setTimestampValue(Timestamp.newBuilder()
+                    .setSeconds(instant.getEpochSecond())
+                    .setNanos(instant.getNano())
+                    .build()).build();
+        } else if (source instanceof LocalDateTime) {
+            LocalDateTime localDateTime = (LocalDateTime) source;
+            Instant instant = localDateTime.atZone(java.time.ZoneOffset.UTC).toInstant();
+            return builder.setTimestampValue(Timestamp.newBuilder()
+                    .setSeconds(instant.getEpochSecond())
+                    .setNanos(instant.getNano())
+                    .build()).build();
+        } else if (source instanceof ZonedDateTime) {
+            ZonedDateTime zonedDateTime = (ZonedDateTime) source;
+            Instant instant = zonedDateTime.toInstant();
+            return builder.setTimestampValue(Timestamp.newBuilder()
+                    .setSeconds(instant.getEpochSecond())
+                    .setNanos(instant.getNano())
+                    .build()).build();
+        } else if (source instanceof java.time.Duration) {
+            java.time.Duration duration = (java.time.Duration) source;
+            return builder.setDurationValue(Duration.newBuilder()
+                    .setSeconds(duration.getSeconds())
+                    .setNanos(duration.getNano())
+                    .build()).build();
+        } else if (source instanceof byte[]) {
+            return builder.setBytesValue(ByteString.copyFrom((byte[]) source)).build();
+        } else {
+            // For any other type, convert to string
+            return builder.setStringValue(source.toString()).build();
+        }
+    }
+}
diff --git a/src/main/java/io/kurrent/dbclient/EventData.java b/src/main/java/io/kurrent/dbclient/EventData.java
@@ -98,3 +98,4 @@ public static EventDataBuilder builderAsBinary(UUID eventId, String eventType, b
     }
 }
 
+
diff --git a/src/main/java/io/kurrent/dbclient/EventDataBuilder.java b/src/main/java/io/kurrent/dbclient/EventDataBuilder.java
@@ -1,8 +1,5 @@
 package io.kurrent.dbclient;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.json.JsonMapper;
-
 import java.util.UUID;
 
 /**

diff --git a/src/main/java/io/kurrent/dbclient/KurrentDBClient.java b/src/main/java/io/kurrent/dbclient/KurrentDBClient.java
@@ -75,7 +75,7 @@ public CompletableFuture<WriteResult> appendToStream(String streamName, AppendTo
         return new AppendToStream(this.getGrpcClient(), streamName, events, options).execute();
     }
 
-    public CompletableFuture<MultiAppendWriteResult> multiAppend(AppendToStreamOptions options, Iterator<AppendStreamRequest> requests) {
+    public CompletableFuture<MultiAppendWriteResult> multiStreamAppend(Iterator<AppendStreamRequest> requests) {
         return new MultiStreamAppend(this.getGrpcClient(), requests).execute();
     }
 

diff --git a/src/main/java/io/kurrent/dbclient/MultiStreamAppend.java b/src/main/java/io/kurrent/dbclient/MultiStreamAppend.java
@@ -12,6 +12,7 @@
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
 class MultiStreamAppend {
@@ -42,22 +43,29 @@ private CompletableFuture<MultiAppendWriteResult> append(WorkItemArgs args) {
            while (this.requests.hasNext()) {
                 AppendStreamRequest request = this.requests.next();
                 io.kurrentdb.protocol.streams.v2.AppendStreamRequest.Builder builder = io.kurrentdb.protocol.streams.v2.AppendStreamRequest.newBuilder()
+                        .setExpectedRevision(request.getExpectedState().toRawLong())
                         .setStream(request.getStreamName());
 
                while (request.getEvents().hasNext()) {
                      EventData event = request.getEvents().next();
-                     builder.addRecords(AppendRecord.newBuilder()
+                     AppendRecord.Builder recordBuilder = AppendRecord.newBuilder()
                              .setData(ByteString.copyFrom(event.getEventData()))
-                                     .setRecordId(event.getEventId().toString())
-                                     .putProperties(SystemMetadataKeys.DATA_FORMAT, DynamicValue
-                                             .newBuilder()
-                                             .setStringValue(ContentTypeMapper.toSchemaDataFormat(event.getContentType()))
-                                             .build())
-                                     .putProperties(SystemMetadataKeys.SCHEMA_NAME, DynamicValue
-                                             .newBuilder()
-                                             .setStringValue(event.getEventType())
-                                             .build())
+                             .setRecordId(event.getEventId().toString())
+                             .putProperties(SystemMetadataKeys.DATA_FORMAT, DynamicValue
+                                     .newBuilder()
+                                     .setStringValue(ContentTypeMapper.toSchemaDataFormat(event.getContentType()))
+                                     .build())
+                             .putProperties(SystemMetadataKeys.SCHEMA_NAME, DynamicValue
+                                     .newBuilder()
+                                     .setStringValue(event.getEventType())
                                      .build());
+
+                     if (event.getUserMetadata() != null) {
+                         Map<String, DynamicValue> userMetadataProperties = DynamicValueMapper.mapJsonToDynamicValueMap(event.getUserMetadata());
+                         recordBuilder.putAllProperties(userMetadataProperties);
+                     }
+
+                     builder.addRecords(recordBuilder.build());
                }
 
                requestStream.onNext(builder.build());
Original file line number	Diff line number	Diff line change
Expand Up		@@ -98,3 +98,4 @@ public static EventDataBuilder builderAsBinary(UUID eventId, String eventType, b
		}
		}