4 Commits

Author SHA1 Message Date
chaoc
c3f4f94a8f chore: update version 2023-12-15 10:36:11 +08:00
chaoc
07beccc732 [GAL-444] fix: fix error caused by invalid flags 2023-12-15 10:27:02 +08:00
梁超
93ed6bcddc Merge branch 'hotfix/duplicate-voip' into 'release/1.1'
merge: merge 1.2

See merge request galaxy/tsg_olap/sip-rtp-correlation!22
2023-12-08 06:36:00 +00:00
chaoc
4ab96737a1 merge: merge 1.2
fix:
1. Duplicate Voip
2. VSysID default value
3. Output sip when them not pair
2023-12-08 14:35:23 +08:00
4 changed files with 9 additions and 55 deletions

View File

@@ -7,7 +7,7 @@
<groupId>com.zdjizhi</groupId>
<artifactId>sip-rtp-correlation</artifactId>
<version>1.2</version>
<version>1.1</version>
<name>Flink : SIP-RTP : Correlation</name>

View File

@@ -59,7 +59,7 @@ public class CorrelateApp {
.<ObjectNode>forBoundedOutOfOrderness(Duration.ofSeconds(5))
.withTimestampAssigner((SerializableTimestampAssigner<ObjectNode>)
(element, recordTimestamp) ->
element.get("start_timestamp_ms").asLong()));
element.get("common_start_timestamp_ms").asLong()));
final ErrorHandler errorHandler = new ErrorHandler(config);

View File

@@ -20,35 +20,31 @@ public class Record {
/**
* 字段名:数据记录中的所属 vsys
*/
public static final String F_COMMON_VSYS_ID = "vsys_id";
public static final String F_COMMON_VSYS_ID = "common_vsys_id";
/**
* 字段名:数据记录中的字段类型
*/
public static final String F_COMMON_SCHEMA_TYPE = "decoded_as";
public static final String F_COMMON_SCHEMA_TYPE = "common_schema_type";
/**
* 字段名:数据记录中的流类型
*/
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";
public static final String F_COMMON_SERVER_IP = "common_server_ip";
/**
* 字段名:数据记录中的服务端端口
*/
public static final String F_COMMON_SERVER_PORT = "server_port";
public static final String F_COMMON_SERVER_PORT = "common_server_port";
/**
* 字段名:数据记录中的客户端地址
*/
public static final String F_COMMON_CLIENT_IP = "client_ip";
public static final String F_COMMON_CLIENT_IP = "common_client_ip";
/**
* 字段名:数据记录中的客户端端口
*/
public static final String F_COMMON_CLIENT_PORT = "client_port";
public static final String F_COMMON_CLIENT_PORT = "common_client_port";
/**
* ObjectNode data.
@@ -80,7 +76,7 @@ public class Record {
* @return The stream direction.
*/
public final StreamDir getStreamDir() {
return StreamDir.ofFlags(Record.getLong(obj, F_FLAGS));
return StreamDir.of(Record.getInt(obj, F_COMMON_STREAM_DIR));
}
/**
@@ -174,30 +170,6 @@ public class Record {
return getInt(obj, field, 0);
}
/**
* 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.isNumber() ? node.asLong() : 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.
*

View File

@@ -48,22 +48,4 @@ 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;
}
return of(v);
}
}