You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by GitBox <gi...@apache.org> on 2022/09/07 09:26:50 UTC
[GitHub] [inlong] vernedeng commented on a diff in pull request #5809: [INLONG-5808][SDK] SortSdk support parse InlongMsg
vernedeng commented on code in PR #5809:
URL: https://github.com/apache/inlong/pull/5809#discussion_r964607159
##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/decode/MessageDeserializer.java:
##########
@@ -48,12 +55,20 @@ public class MessageDeserializer implements Deserializer {
private static final String INLONG_GROUPID_KEY = "inlongGroupId";
private static final String INLONG_STREAMID_KEY = "inlongStreamId";
+ private static final String INLONGMSG_ATTR_STREAM_ID = "streamId";
+ private static final String INLONGMSG_ATTR_GROUP_ID = "groupId";
+ private static final String INLONGMSG_ATTR_TIME_T = "t";
+ private static final String INLONGMSG_ATTR_TIME_DT = "dt";
+ private static final String INLONG_ATTR_SOURCE_IP = "srcIp";
+ private static final char INLONGMSG_ATTR_ENTRY_DELIMITER = '&';
+ private static final char INLONGMSG_ATTR_KV_DELIMITER = '=';
+
public MessageDeserializer() {
}
@Override
public List<InLongMessage> deserialize(ClientContext context, InLongTopic inLongTopic, Map<String, String> headers,
- byte[] data) throws Exception {
+ byte[] data) throws Exception {
Review Comment:
fixed, thx
##########
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/decode/MessageDeserializer.java:
##########
@@ -144,4 +162,64 @@ private List<InLongMessage> transformMessageObjs(ClientContext context, InLongTo
}
return inLongMessages;
}
+
+ private List<InLongMessage> decodeInlongMsg(
+ ClientContext context,
+ InLongTopic inLongTopic,
+ byte[] msgBytes,
+ Map<String, String> headers) {
+ List<InLongMessage> messageList = new ArrayList<>();
+
+ String groupId = Optional.ofNullable(headers.get(INLONGMSG_ATTR_GROUP_ID))
+ .orElseThrow(() -> new IllegalArgumentException("Could not find "
+ + INLONGMSG_ATTR_GROUP_ID + " in attributes!"));
+
+ InLongMsg inLongMsg = InLongMsg.parseFrom(msgBytes);
+ for (String attr : inLongMsg.getAttrs()) {
+ Map<String, String> attributes = StringUtil.splitKv(attr, INLONGMSG_ATTR_ENTRY_DELIMITER,
+ INLONGMSG_ATTR_KV_DELIMITER, null, null);
+
+ /*String groupId = Optional.ofNullable(attributes.get(INLONGMSG_ATTR_GROUP_ID))
Review Comment:
fixed, thx
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org