[DXP-1336] Publishing with properties

Author: jgardo-ds

core/src/main/java/dev/streamx/cli/command/ingestion/IngestionMessageJsonFactory.java

@@ -12,23 +12,6 @@ public class IngestionMessageJsonFactory {
 
   private static final ObjectMapper mapper = new ObjectMapper();
 
-  /**
-   * @param key            of resource
-   * @param action         publish or unpublish
-   * @param payloadContent to include as a payload in returned JsonNode
-   * @param payloadType    type matching registered ingestion API schema
-   * @return JsonNode representation of {@link dev.streamx.clients.ingestion.publisher.Message}
-   */
-
-  public JsonNode from(
-      String key,
-      String action,
-      JsonNode payloadContent,
-      String payloadType
-  ) {
-    return from(key, action, payloadContent, Map.of(), payloadType);
-  }
-
   /**
    * @param key            of resource
    * @param action         publish or unpublish

core/src/main/java/dev/streamx/cli/command/ingestion/publish/PublishCommand.java

@@ -14,11 +14,13 @@
 import jakarta.inject.Inject;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Stream;
 import org.jetbrains.annotations.NotNull;
 import picocli.CommandLine.ArgGroup;
 import picocli.CommandLine.Command;
+import picocli.CommandLine.Option;
 
 @Command(name = PublishCommand.COMMAND_NAME,
     mixinStandardHelpOptions = true,
@@ -34,6 +36,10 @@ public class PublishCommand extends BaseIngestionCommand {
   @ArgGroup(exclusive = false)
   PayloadArguments payloadArguments;
 
+  @Option(names = { "-p", "--property" },
+      description = "Message property")
+  Map<String, String> properties;
+
   @Inject
   PayloadResolver payloadResolver;
 
@@ -62,6 +68,7 @@ private JsonNode prepareIngestionMessage() {
         publishArguments.getKey(),
         COMMAND_NAME,
         payload,
+        properties,
         payloadPropertyName
     );
   }

core/src/test/java/dev/streamx/cli/command/ingestion/IngestionArgumentsValidationTest.java

@@ -49,9 +49,10 @@ public void shouldRejectIngestionWithMissingParameter(QuarkusMainLauncher launch
     // then
     expectError(result, """
         Missing required parameter for option '--string-fragment' (<string>)
-        Usage: streamx publish [-hV] [[--ingestion-url=<restIngestionServiceUrl>]]
-                               (<channel> <key> [payloadFile]) [[[-s=<string> |
-                               -b=<binary> | -j=<json>]]...]
+        Usage: streamx publish [-hV] [-p=<String=String>]...
+                               [[--ingestion-url=<restIngestionServiceUrl>]] (<channel>
+                               <key> [payloadFile]) [[[-s=<string> | -b=<binary> |
+                               -j=<json>]]...]
 
         Try 'streamx publish --help' for more information.""");
   }

core/src/test/java/dev/streamx/cli/command/ingestion/publish/PublishPayloadCommandTest.java

@@ -120,6 +120,26 @@ public void shouldPublishReplacedJsonPath(QuarkusMainLauncher launcher) {
                     "{\"content\": {\"bytes\": \"<h1>Hello changed value!</h1>\"}}"))));
   }
 
+  @Test
+  public void shouldPublishWithProperties(QuarkusMainLauncher launcher) {
+    // when
+    LaunchResult result = launcher.launch("publish",
+        "--ingestion-url=" + getIngestionUrl(),
+        "-s", "content.bytes=<h1>Hello changed value!</h1>",
+        "-p", "sx:type=type/subtype",
+        CHANNEL, KEY);
+
+    // then
+    expectSuccess(result);
+    wm.verify(postRequestedFor(urlEqualTo(
+        getPublicationPath(CHANNEL)))
+        .withRequestBody(
+            equalToJson(
+                buildResponseWith(
+                    "{\"content\": {\"bytes\": \"<h1>Hello changed value!</h1>\"}}",
+                    "{ \"sx:type\": \"type/subtype\" }"))));
+  }
+
   @Test
   public void shouldPublishReplacedWithObjectJsonPath(QuarkusMainLauncher launcher) {
     // when
@@ -253,18 +273,22 @@ public void shouldPublishReplacedJsonPathWithBinaryValue(QuarkusMainLauncher lau
   }
 
   private String buildResponseWith(String content) {
+    return buildResponseWith(content, "{ }");
+  }
+
+  private String buildResponseWith(String content, String properties) {
     return
         """
             {
               "key" : "index.html",
               "action" : "publish",
               "eventTime" : null,
-              "properties" : { },
+              "properties" : %s,
               "payload" : {
                 "dev.streamx.blueprints.data.Page" : %s
               }
             }
-            """.formatted(content);
+            """.formatted(properties, content);
   }