Include data payload as hex in the output of log command

aklivity · Oct 25, 2023 · 3f02596 · 3f02596
1 parent f1985a3
commit 3f02596
Show file tree

Hide file tree

Showing 3 changed files with 35 additions and 2 deletions.
diff --git a/.../command-log/src/main/java/io/aklivity/zilla/runtime/command/log/internal/LogCommand.java b/.../command-log/src/main/java/io/aklivity/zilla/runtime/command/log/internal/LogCommand.java
@@ -65,12 +65,14 @@ public static void main(
         options.addOption(builder("i").hasArg().longOpt("interval").desc("run command continuously at interval").build());
         options.addOption(builder("s").longOpt("separator").desc("include thousands separator in integer values").build());
         options.addOption(builder("a").hasArg().longOpt("affinity").desc("affinity mask").build());
+        options.addOption(builder("p").longOpt("payload").desc("include data payloads as hex").build());
 
         final CommandLine cmdline = parser.parse(options, args);
         final Logger out = System.out::printf;
 
         final boolean hasVersion = cmdline.hasOption("version");
         final boolean hasDirectory = cmdline.hasOption("directory");
+        final boolean isPayloadIncluded = cmdline.hasOption("payload");
         final boolean hasHelp = cmdline.hasOption("help");
 
         if (hasVersion)
@@ -120,7 +122,7 @@ public static void main(
                     frameTypes == null ? t -> true : Arrays.asList(frameTypes)::contains;
 
                 command = new LogStreamsCommand(config, out, hasFrameTypes, hasExtensionType, verbose,
-                    continuous, affinity, position);
+                    continuous, isPayloadIncluded, affinity, position);
             }
             else if ("buffers".equals(type))
             {

diff --git a/...d-log/src/main/java/io/aklivity/zilla/runtime/command/log/internal/LogStreamsCommand.java b/...d-log/src/main/java/io/aklivity/zilla/runtime/command/log/internal/LogStreamsCommand.java
@@ -50,6 +50,7 @@ public final class LogStreamsCommand implements Runnable
     private final LabelManager labels;
     private final boolean verbose;
     private final boolean continuous;
+    private final boolean isPayloadIncluded;
     private final long affinity;
     private final SpyPosition position;
     private final Logger out;
@@ -63,13 +64,15 @@ public final class LogStreamsCommand implements Runnable
         Predicate<String> hasExtensionType,
         boolean verbose,
         boolean continuous,
+        boolean isPayloadIncluded,
         long affinity,
         SpyPosition position)
     {
         this.directory = config.directory();
         this.labels = new LabelManager(directory);
         this.verbose = verbose;
         this.continuous = continuous;
+        this.isPayloadIncluded = isPayloadIncluded;
         this.affinity = affinity;
         this.position = position;
         this.out = out;
@@ -104,7 +107,8 @@ private LoggableStream newLoggable(
                 .spyAt(position)
                 .build();
 
-        return new LoggableStream(index, labels, layout, out, hasFrameType, hasExtensionType, this::nextTimestamp);
+        return new LoggableStream(index, labels, layout, out, hasFrameType, hasExtensionType, isPayloadIncluded,
+            this::nextTimestamp);
     }
 
     private void onDiscovered(

diff --git a/...mand-log/src/main/java/io/aklivity/zilla/runtime/command/log/internal/LoggableStream.java b/...mand-log/src/main/java/io/aklivity/zilla/runtime/command/log/internal/LoggableStream.java
@@ -157,6 +157,7 @@ public final class LoggableStream implements AutoCloseable
     private final StreamsLayout layout;
     private final RingBufferSpy streamsBuffer;
     private final Logger out;
+    private final boolean isPayloadIncluded;
     private final LongPredicate nextTimestamp;
 
     private final Int2ObjectHashMap<MessageConsumer> frameHandlers;
@@ -173,6 +174,7 @@ public final class LoggableStream implements AutoCloseable
         Logger logger,
         Predicate<String> hasFrameType,
         Predicate<String> hasExtensionType,
+        boolean isPayloadIncluded,
         LongPredicate nextTimestamp)
     {
         this.index = index;
@@ -184,6 +186,7 @@ public final class LoggableStream implements AutoCloseable
         this.layout = layout;
         this.streamsBuffer = layout.streamsBuffer();
         this.out = logger;
+        this.isPayloadIncluded = isPayloadIncluded;
         this.nextTimestamp = nextTimestamp;
 
         final Int2ObjectHashMap<MessageConsumer> frameHandlers = new Int2ObjectHashMap<>();
@@ -387,6 +390,11 @@ private void onData(
             sequence - acknowledge + reserved, maximum, format("DATA [0x%016x] [%d] [%d] [%x] [0x%016x]",
                     budgetId, length, reserved, flags, authorization));
 
+        if (isPayloadIncluded)
+        {
+            onDataPayload(data);
+        }
+
         final ExtensionFW extension = data.extension().get(extensionRO::tryWrap);
         if (extension != null)
         {
@@ -398,6 +406,25 @@ private void onData(
         }
     }
 
+    private void onDataPayload(
+        final DataFW data)
+    {
+        final int offset = data.offset() - HEADER_LENGTH;
+        final long timestamp = data.timestamp();
+        final OctetsFW payload = data.payload();
+
+        if (payload != null)
+        {
+            StringBuilder hexData = new StringBuilder();
+            for (int i = 0; i < data.length(); i++)
+            {
+                hexData.append(String.format("%02x", payload.buffer().getByte(payload.offset() + i)));
+                hexData.append(i < data.length() - 1 ? ':' : "");
+            }
+            out.printf(verboseFormat, index, offset, timestamp, hexData);
+        }
+    }
+
     private void onEnd(
         final EndFW end)
     {