feat: update stream dir field read method

2 files changed, 50 insertions, 2 deletions

diff --git a/src/main/java/com/zdjizhi/flink/voip/records/Record.java b/src/main/java/com/zdjizhi/flink/voip/records/Record.java
index fab7299..555db18 100644
--- a/src/main/java/com/zdjizhi/flink/voip/records/Record.java
+++ b/src/main/java/com/zdjizhi/flink/voip/records/Record.java
@@ -28,9 +28,12 @@ public class Record {
     /**
      * 字段名：数据记录中的流类型
      */
-    // TODO
     public static final String F_COMMON_STREAM_DIR = "common_stream_dir";
     /**
+     * 字段名：数据记录中的流类型的 Flags
+     */
+    public static final String F_FLAGS = "flags";
+    /**
      * 字段名：数据记录中的服务端地址
      */
     public static final String F_COMMON_SERVER_IP = "server_ip";
@@ -76,7 +79,7 @@ public class Record {
      * @return The stream direction.
      */
     public final StreamDir getStreamDir() {
-        return StreamDir.of(Record.getInt(obj, F_COMMON_STREAM_DIR));
+        return StreamDir.ofFlags(Record.getLong(obj, F_FLAGS));
     }
 
     /**
@@ -171,6 +174,30 @@ public class Record {
     }
 
     /**
+     * Gets a long value from the specified field in the ObjectNode.
+     *
+     * @param obj          The ObjectNode to get the value from.
+     * @param field        The name of the field.
+     * @param defaultValue The default value to return if the field is not found or is not a long.
+     * @return The long value from the field or the default value if the field is not found or is not a long.
+     */
+    public static long getLong(final ObjectNode obj, final String field, final long defaultValue) {
+        final JsonNode node = obj.get(field);
+        return node != null && node.isLong() ? node.asLong(defaultValue) : defaultValue;
+    }
+
+    /**
+     * Gets a long value from the specified field in the ObjectNode.
+     *
+     * @param obj   The ObjectNode to get the value from.
+     * @param field The name of the field.
+     * @return The long value from the field or 0 if the field is not found or is not a long.
+     */
+    private static long getLong(final ObjectNode obj, final String field) {
+        return getLong(obj, field, 0L);
+    }
+
+    /**
      * Get a string value from the specified field in the ObjectNode.
      *
      * @param obj          The ObjectNode to get the value from.
diff --git a/src/main/java/com/zdjizhi/flink/voip/records/StreamDir.java b/src/main/java/com/zdjizhi/flink/voip/records/StreamDir.java
index 84418c8..19b8a7e 100644
--- a/src/main/java/com/zdjizhi/flink/voip/records/StreamDir.java
+++ b/src/main/java/com/zdjizhi/flink/voip/records/StreamDir.java
@@ -48,4 +48,25 @@ public enum StreamDir {
         }
         throw new IllegalArgumentException("Unknown StreamDir value '" + value + "'.");
     }
+
+    /**
+     * Get the StreamDir enum based on the provided flags value.
+     *
+     * @param flags The flags.
+     * @return The corresponding StreamDir enum.
+     * @throws IllegalArgumentException if the provided value does not match any known StreamDir.
+     */
+    public static StreamDir ofFlags(long flags) {
+        int v = 0;
+        if ((flags & 8192) == 8192) {
+            v += 1;
+        }
+        if ((flags & 16384) == 16384) {
+            v += 2;
+        }
+        if ((flags & 32768) == 32768) {
+            v = 3;
+        }
+        return of(v);
+    }
 }