[ISSUE apache#366] remove custom concept [dcn&&region] (apache#390)

* remove custom concept{dcn} * remove custom concept{dcn} * remove custom concept{dcn} * remove custom concept{dcn} * remove custom concept{dcn} * remove custom concept{dcn} * remove custom concept{dcn} close apache#366
jjz921024 · Jul 25, 2021 · 9000aeb · 9000aeb
1 parent 51b9672
commit 9000aeb
Show file tree

Hide file tree

Showing 82 changed files with 403 additions and 1,169 deletions.
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java
@@ -21,10 +21,6 @@ public class Constants {
 
     public static final String DEFAULT_CHARSET = "UTF-8";
 
-    public static final String TARGET_EVENTMESH_REGION = "TARGET_EVENTMESH_REGION";
-
-    public static final String CONSTANTS_DEFAULT_REGION_KEY = "default";
-
     public static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss.SSS";
 
     public static final String LANGUAGE_JAVA = "JAVA";
@@ -47,8 +43,6 @@ public class Constants {
 
     public static final String CONSTANTS_INSTANCE_DESC_IDC = "idc";
 
-    public static final String CONSTANTS_INSTANCE_DESC_DCN = "dcn";
-
     public static final String CONSTANTS_INSTANCE_DESC_SYSID = "sysId";
 
     public static final String CONSTANTS_INSTANCE_DESC_IP = "ip";

diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/command/HttpCommand.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/command/HttpCommand.java
@@ -17,19 +17,9 @@
 
 package org.apache.eventmesh.common.command;
 
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
-
 import com.alibaba.fastjson.JSON;
-
 import io.netty.buffer.Unpooled;
-import io.netty.handler.codec.http.DefaultFullHttpResponse;
-import io.netty.handler.codec.http.HttpHeaderNames;
-import io.netty.handler.codec.http.HttpHeaderValues;
-import io.netty.handler.codec.http.HttpHeaders;
-import io.netty.handler.codec.http.HttpResponseStatus;
-import io.netty.handler.codec.http.HttpVersion;
-
+import io.netty.handler.codec.http.*;
 import org.apache.commons.collections4.MapUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.eventmesh.common.Constants;
@@ -38,6 +28,9 @@
 import org.apache.eventmesh.common.protocol.http.header.BaseResponseHeader;
 import org.apache.eventmesh.common.protocol.http.header.Header;
 
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
 public class HttpCommand {
 
     private static AtomicLong requestId = new AtomicLong(0);

diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/CommonConfiguration.java
@@ -31,9 +31,7 @@
 
 public class CommonConfiguration {
     public String eventMeshEnv = "P";
-    public String eventMeshRegion = "";
     public String eventMeshIDC = "FT";
-    public String eventMeshDCN = "1C0";
     public String eventMeshCluster = "LS";
     public String eventMeshName = "";
     public String sysID = "5477";
@@ -68,10 +66,6 @@ public void init() {
             Preconditions.checkState(StringUtils.isNotEmpty(eventMeshEnvStr), String.format("%s error", ConfKeys.KEYS_EVENTMESH_ENV));
             eventMeshEnv = StringUtils.deleteWhitespace(eventMeshEnvStr);
 
-            String eventMeshRegionStr = configurationWraper.getProp(ConfKeys.KEYS_EVENTMESH_REGION);
-            Preconditions.checkState(StringUtils.isNotEmpty(eventMeshRegionStr), String.format("%s error", ConfKeys.KEYS_EVENTMESH_REGION));
-            eventMeshRegion = StringUtils.deleteWhitespace(eventMeshRegionStr);
-
             String sysIdStr = configurationWraper.getProp(ConfKeys.KEYS_EVENTMESH_SYSID);
             Preconditions.checkState(StringUtils.isNotEmpty(sysIdStr) && StringUtils.isNumeric(sysIdStr), String.format("%s error", ConfKeys.KEYS_EVENTMESH_SYSID));
             sysID = StringUtils.deleteWhitespace(sysIdStr);
@@ -88,10 +82,6 @@ public void init() {
             Preconditions.checkState(StringUtils.isNotEmpty(eventMeshIDCStr), String.format("%s error", ConfKeys.KEYS_EVENTMESH_IDC));
             eventMeshIDC = StringUtils.deleteWhitespace(eventMeshIDCStr);
 
-            String eventMeshDCNStr = configurationWraper.getProp(ConfKeys.KEYS_EVENTMESH_DCN);
-            Preconditions.checkState(StringUtils.isNotEmpty(eventMeshDCNStr), String.format("%s error", ConfKeys.KEYS_EVENTMESH_DCN));
-            eventMeshDCN = StringUtils.deleteWhitespace(eventMeshDCNStr);
-
             eventMeshServerIp = configurationWraper.getProp(ConfKeys.KEYS_EVENTMESH_SERVER_HOST_IP);
             if (StringUtils.isBlank(eventMeshServerIp)) {
                 eventMeshServerIp = getLocalAddr();
@@ -102,12 +92,8 @@ public void init() {
     static class ConfKeys {
         public static String KEYS_EVENTMESH_ENV = "eventMesh.server.env";
 
-        public static String KEYS_EVENTMESH_REGION = "eventMesh.server.region";
-
         public static String KEYS_EVENTMESH_IDC = "eventMesh.server.idc";
 
-        public static String KEYS_EVENTMESH_DCN = "eventMesh.server.dcn";
-
         public static String KEYS_EVENTMESH_SYSID = "eventMesh.sysid";
 
         public static String KEYS_EVENTMESH_SERVER_CLUSTER = "eventMesh.server.cluster";

diff --git a/...main/java/org/apache/eventmesh/common/protocol/http/body/client/HeartbeatRequestBody.java b/...main/java/org/apache/eventmesh/common/protocol/http/body/client/HeartbeatRequestBody.java
@@ -31,6 +31,9 @@ public class HeartbeatRequestBody extends Body {
 
     public static final String CLIENTTYPE = "clientType";
     public static final String HEARTBEATENTITIES = "heartbeatEntities";
+    public static final String CONSUMERGROUP = "consumerGroup";
+
+    private String consumerGroup;
 
     private String clientType;
 
@@ -52,9 +55,18 @@ public void setHeartbeatEntities(List<HeartbeatEntity> heartbeatEntities) {
         this.heartbeatEntities = heartbeatEntities;
     }
 
+    public String getConsumerGroup() {
+        return consumerGroup;
+    }
+
+    public void setConsumerGroup(String consumerGroup) {
+        this.consumerGroup = consumerGroup;
+    }
+
     public static HeartbeatRequestBody buildBody(Map<String, Object> bodyParam) {
         HeartbeatRequestBody body = new HeartbeatRequestBody();
         body.setClientType(MapUtils.getString(bodyParam, CLIENTTYPE));
+        body.setConsumerGroup(MapUtils.getString(bodyParam, CONSUMERGROUP));
         body.setHeartbeatEntities(JSONArray.parseArray(MapUtils.getString(bodyParam, HEARTBEATENTITIES), HeartbeatEntity.class));
         return body;
     }
@@ -63,6 +75,7 @@ public static HeartbeatRequestBody buildBody(Map<String, Object> bodyParam) {
     public Map<String, Object> toMap() {
         Map<String, Object> map = new HashMap<String, Object>();
         map.put(CLIENTTYPE, clientType);
+        map.put(CONSUMERGROUP, consumerGroup);
         map.put(HEARTBEATENTITIES, JSON.toJSONString(heartbeatEntities));
         return map;
     }
@@ -89,6 +102,7 @@ public String toString() {
     public String toString() {
         StringBuilder sb = new StringBuilder();
         sb.append("heartbeatRequestBody={")
+                .append("consumerGroup=").append(consumerGroup).append(",")
                 .append("clientType=").append(clientType).append("}");
         return sb.toString();
     }

diff --git a/...main/java/org/apache/eventmesh/common/protocol/http/body/client/SubscribeRequestBody.java b/...main/java/org/apache/eventmesh/common/protocol/http/body/client/SubscribeRequestBody.java
@@ -34,8 +34,22 @@ public class SubscribeRequestBody extends Body {
 
     public static final String URL = "url";
 
+    public static final String CONSUMERGROUP = "consumerGroup";
+
     private List<SubscriptionItem> topics;
 
+    private String url;
+
+    private String consumerGroup;
+
+    public String getConsumerGroup() {
+        return consumerGroup;
+    }
+
+    public void setConsumerGroup(String consumerGroup) {
+        this.consumerGroup = consumerGroup;
+    }
+
     public List<SubscriptionItem> getTopics() {
         return topics;
     }
@@ -44,8 +58,6 @@ public void setTopics(List<SubscriptionItem> topics) {
         this.topics = topics;
     }
 
-    private String url;
-
     public String getUrl() {
         return url;
     }
@@ -58,6 +70,7 @@ public static SubscribeRequestBody buildBody(Map<String, Object> bodyParam) {
         SubscribeRequestBody body = new SubscribeRequestBody();
         body.setUrl(MapUtils.getString(bodyParam, URL));
         body.setTopics(JSONArray.parseArray(MapUtils.getString(bodyParam, TOPIC), SubscriptionItem.class));
+        body.setConsumerGroup(MapUtils.getString(bodyParam, CONSUMERGROUP));
         return body;
     }
 
@@ -66,13 +79,15 @@ public Map<String, Object> toMap() {
         Map<String, Object> map = new HashMap<String, Object>();
         map.put(URL, url);
         map.put(TOPIC, JSON.toJSONString(topics));
+        map.put(CONSUMERGROUP, consumerGroup);
         return map;
     }
 
     @Override
     public String toString() {
         return "subscribeBody{" +
-                "url='" + url + '\'' +
+                "consumerGroup='" + consumerGroup + '\'' +
+                ", url='" + url + '\'' +
                 ", topics=" + topics +
                 '}';
     }

diff --git a/...in/java/org/apache/eventmesh/common/protocol/http/body/client/UnSubscribeRequestBody.java b/...in/java/org/apache/eventmesh/common/protocol/http/body/client/UnSubscribeRequestBody.java
@@ -33,10 +33,14 @@ public class UnSubscribeRequestBody extends Body {
 
     public static final String URL = "url";
 
+    public static final String CONSUMERGROUP = "consumerGroup";
+
     private List<String> topics;
 
     private String url;
 
+    private String consumerGroup;
+
     public List<String> getTopics() {
         return topics;
     }
@@ -53,10 +57,19 @@ public void setUrl(String url) {
         this.url = url;
     }
 
+    public String getConsumerGroup() {
+        return consumerGroup;
+    }
+
+    public void setConsumerGroup(String consumerGroup) {
+        this.consumerGroup = consumerGroup;
+    }
+
     public static UnSubscribeRequestBody buildBody(Map<String, Object> bodyParam) {
         UnSubscribeRequestBody body = new UnSubscribeRequestBody();
         body.setUrl(MapUtils.getString(bodyParam, URL));
         body.setTopics(JSONArray.parseArray(MapUtils.getString(bodyParam, TOPIC), String.class));
+        body.setConsumerGroup(MapUtils.getString(bodyParam, CONSUMERGROUP));
         return body;
     }
 
@@ -65,13 +78,15 @@ public Map<String, Object> toMap() {
         Map<String, Object> map = new HashMap<String, Object>();
         map.put(URL, url);
         map.put(TOPIC, JSON.toJSONString(topics));
+        map.put(CONSUMERGROUP, consumerGroup);
         return map;
     }
 
     @Override
     public String toString() {
         return "unSubscribeRequestBody{" +
-                "url='" + url + '\'' +
+                "consumerGroup='" + consumerGroup + '\'' +
+                ", url='" + url + '\'' +
                 ", topics=" + topics +
                 '}';
     }

diff --git a/.../java/org/apache/eventmesh/common/protocol/http/body/message/ReplyMessageRequestBody.java b/.../java/org/apache/eventmesh/common/protocol/http/body/message/ReplyMessageRequestBody.java
@@ -29,12 +29,12 @@
 
 public class ReplyMessageRequestBody extends Body {
 
-
     public static final String ORIGTOPIC = "origTopic";
     public static final String BIZSEQNO = "bizSeqNo";
     public static final String UNIQUEID = "uniqueId";
     public static final String CONTENT = "content";
     public static final String EXTFIELDS = "extFields";
+    public static final String PRODUCERGROUP = "producerGroup";
 
     private String bizSeqNo;
 
@@ -46,6 +46,8 @@ public class ReplyMessageRequestBody extends Body {
 
     private HashMap<String, String> extFields;
 
+    private String producerGroup;
+
     public String getOrigTopic() {
         return origTopic;
     }
@@ -86,6 +88,14 @@ public void setExtFields(HashMap<String, String> extFields) {
         this.extFields = extFields;
     }
 
+    public String getProducerGroup() {
+        return producerGroup;
+    }
+
+    public void setProducerGroup(String producerGroup) {
+        this.producerGroup = producerGroup;
+    }
+
     @SuppressWarnings("unchecked")
     public static ReplyMessageRequestBody buildBody(Map<String, Object> bodyParam) {
         ReplyMessageRequestBody body = new ReplyMessageRequestBody();
@@ -97,6 +107,7 @@ public static ReplyMessageRequestBody buildBody(Map<String, Object> bodyParam) {
         if (StringUtils.isNotBlank(extFields)) {
             body.setExtFields((HashMap<String, String>) JSONObject.parseObject(extFields, HashMap.class));
         }
+        body.setProducerGroup(MapUtils.getString(bodyParam, PRODUCERGROUP));
         return body;
     }
 
@@ -108,6 +119,7 @@ public String toString() {
                 .append("uniqueId=").append(uniqueId).append(",")
                 .append("origTopic=").append(origTopic).append(",")
                 .append("content=").append(content).append(",")
+                .append("producerGroup=").append(producerGroup).append(",")
                 .append("extFields=").append(extFields).append("}");
         return sb.toString();
     }
@@ -120,6 +132,7 @@ public Map<String, Object> toMap() {
         map.put(UNIQUEID, uniqueId);
         map.put(CONTENT, content);
         map.put(EXTFIELDS, JSON.toJSONString(extFields));
+        map.put(PRODUCERGROUP, producerGroup);
         return map;
     }
 }
diff --git a/...a/org/apache/eventmesh/common/protocol/http/body/message/SendMessageBatchRequestBody.java b/...a/org/apache/eventmesh/common/protocol/http/body/message/SendMessageBatchRequestBody.java
@@ -33,13 +33,16 @@ public class SendMessageBatchRequestBody extends Body {
     public static final String BATCHID = "batchId";
     public static final String CONTENTS = "contents";
     public static final String SIZE = "size";
+    public static final String PRODUCERGROUP = "producerGroup";
 
     private String batchId;
 
     private List<BatchMessageEntity> contents;
 
     private String size;
 
+    private String producerGroup;
+
     public SendMessageBatchRequestBody() {
     }
 
@@ -67,12 +70,21 @@ public void setSize(String size) {
         this.size = size;
     }
 
+    public String getProducerGroup() {
+        return producerGroup;
+    }
+
+    public void setProducerGroup(String producerGroup) {
+        this.producerGroup = producerGroup;
+    }
+
     @Override
     public String toString() {
         StringBuilder sb = new StringBuilder();
         sb.append("sendMessageBatchRequestBody={")
                 .append("batchId=").append(batchId).append(",")
                 .append("size=").append(size).append(",")
+                .append("producerGroup=").append(producerGroup).append(",")
                 .append("contents=").append(JSON.toJSONString(contents)).append("}");
         return sb.toString();
     }
@@ -111,6 +123,7 @@ public static SendMessageBatchRequestBody buildBody(final Map<String, Object> bo
             body.setContents(JSONArray.parseArray(contents, BatchMessageEntity.class));
         }
         body.setSize(size);
+        body.setProducerGroup(MapUtils.getString(bodyParam, PRODUCERGROUP));
         return body;
     }
 
@@ -120,6 +133,7 @@ public Map<String, Object> toMap() {
         map.put(BATCHID, batchId);
         map.put(SIZE, size);
         map.put(CONTENTS, contents);
+        map.put(PRODUCERGROUP, producerGroup);
         return map;
     }