[ISSUE #285] add SubscriptionManager

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/HTTPClientHandler.java

@@ -77,7 +77,7 @@ void delete(HttpExchange httpExchange) throws IOException {
             DeleteHTTPClientRequest deleteHTTPClientRequest = JsonUtils.toObject(request, DeleteHTTPClientRequest.class);
             String url = deleteHTTPClientRequest.url;
 
-            for (List<Client> clientList : eventMeshHTTPServer.localClientInfoMapping.values()) {
+            for (List<Client> clientList : eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping().values()) {
                 clientList.removeIf(client -> Objects.equals(client.getUrl(), url));
             }
 
@@ -118,7 +118,7 @@ void list(HttpExchange httpExchange) throws IOException {
             // Get the list of HTTP clients
             List<GetClientResponse> getClientResponseList = new ArrayList<>();
 
-            for (List<Client> clientList : eventMeshHTTPServer.localClientInfoMapping.values()) {
+            for (List<Client> clientList : eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping().values()) {
                 for (Client client : clientList) {
                     GetClientResponse getClientResponse = new GetClientResponse(
                         Optional.ofNullable(client.getEnv()).orElse(""),

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java

@@ -29,7 +29,7 @@
 import org.apache.eventmesh.runtime.common.ServiceState;
 import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
 import org.apache.eventmesh.runtime.constants.EventMeshConstants;
-import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
+import org.apache.eventmesh.runtime.core.consumer.SubscriptionManager;
 import org.apache.eventmesh.runtime.core.protocol.http.consumer.ConsumerManager;
 import org.apache.eventmesh.runtime.core.protocol.http.processor.AdminMetricsProcessor;
 import org.apache.eventmesh.runtime.core.protocol.http.processor.BatchSendMessageProcessor;
@@ -86,6 +86,8 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
 
     private transient ConsumerManager consumerManager;
 
+    private transient SubscriptionManager subscriptionManager;
+
     private transient ProducerManager producerManager;
 
     private transient HttpRetryer httpRetryer;
@@ -112,12 +114,6 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
 
     public transient HTTPClientPool httpClientPool = new HTTPClientPool(10);
 
-    public final ConcurrentHashMap<String /**group*/, ConsumerGroupConf> localConsumerGroupMapping =
-            new ConcurrentHashMap<>();
-
-    public final ConcurrentHashMap<String /**group@topic*/, List<Client>> localClientInfoMapping =
-            new ConcurrentHashMap<>();
-
     public EventMeshHTTPServer(final EventMeshServer eventMeshServer,
                                final EventMeshHTTPConfiguration eventMeshHttpConfiguration) {
         super(eventMeshHttpConfiguration.httpServerPort, eventMeshHttpConfiguration.eventMeshServerUseTls, eventMeshHttpConfiguration);
@@ -246,6 +242,8 @@ private void init() throws Exception {
 
         this.setMetrics(new HTTPMetricsServer(this, metricsRegistries));
 
+        subscriptionManager = new SubscriptionManager();
+
         consumerManager = new ConsumerManager(this);
         consumerManager.init();
 
@@ -409,6 +407,10 @@ private void initWebhook() throws Exception {
         this.getHandlerService().register(webHookProcessor, webhookExecutor);
     }
 
+    public SubscriptionManager getSubscriptionManager() {
+        return subscriptionManager;
+    }
+
     public ConsumerManager getConsumerManager() {
         return consumerManager;
     }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/consumer/ClientInfo.java

@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.core.consumer;
+
+public class ClientInfo {
+    private String env;
+
+    private String idc;
+
+    private String sys;
+
+    private String pid;
+
+    private String ip;
+
+    public String getEnv() {
+        return env;
+    }
+
+    public void setEnv(String env) {
+        this.env = env;
+    }
+
+    public String getIdc() {
+        return idc;
+    }
+
+    public void setIdc(String idc) {
+        this.idc = idc;
+    }
+
+    public String getSys() {
+        return sys;
+    }
+
+    public void setSys(String sys) {
+        this.sys = sys;
+    }
+
+    public String getPid() {
+        return pid;
+    }
+
+    public void setPid(String pid) {
+        this.pid = pid;
+    }
+
+    public String getIp() {
+        return ip;
+    }
+
+    public void setIp(String ip) {
+        this.ip = ip;
+    }
+}

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/consumer/SubscriptionManager.java

@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.core.consumer;
+
+import org.apache.eventmesh.common.protocol.SubscriptionItem;
+import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
+import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupTopicConf;
+import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.Client;
+
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SubscriptionManager {
+    private static final Logger logger = LoggerFactory.getLogger(SubscriptionManager.class);
+    private final ConcurrentHashMap<String /**group*/, ConsumerGroupConf> localConsumerGroupMapping =
+            new ConcurrentHashMap<>();
+
+    private final ConcurrentHashMap<String /**group@topic*/, List<Client>> localClientInfoMapping =
+            new ConcurrentHashMap<>();
+
+    public ConcurrentHashMap<String, ConsumerGroupConf> getLocalConsumerGroupMapping() {
+        return localConsumerGroupMapping;
+    }
+
+    public ConcurrentHashMap<String, List<Client>> getLocalClientInfoMapping() {
+        return localClientInfoMapping;
+    }
+
+    public void registerClient(final ClientInfo clientInfo, final String consumerGroup,
+                               final List<SubscriptionItem> subscriptionItems, final String url) {
+        for (final SubscriptionItem subscription : subscriptionItems) {
+            final String groupTopicKey = consumerGroup + "@" + subscription.getTopic();
+
+            List<Client> localClients = localClientInfoMapping.get(groupTopicKey);
+
+            if (localClients == null) {
+                localClientInfoMapping.putIfAbsent(groupTopicKey, new ArrayList<>());
+                localClients = localClientInfoMapping.get(groupTopicKey);
+            }
+
+            boolean isContains = false;
+            for (final Client localClient : localClients) {
+                //TODO: compare the whole Client would be better?
+                if (StringUtils.equals(localClient.getUrl(), url)) {
+                    isContains = true;
+                    localClient.setLastUpTime(new Date());
+                    break;
+                }
+            }
+
+            if (!isContains) {
+                Client client = new Client();
+                client.setEnv(clientInfo.getEnv());
+                client.setIdc(clientInfo.getIdc());
+                client.setSys(clientInfo.getSys());
+                client.setIp(clientInfo.getIp());
+                client.setPid(clientInfo.getPid());
+                client.setConsumerGroup(consumerGroup);
+                client.setTopic(subscription.getTopic());
+                client.setUrl(url);
+                client.setLastUpTime(new Date());
+                localClients.add(client);
+            }
+        }
+    }
+
+    public void updateSubscription(ClientInfo clientInfo, String consumerGroup,
+                                   String url, List<SubscriptionItem> subscriptionList) {
+        for (final SubscriptionItem subscription : subscriptionList) {
+            final List<Client> groupTopicClients = localClientInfoMapping
+                    .get(consumerGroup + "@" + subscription.getTopic());
+
+            if (CollectionUtils.isEmpty(groupTopicClients)) {
+                logger.error("group {} topic {} clients is empty", consumerGroup, subscription);
+            }
+
+            ConsumerGroupConf consumerGroupConf = localConsumerGroupMapping.get(consumerGroup);
+            if (consumerGroupConf == null) {
+                // new subscription
+                ConsumerGroupConf prev = localConsumerGroupMapping.putIfAbsent(consumerGroup, new ConsumerGroupConf(consumerGroup));
+                if (prev == null) {
+                    logger.info("add new subscription, consumer group: {}", consumerGroup);
+                }
+                consumerGroupConf = localConsumerGroupMapping.get(consumerGroup);
+            }
+
+            ConsumerGroupTopicConf consumerGroupTopicConf = consumerGroupConf.getConsumerGroupTopicConf()
+                    .get(subscription.getTopic());
+            if (consumerGroupTopicConf == null) {
+                consumerGroupConf.getConsumerGroupTopicConf().computeIfAbsent(subscription.getTopic(), (topic) -> {
+                    ConsumerGroupTopicConf newTopicConf = new ConsumerGroupTopicConf();
+                    newTopicConf.setConsumerGroup(consumerGroup);
+                    newTopicConf.setTopic(topic);
+                    newTopicConf.setSubscriptionItem(subscription);
+                    logger.info("add new {}", newTopicConf);
+                    return newTopicConf;
+                });
+                consumerGroupTopicConf = consumerGroupConf.getConsumerGroupTopicConf().get(subscription.getTopic());
+            }
+
+            consumerGroupTopicConf.getUrls().add(url);
+            if (!consumerGroupTopicConf.getIdcUrls().containsKey(clientInfo.getIdc())) {
+                consumerGroupTopicConf.getIdcUrls().putIfAbsent(clientInfo.getIdc(), new ArrayList<>());
+            }
+            //TODO: idcUrl list is not thread-safe
+            consumerGroupTopicConf.getIdcUrls().get(clientInfo.getIdc()).add(url);
+        }
+    }
+}

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/consumergroup/ConsumerGroupConf.java

@@ -20,14 +20,16 @@
 import java.io.Serializable;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
 
 import com.google.common.collect.Maps;
 
 public class ConsumerGroupConf implements Serializable {
     //eg . 5013-1A0
     private String consumerGroup;
 
-    private Map<String, ConsumerGroupTopicConf> consumerGroupTopicConf = Maps.newConcurrentMap();
+    private final ConcurrentHashMap<String/*topic*/, ConsumerGroupTopicConf> consumerGroupTopicConf
+            = new ConcurrentHashMap<String, ConsumerGroupTopicConf>();
 
     public ConsumerGroupConf(String consumerGroup) {
         this.consumerGroup = consumerGroup;
@@ -45,10 +47,6 @@ public Map<String, ConsumerGroupTopicConf> getConsumerGroupTopicConf() {
         return consumerGroupTopicConf;
     }
 
-    public void setConsumerGroupTopicConf(Map<String, ConsumerGroupTopicConf> consumerGroupTopicConf) {
-        this.consumerGroupTopicConf = consumerGroupTopicConf;
-    }
-
     @Override
     public boolean equals(Object o) {
         if (this == o) {

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/HttpClientGroupMapping.java

@@ -229,9 +229,7 @@ private boolean addSubscriptionByTopic(final String consumerGroup, final String
             urls.add(url);
             idcUrls.put(clientIdc, urls);
             consumeTopicConfig.setIdcUrls(idcUrls);
-            final Map<String, ConsumerGroupTopicConf> map = new HashMap<>();
-            map.put(subTopic.getTopic(), consumeTopicConfig);
-            consumerGroupConf.setConsumerGroupTopicConf(map);
+            consumerGroupConf.getConsumerGroupTopicConf().put(subTopic.getTopic(), consumeTopicConfig);
             localConsumerGroupMapping.put(consumerGroup, consumerGroupConf);
             isChange = true;
         } else {

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java

@@ -162,17 +162,17 @@ public void processRequest(final ChannelHandlerContext ctx, final AsyncContext<H
 
         }
 
-        synchronized (eventMeshHTTPServer.localClientInfoMapping) {
+        synchronized (eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping()) {
             for (final Map.Entry<String, List<Client>> groupTopicClientMapping : tmpMap.entrySet()) {
                 final List<Client> localClientList =
-                        eventMeshHTTPServer.localClientInfoMapping.get(groupTopicClientMapping.getKey());
+                        eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping().get(groupTopicClientMapping.getKey());
                 if (CollectionUtils.isEmpty(localClientList)) {
-                    eventMeshHTTPServer.localClientInfoMapping
+                    eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping()
                             .put(groupTopicClientMapping.getKey(), groupTopicClientMapping.getValue());
                 } else {
                     final List<Client> tmpClientList = groupTopicClientMapping.getValue();
                     supplyClientInfoList(tmpClientList, localClientList);
-                    eventMeshHTTPServer.localClientInfoMapping.put(groupTopicClientMapping.getKey(), localClientList);
+                    eventMeshHTTPServer.getSubscriptionManager().getLocalClientInfoMapping().put(groupTopicClientMapping.getKey(), localClientList);
                 }
 
             }