From c075a689c5ef2d213ae027208152c1c8cf7bec30 Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Tue, 31 May 2022 13:23:53 +0800
Subject: [PATCH 01/17] add consul registry implement

---
 .../eventmesh-registry-consul/build.gradle    |  24 +++
 .../consul/constant/ConsulConstant.java       |  45 +++++
 .../consul/service/ConsulRegistryService.java | 165 ++++++++++++++++++
 .../test/java/ConsulRegistryServiceTest.java  | 129 ++++++++++++++
 4 files changed, 363 insertions(+)
 create mode 100644 eventmesh-registry-plugin/eventmesh-registry-consul/build.gradle
 create mode 100644 eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/constant/ConsulConstant.java
 create mode 100644 eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
 create mode 100644 eventmesh-registry-plugin/eventmesh-registry-consul/src/test/java/ConsulRegistryServiceTest.java

diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/build.gradle b/eventmesh-registry-plugin/eventmesh-registry-consul/build.gradle
new file mode 100644
index 0000000000..53d7f67b4c
--- /dev/null
+++ b/eventmesh-registry-plugin/eventmesh-registry-consul/build.gradle
@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+dependencies {
+    implementation 'com.ecwid.consul:consul-api:1.4.5'
+    implementation 'org.apache.httpcomponents:httpclient:4.5.13'
+    implementation project(":eventmesh-registry-plugin:eventmesh-registry-api")
+    implementation project(":eventmesh-common")
+    testImplementation "org.mockito:mockito-core"
+}
diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/constant/ConsulConstant.java b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/constant/ConsulConstant.java
new file mode 100644
index 0000000000..bd8aa876c6
--- /dev/null
+++ b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/constant/ConsulConstant.java
@@ -0,0 +1,45 @@
+/*
+ * 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.registry.consul.constant;
+
+public class ConsulConstant {
+
+    public static final int DEFAULT_MAX_CONNECTIONS = 1000;
+
+    public static final int DEFAULT_MAX_PER_ROUTE_CONNECTIONS = 500;
+
+    /**
+     * 3 sec
+     */
+    public static final int DEFAULT_CONNECTION_TIMEOUT = 3000;
+
+    /**
+     * 2 sec
+     * 0 minutes for read timeout due to blocking queries timeout
+     * <a href="https://www.consul.io/api/index.html#blocking-queries">...</a>
+     */
+    public static final int DEFAULT_READ_TIMEOUT = 2000;
+
+    public static final String SERVICE_ADDR = "serverAddr";
+
+    public static final String IP_PORT_SEPARATOR = ":";
+
+    public static final String DEFAULT_GROUP = "DEFAULT_GROUP";
+
+
+}
diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
new file mode 100644
index 0000000000..3deea5cda5
--- /dev/null
+++ b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
@@ -0,0 +1,165 @@
+/*
+ * 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.registry.consul.service;
+
+import com.ecwid.consul.v1.ConsulClient;
+import com.ecwid.consul.v1.ConsulRawClient;
+import com.ecwid.consul.v1.agent.model.NewService;
+import com.ecwid.consul.v1.agent.model.Service;
+import com.ecwid.consul.v1.health.HealthServicesRequest;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.eventmesh.api.exception.RegistryException;
+import org.apache.eventmesh.api.registry.RegistryService;
+import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo;
+import org.apache.eventmesh.api.registry.dto.EventMeshRegisterInfo;
+import org.apache.eventmesh.api.registry.dto.EventMeshUnRegisterInfo;
+import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.apache.eventmesh.registry.consul.constant.ConsulConstant.*;
+
+public class ConsulRegistryService implements RegistryService {
+
+    private static final Logger logger = LoggerFactory.getLogger(ConsulRegistryService.class);
+
+    private static final AtomicBoolean INIT_STATUS = new AtomicBoolean(false);
+
+    private static final AtomicBoolean START_STATUS = new AtomicBoolean(false);
+
+    private String consulHost;
+
+    private String consulPort;
+
+    private ConsulClient consulClient;
+
+    @Override
+    public void init() throws RegistryException {
+        if (INIT_STATUS.compareAndSet(false, true)) {
+            for (String key : ConfigurationContextUtil.KEYS) {
+                CommonConfiguration commonConfiguration = ConfigurationContextUtil.get(key);
+                if (null != commonConfiguration) {
+                    String namesrvAddr = commonConfiguration.namesrvAddr;
+                    if (StringUtils.isBlank(namesrvAddr)) {
+                        throw new RegistryException("namesrvAddr cannot be null");
+                    }
+                    String[] addr = namesrvAddr.split(":");
+                    if (addr.length != 2) {
+                        throw new RegistryException("Illegal namesrvAddr");
+                    }
+                    this.consulHost = addr[0];
+                    this.consulPort = addr[1];
+                    break;
+                }
+            }
+        }
+    }
+
+    @Override
+    public void start() throws RegistryException {
+        consulClient = new ConsulClient(new ConsulRawClient(consulHost, Integer.parseInt(consulPort), getHttpClient()));
+    }
+
+    @Override
+    public void shutdown() throws RegistryException {
+        INIT_STATUS.compareAndSet(true, false);
+        START_STATUS.compareAndSet(true, false);
+        consulClient = null;
+    }
+
+    @Override
+    public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo) throws RegistryException {
+        try {
+            String[] ipPort = eventMeshRegisterInfo.getEndPoint().split(IP_PORT_SEPARATOR);
+            NewService service = new NewService();
+            service.setPort(Integer.parseInt(ipPort[1]));
+            service.setAddress(ipPort[0]);
+            service.setName(eventMeshRegisterInfo.getEventMeshClusterName());
+            service.setId(eventMeshRegisterInfo.getEventMeshClusterName() + "-" + eventMeshRegisterInfo.getEventMeshName());
+            consulClient.agentServiceRegister(service);
+        } catch (Exception e) {
+            throw new RegistryException(e.getMessage());
+        }
+        logger.info("EventMesh successfully registered to consul");
+        return true;
+    }
+
+    @Override
+    public boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo) throws RegistryException {
+        try {
+            consulClient.agentServiceDeregister(eventMeshUnRegisterInfo.getEventMeshClusterName() + "-" + eventMeshUnRegisterInfo.getEventMeshName());
+        } catch (Exception e) {
+            throw new RegistryException(e.getMessage());
+        }
+        logger.info("EventMesh successfully unregistered to consul");
+        return true;
+    }
+
+    @Override
+    public List<EventMeshDataInfo> findEventMeshInfoByCluster(String clusterName) throws RegistryException {
+        Map<String, Service> agentServices = consulClient.getAgentServices().getValue();
+        HealthServicesRequest request = HealthServicesRequest.newBuilder().setPassing(true).build();
+        consulClient.getHealthServices(clusterName, request);
+        List<EventMeshDataInfo> eventMeshDataInfos = new ArrayList<>();
+        agentServices.forEach((k, v) -> {
+            String[] split = v.getId().split("-");
+            eventMeshDataInfos.add(new EventMeshDataInfo(split[0], split[1], v.getAddress() + ":" + v.getPort(),0));
+        });
+        return eventMeshDataInfos;
+    }
+
+    @Override
+    public Map<String, Map<String, Integer>> findEventMeshClientDistributionData(String clusterName, String group, String purpose) throws RegistryException {
+        return Collections.emptyMap();
+    }
+
+
+    private HttpClient getHttpClient() {
+        PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
+        connectionManager.setMaxTotal(DEFAULT_MAX_CONNECTIONS);
+        connectionManager.setDefaultMaxPerRoute(DEFAULT_MAX_PER_ROUTE_CONNECTIONS);
+
+        RequestConfig requestConfig = RequestConfig.custom().
+                setConnectTimeout(DEFAULT_CONNECTION_TIMEOUT).
+                setConnectionRequestTimeout(DEFAULT_CONNECTION_TIMEOUT).
+                setSocketTimeout(DEFAULT_READ_TIMEOUT).
+                build();
+
+        HttpClientBuilder httpClientBuilder = HttpClientBuilder.create().
+                setConnectionManager(connectionManager).
+                setDefaultRequestConfig(requestConfig).
+                useSystemProperties();
+
+        return httpClientBuilder.build();
+    }
+
+    public ConsulClient getConsulClient() {
+        return consulClient;
+    }
+}
diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/src/test/java/ConsulRegistryServiceTest.java b/eventmesh-registry-plugin/eventmesh-registry-consul/src/test/java/ConsulRegistryServiceTest.java
new file mode 100644
index 0000000000..a8c4c78391
--- /dev/null
+++ b/eventmesh-registry-plugin/eventmesh-registry-consul/src/test/java/ConsulRegistryServiceTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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.
+ */
+
+import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo;
+import org.apache.eventmesh.api.registry.dto.EventMeshRegisterInfo;
+import org.apache.eventmesh.api.registry.dto.EventMeshUnRegisterInfo;
+import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
+import org.apache.eventmesh.registry.consul.service.ConsulRegistryService;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import java.lang.reflect.Field;
+import java.util.List;
+
+/**
+ * @author huyuanxin
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class ConsulRegistryServiceTest {
+
+    @Mock
+    private EventMeshRegisterInfo eventMeshRegisterInfo;
+    @Mock
+    private EventMeshUnRegisterInfo eventMeshUnRegisterInfo;
+
+    private ConsulRegistryService consulRegistryService;
+
+    @Before
+    public void registryTest() {
+        consulRegistryService = new ConsulRegistryService();
+        CommonConfiguration configuration = new CommonConfiguration(null);
+        ConfigurationContextUtil.putIfAbsent(ConfigurationContextUtil.HTTP, configuration);
+        configuration.namesrvAddr = "127.0.0.1:8500";
+        Mockito.when(eventMeshRegisterInfo.getEventMeshClusterName()).thenReturn("eventmesh");
+        Mockito.when(eventMeshRegisterInfo.getEventMeshName()).thenReturn("eventmesh");
+        Mockito.when(eventMeshRegisterInfo.getEndPoint()).thenReturn("127.0.0.1:8500");
+
+        Mockito.when(eventMeshUnRegisterInfo.getEventMeshClusterName()).thenReturn("eventmesh");
+        Mockito.when(eventMeshUnRegisterInfo.getEventMeshName()).thenReturn("eventmesh");
+    }
+
+    @After
+    public void after() {
+        consulRegistryService.shutdown();
+    }
+
+    @Test
+    public void testInit() {
+        consulRegistryService.init();
+        consulRegistryService.start();
+        Assert.assertNotNull(consulRegistryService.getConsulClient());
+    }
+
+    @Test
+    public void testStart() {
+        consulRegistryService.init();
+        consulRegistryService.start();
+        Assert.assertNotNull(consulRegistryService.getConsulClient());
+    }
+
+    @Test
+    public void testShutdown() throws NoSuchFieldException, IllegalAccessException {
+        consulRegistryService.init();
+        consulRegistryService.start();
+        consulRegistryService.shutdown();
+        Assert.assertNull(consulRegistryService.getConsulClient());
+        Class<ConsulRegistryService> consulRegistryServiceClass = ConsulRegistryService.class;
+        Field initStatus = consulRegistryServiceClass.getDeclaredField("INIT_STATUS");
+        initStatus.setAccessible(true);
+        Object initStatusField = initStatus.get(consulRegistryService);
+
+        Field startStatus = consulRegistryServiceClass.getDeclaredField("START_STATUS");
+        startStatus.setAccessible(true);
+        Object startStatusField = startStatus.get(consulRegistryService);
+
+
+        Assert.assertFalse((Boolean.parseBoolean(initStatusField.toString())));
+        Assert.assertFalse((Boolean.parseBoolean(startStatusField.toString())));
+    }
+
+    @Test
+    public void testRegister() {
+        consulRegistryService.init();
+        consulRegistryService.start();
+        consulRegistryService.register(eventMeshRegisterInfo);
+        List<EventMeshDataInfo> eventmesh = consulRegistryService.findEventMeshInfoByCluster("eventmesh");
+        Assert.assertEquals(1,eventmesh.size());
+    }
+
+    @Test
+    public void testUnRegister() {
+        consulRegistryService.init();
+        consulRegistryService.start();
+        consulRegistryService.unRegister(eventMeshUnRegisterInfo);
+        List<EventMeshDataInfo> eventmesh = consulRegistryService.findEventMeshInfoByCluster("eventmesh");
+        Assert.assertEquals(0,eventmesh.size());
+    }
+
+    @Test
+    public void findEventMeshInfoByCluster() {
+        consulRegistryService.init();
+        consulRegistryService.start();
+        consulRegistryService.register(eventMeshRegisterInfo);
+        List<EventMeshDataInfo> eventmesh = consulRegistryService.findEventMeshInfoByCluster("eventmesh");
+        Assert.assertEquals(1,eventmesh.size());
+        consulRegistryService.unRegister(eventMeshUnRegisterInfo);
+    }
+}

From ffc6b8f1589b9a1c20f86c0a1a29d6659d38266b Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Tue, 31 May 2022 13:23:53 +0800
Subject: [PATCH 02/17] add consul registry implement

---
 settings.gradle | 1 +
 1 file changed, 1 insertion(+)

diff --git a/settings.gradle b/settings.gradle
index 3819bdedf2..f205edfc0a 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -30,6 +30,7 @@ include 'eventmesh-security-plugin:eventmesh-security-api'
 include 'eventmesh-security-plugin:eventmesh-security-acl'
 include 'eventmesh-registry-plugin:eventmesh-registry-api'
 include 'eventmesh-registry-plugin:eventmesh-registry-nacos'
+include 'eventmesh-registry-plugin:eventmesh-registry-consul'
 include 'eventmesh-admin'
 include 'eventmesh-protocol-plugin'
 include 'eventmesh-protocol-plugin:eventmesh-protocol-api'

From c2f90e760b680e07e52dfd0e62badf724c9f8bee Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Mon, 6 Jun 2022 10:15:23 +0800
Subject: [PATCH 03/17] add consul licence

---
 .../licenses/LICENSE-consul-api.txt           | 201 ++++++++++++++++++
 1 file changed, 201 insertions(+)
 create mode 100644 tools/third-party-licenses/licenses/LICENSE-consul-api.txt

diff --git a/tools/third-party-licenses/licenses/LICENSE-consul-api.txt b/tools/third-party-licenses/licenses/LICENSE-consul-api.txt
new file mode 100644
index 0000000000..5c304d1a4a
--- /dev/null
+++ b/tools/third-party-licenses/licenses/LICENSE-consul-api.txt
@@ -0,0 +1,201 @@
+Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "{}"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright {yyyy} {name of copyright owner}
+
+   Licensed 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.

From ce1f9a369957205bef10bb8f59b29d8ac8504e2f Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Mon, 6 Jun 2022 10:15:28 +0800
Subject: [PATCH 04/17] add consul

---
 tools/dependency-check/known-dependencies.txt | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/tools/dependency-check/known-dependencies.txt b/tools/dependency-check/known-dependencies.txt
index 7179efd38c..0a23324b7c 100644
--- a/tools/dependency-check/known-dependencies.txt
+++ b/tools/dependency-check/known-dependencies.txt
@@ -131,4 +131,5 @@ snakeyaml-1.23.jar
 snakeyaml-1.30.jar
 bcpkix-jdk15on-1.69.jar
 bcprov-jdk15on-1.69.jar
-bcutil-jdk15on-1.69.jar
\ No newline at end of file
+bcutil-jdk15on-1.69.jar
+consul-api-1.4.5.jar
\ No newline at end of file

From 9a80a7786ee936def36449dc00c746df047a17a3 Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Mon, 6 Jun 2022 10:15:43 +0800
Subject: [PATCH 05/17] style fix lint

---
 .../consul/service/ConsulRegistryService.java | 44 +++++++++++--------
 1 file changed, 26 insertions(+), 18 deletions(-)

diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
index 3deea5cda5..d2b1faafa0 100644
--- a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
+++ b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
@@ -17,12 +17,12 @@
 
 package org.apache.eventmesh.registry.consul.service;
 
-import com.ecwid.consul.v1.ConsulClient;
-import com.ecwid.consul.v1.ConsulRawClient;
-import com.ecwid.consul.v1.agent.model.NewService;
-import com.ecwid.consul.v1.agent.model.Service;
-import com.ecwid.consul.v1.health.HealthServicesRequest;
-import org.apache.commons.lang3.StringUtils;
+import static org.apache.eventmesh.registry.consul.constant.ConsulConstant.DEFAULT_CONNECTION_TIMEOUT;
+import static org.apache.eventmesh.registry.consul.constant.ConsulConstant.DEFAULT_MAX_CONNECTIONS;
+import static org.apache.eventmesh.registry.consul.constant.ConsulConstant.DEFAULT_MAX_PER_ROUTE_CONNECTIONS;
+import static org.apache.eventmesh.registry.consul.constant.ConsulConstant.DEFAULT_READ_TIMEOUT;
+import static org.apache.eventmesh.registry.consul.constant.ConsulConstant.IP_PORT_SEPARATOR;
+
 import org.apache.eventmesh.api.exception.RegistryException;
 import org.apache.eventmesh.api.registry.RegistryService;
 import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo;
@@ -30,12 +30,12 @@
 import org.apache.eventmesh.api.registry.dto.EventMeshUnRegisterInfo;
 import org.apache.eventmesh.common.config.CommonConfiguration;
 import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
+
+import org.apache.commons.lang3.StringUtils;
 import org.apache.http.client.HttpClient;
 import org.apache.http.client.config.RequestConfig;
 import org.apache.http.impl.client.HttpClientBuilder;
 import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -43,7 +43,14 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import static org.apache.eventmesh.registry.consul.constant.ConsulConstant.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.ecwid.consul.v1.ConsulClient;
+import com.ecwid.consul.v1.ConsulRawClient;
+import com.ecwid.consul.v1.agent.model.NewService;
+import com.ecwid.consul.v1.agent.model.Service;
+import com.ecwid.consul.v1.health.HealthServicesRequest;
 
 public class ConsulRegistryService implements RegistryService {
 
@@ -129,13 +136,14 @@ public List<EventMeshDataInfo> findEventMeshInfoByCluster(String clusterName) th
         List<EventMeshDataInfo> eventMeshDataInfos = new ArrayList<>();
         agentServices.forEach((k, v) -> {
             String[] split = v.getId().split("-");
-            eventMeshDataInfos.add(new EventMeshDataInfo(split[0], split[1], v.getAddress() + ":" + v.getPort(),0));
+            eventMeshDataInfos.add(new EventMeshDataInfo(split[0], split[1], v.getAddress() + ":" + v.getPort(), 0));
         });
         return eventMeshDataInfos;
     }
 
     @Override
-    public Map<String, Map<String, Integer>> findEventMeshClientDistributionData(String clusterName, String group, String purpose) throws RegistryException {
+    public Map<String, Map<String, Integer>> findEventMeshClientDistributionData(String clusterName, String group, String purpose)
+        throws RegistryException {
         return Collections.emptyMap();
     }
 
@@ -146,15 +154,15 @@ private HttpClient getHttpClient() {
         connectionManager.setDefaultMaxPerRoute(DEFAULT_MAX_PER_ROUTE_CONNECTIONS);
 
         RequestConfig requestConfig = RequestConfig.custom().
-                setConnectTimeout(DEFAULT_CONNECTION_TIMEOUT).
-                setConnectionRequestTimeout(DEFAULT_CONNECTION_TIMEOUT).
-                setSocketTimeout(DEFAULT_READ_TIMEOUT).
-                build();
+            setConnectTimeout(DEFAULT_CONNECTION_TIMEOUT).
+            setConnectionRequestTimeout(DEFAULT_CONNECTION_TIMEOUT).
+            setSocketTimeout(DEFAULT_READ_TIMEOUT).
+            build();
 
         HttpClientBuilder httpClientBuilder = HttpClientBuilder.create().
-                setConnectionManager(connectionManager).
-                setDefaultRequestConfig(requestConfig).
-                useSystemProperties();
+            setConnectionManager(connectionManager).
+            setDefaultRequestConfig(requestConfig).
+            useSystemProperties();
 
         return httpClientBuilder.build();
     }

From 6dc84c225874efad09bf739925ed60e369480790 Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Mon, 6 Jun 2022 10:38:31 +0800
Subject: [PATCH 06/17] style fix lint

---
 .../third-party-licenses/licenses/LICENSE-consul-api.txt  | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/tools/third-party-licenses/licenses/LICENSE-consul-api.txt b/tools/third-party-licenses/licenses/LICENSE-consul-api.txt
index 5c304d1a4a..5a5a8ce54c 100644
--- a/tools/third-party-licenses/licenses/LICENSE-consul-api.txt
+++ b/tools/third-party-licenses/licenses/LICENSE-consul-api.txt
@@ -1,4 +1,4 @@
-Apache License
+                                 Apache License
                            Version 2.0, January 2004
                         http://www.apache.org/licenses/
 
@@ -178,9 +178,9 @@ Apache License
    APPENDIX: How to apply the Apache License to your work.
 
       To apply the Apache License to your work, attach the following
-      boilerplate notice, with the fields enclosed by brackets "{}"
+      boilerplate notice, with the fields enclosed by brackets "[]"
       replaced with your own identifying information. (Don't include
-      the brackets!)  The text should be enclosed in the appropriate
+      the brackets!) The text should be enclosed in the appropriate
       comment syntax for the file format. We also recommend that a
       file or class name and description of purpose be included on the
       same "printed page" as the copyright notice for easier
@@ -198,4 +198,4 @@ Apache License
    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.
+   limitations under the License.
\ No newline at end of file

From a6103d6b87f8be7e72686aaa4b7eb1e7545d4630 Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Mon, 6 Jun 2022 10:44:13 +0800
Subject: [PATCH 07/17] style fix lint

---
 .../consul/service/ConsulRegistryService.java | 21 ++++++++++---------
 1 file changed, 11 insertions(+), 10 deletions(-)

diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
index d2b1faafa0..cf499a28ea 100644
--- a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
+++ b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
@@ -153,16 +153,17 @@ private HttpClient getHttpClient() {
         connectionManager.setMaxTotal(DEFAULT_MAX_CONNECTIONS);
         connectionManager.setDefaultMaxPerRoute(DEFAULT_MAX_PER_ROUTE_CONNECTIONS);
 
-        RequestConfig requestConfig = RequestConfig.custom().
-            setConnectTimeout(DEFAULT_CONNECTION_TIMEOUT).
-            setConnectionRequestTimeout(DEFAULT_CONNECTION_TIMEOUT).
-            setSocketTimeout(DEFAULT_READ_TIMEOUT).
-            build();
-
-        HttpClientBuilder httpClientBuilder = HttpClientBuilder.create().
-            setConnectionManager(connectionManager).
-            setDefaultRequestConfig(requestConfig).
-            useSystemProperties();
+        RequestConfig requestConfig = RequestConfig.custom()
+            .setConnectTimeout(DEFAULT_CONNECTION_TIMEOUT)
+            .setConnectionRequestTimeout(DEFAULT_CONNECTION_TIMEOUT)
+            .setSocketTimeout(DEFAULT_READ_TIMEOUT)
+            .build();
+
+        HttpClientBuilder httpClientBuilder = HttpClientBuilder
+            .create()
+            .setConnectionManager(connectionManager)
+            .setDefaultRequestConfig(requestConfig)
+            .useSystemProperties();
 
         return httpClientBuilder.build();
     }

From 41b5c523c7a03e2a6f5b3b36f4a1523c5a972d5d Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Mon, 6 Jun 2022 11:01:07 +0800
Subject: [PATCH 08/17] style fix lint

---
 .../src/test/java/ConsulRegistryServiceTest.java    | 13 +++++++------
 1 file changed, 7 insertions(+), 6 deletions(-)

diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/src/test/java/ConsulRegistryServiceTest.java b/eventmesh-registry-plugin/eventmesh-registry-consul/src/test/java/ConsulRegistryServiceTest.java
index a8c4c78391..fd3a77edff 100644
--- a/eventmesh-registry-plugin/eventmesh-registry-consul/src/test/java/ConsulRegistryServiceTest.java
+++ b/eventmesh-registry-plugin/eventmesh-registry-consul/src/test/java/ConsulRegistryServiceTest.java
@@ -21,6 +21,10 @@
 import org.apache.eventmesh.common.config.CommonConfiguration;
 import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
 import org.apache.eventmesh.registry.consul.service.ConsulRegistryService;
+
+import java.lang.reflect.Field;
+import java.util.List;
+
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -30,9 +34,6 @@
 import org.mockito.Mockito;
 import org.mockito.junit.MockitoJUnitRunner;
 
-import java.lang.reflect.Field;
-import java.util.List;
-
 /**
  * @author huyuanxin
  */
@@ -105,7 +106,7 @@ public void testRegister() {
         consulRegistryService.start();
         consulRegistryService.register(eventMeshRegisterInfo);
         List<EventMeshDataInfo> eventmesh = consulRegistryService.findEventMeshInfoByCluster("eventmesh");
-        Assert.assertEquals(1,eventmesh.size());
+        Assert.assertEquals(1, eventmesh.size());
     }
 
     @Test
@@ -114,7 +115,7 @@ public void testUnRegister() {
         consulRegistryService.start();
         consulRegistryService.unRegister(eventMeshUnRegisterInfo);
         List<EventMeshDataInfo> eventmesh = consulRegistryService.findEventMeshInfoByCluster("eventmesh");
-        Assert.assertEquals(0,eventmesh.size());
+        Assert.assertEquals(0, eventmesh.size());
     }
 
     @Test
@@ -123,7 +124,7 @@ public void findEventMeshInfoByCluster() {
         consulRegistryService.start();
         consulRegistryService.register(eventMeshRegisterInfo);
         List<EventMeshDataInfo> eventmesh = consulRegistryService.findEventMeshInfoByCluster("eventmesh");
-        Assert.assertEquals(1,eventmesh.size());
+        Assert.assertEquals(1, eventmesh.size());
         consulRegistryService.unRegister(eventMeshUnRegisterInfo);
     }
 }

From a1ca093bd9c47aa17329805157dc152d1bd0b352 Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Mon, 6 Jun 2022 11:33:05 +0800
Subject: [PATCH 09/17] style fix lint

---
 .../src/test/java/ConsulRegistryServiceTest.java               | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/src/test/java/ConsulRegistryServiceTest.java b/eventmesh-registry-plugin/eventmesh-registry-consul/src/test/java/ConsulRegistryServiceTest.java
index fd3a77edff..f0d9e4c02b 100644
--- a/eventmesh-registry-plugin/eventmesh-registry-consul/src/test/java/ConsulRegistryServiceTest.java
+++ b/eventmesh-registry-plugin/eventmesh-registry-consul/src/test/java/ConsulRegistryServiceTest.java
@@ -34,9 +34,6 @@
 import org.mockito.Mockito;
 import org.mockito.junit.MockitoJUnitRunner;
 
-/**
- * @author huyuanxin
- */
 @RunWith(MockitoJUnitRunner.class)
 public class ConsulRegistryServiceTest {
 

From 3e1156b9d1de5f20e4b3f015d6b8ca02d9832c85 Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Thu, 16 Jun 2022 10:46:17 +0800
Subject: [PATCH 10/17] create spi file

---
 ...apache.eventmesh.api.registry.RegistryService | 16 ++++++++++++++++
 1 file changed, 16 insertions(+)
 create mode 100644 eventmesh-registry-plugin/eventmesh-registry-consul/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.registry.RegistryService

diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.registry.RegistryService b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.registry.RegistryService
new file mode 100644
index 0000000000..d18efe974c
--- /dev/null
+++ b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.registry.RegistryService
@@ -0,0 +1,16 @@
+# 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.
+
+consul=org.apache.eventmesh.registry.consul.service.ConsulRegistryService
\ No newline at end of file

From c86b3be98961540d128d8f7ff76b400ff82c1efe Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Thu, 16 Jun 2022 10:55:15 +0800
Subject: [PATCH 11/17] remove unused file

---
 .../consul/constant/ConsulConstant.java       | 45 -------------------
 1 file changed, 45 deletions(-)
 delete mode 100644 eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/constant/ConsulConstant.java

diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/constant/ConsulConstant.java b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/constant/ConsulConstant.java
deleted file mode 100644
index bd8aa876c6..0000000000
--- a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/constant/ConsulConstant.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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.registry.consul.constant;
-
-public class ConsulConstant {
-
-    public static final int DEFAULT_MAX_CONNECTIONS = 1000;
-
-    public static final int DEFAULT_MAX_PER_ROUTE_CONNECTIONS = 500;
-
-    /**
-     * 3 sec
-     */
-    public static final int DEFAULT_CONNECTION_TIMEOUT = 3000;
-
-    /**
-     * 2 sec
-     * 0 minutes for read timeout due to blocking queries timeout
-     * <a href="https://www.consul.io/api/index.html#blocking-queries">...</a>
-     */
-    public static final int DEFAULT_READ_TIMEOUT = 2000;
-
-    public static final String SERVICE_ADDR = "serverAddr";
-
-    public static final String IP_PORT_SEPARATOR = ":";
-
-    public static final String DEFAULT_GROUP = "DEFAULT_GROUP";
-
-
-}

From 42bd6d18e0c0dfe060a94d393c7249237cf29869 Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Thu, 16 Jun 2022 10:55:33 +0800
Subject: [PATCH 12/17] resolve conversation

---
 .../consul/service/ConsulRegistryService.java | 37 ++-----------------
 1 file changed, 4 insertions(+), 33 deletions(-)

diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
index cf499a28ea..c981c84bac 100644
--- a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
+++ b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
@@ -17,12 +17,6 @@
 
 package org.apache.eventmesh.registry.consul.service;
 
-import static org.apache.eventmesh.registry.consul.constant.ConsulConstant.DEFAULT_CONNECTION_TIMEOUT;
-import static org.apache.eventmesh.registry.consul.constant.ConsulConstant.DEFAULT_MAX_CONNECTIONS;
-import static org.apache.eventmesh.registry.consul.constant.ConsulConstant.DEFAULT_MAX_PER_ROUTE_CONNECTIONS;
-import static org.apache.eventmesh.registry.consul.constant.ConsulConstant.DEFAULT_READ_TIMEOUT;
-import static org.apache.eventmesh.registry.consul.constant.ConsulConstant.IP_PORT_SEPARATOR;
-
 import org.apache.eventmesh.api.exception.RegistryException;
 import org.apache.eventmesh.api.registry.RegistryService;
 import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo;
@@ -32,10 +26,6 @@
 import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
 
 import org.apache.commons.lang3.StringUtils;
-import org.apache.http.client.HttpClient;
-import org.apache.http.client.config.RequestConfig;
-import org.apache.http.impl.client.HttpClientBuilder;
-import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -54,6 +44,8 @@
 
 public class ConsulRegistryService implements RegistryService {
 
+    public static final String IP_PORT_SEPARATOR = ":";
+
     private static final Logger logger = LoggerFactory.getLogger(ConsulRegistryService.class);
 
     private static final AtomicBoolean INIT_STATUS = new AtomicBoolean(false);
@@ -90,7 +82,7 @@ public void init() throws RegistryException {
 
     @Override
     public void start() throws RegistryException {
-        consulClient = new ConsulClient(new ConsulRawClient(consulHost, Integer.parseInt(consulPort), getHttpClient()));
+        consulClient = new ConsulClient(new ConsulRawClient(consulHost, Integer.parseInt(consulPort)));
     }
 
     @Override
@@ -107,7 +99,7 @@ public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo) throws Regi
             NewService service = new NewService();
             service.setPort(Integer.parseInt(ipPort[1]));
             service.setAddress(ipPort[0]);
-            service.setName(eventMeshRegisterInfo.getEventMeshClusterName());
+            service.setName(eventMeshRegisterInfo.getEventMeshName());
             service.setId(eventMeshRegisterInfo.getEventMeshClusterName() + "-" + eventMeshRegisterInfo.getEventMeshName());
             consulClient.agentServiceRegister(service);
         } catch (Exception e) {
@@ -147,27 +139,6 @@ public Map<String, Map<String, Integer>> findEventMeshClientDistributionData(Str
         return Collections.emptyMap();
     }
 
-
-    private HttpClient getHttpClient() {
-        PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
-        connectionManager.setMaxTotal(DEFAULT_MAX_CONNECTIONS);
-        connectionManager.setDefaultMaxPerRoute(DEFAULT_MAX_PER_ROUTE_CONNECTIONS);
-
-        RequestConfig requestConfig = RequestConfig.custom()
-            .setConnectTimeout(DEFAULT_CONNECTION_TIMEOUT)
-            .setConnectionRequestTimeout(DEFAULT_CONNECTION_TIMEOUT)
-            .setSocketTimeout(DEFAULT_READ_TIMEOUT)
-            .build();
-
-        HttpClientBuilder httpClientBuilder = HttpClientBuilder
-            .create()
-            .setConnectionManager(connectionManager)
-            .setDefaultRequestConfig(requestConfig)
-            .useSystemProperties();
-
-        return httpClientBuilder.build();
-    }
-
     public ConsulClient getConsulClient() {
         return consulClient;
     }

From 3f916772ee629af526824ea2f441ae2d6bc8199a Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Tue, 28 Jun 2022 11:36:52 +0800
Subject: [PATCH 13/17] add token support

---
 .../consul/service/ConsulRegistryService.java |  9 +-
 .../consul/service/HeatBeatScheduler.java     | 89 +++++++++++++++++++
 2 files changed, 95 insertions(+), 3 deletions(-)
 create mode 100644 eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/HeatBeatScheduler.java

diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
index c981c84bac..e68664c42d 100644
--- a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
+++ b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
@@ -58,6 +58,8 @@ public class ConsulRegistryService implements RegistryService {
 
     private ConsulClient consulClient;
 
+    private String token;
+
     @Override
     public void init() throws RegistryException {
         if (INIT_STATUS.compareAndSet(false, true)) {
@@ -101,7 +103,7 @@ public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo) throws Regi
             service.setAddress(ipPort[0]);
             service.setName(eventMeshRegisterInfo.getEventMeshName());
             service.setId(eventMeshRegisterInfo.getEventMeshClusterName() + "-" + eventMeshRegisterInfo.getEventMeshName());
-            consulClient.agentServiceRegister(service);
+            consulClient.agentServiceRegister(service, token);
         } catch (Exception e) {
             throw new RegistryException(e.getMessage());
         }
@@ -112,7 +114,8 @@ public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo) throws Regi
     @Override
     public boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo) throws RegistryException {
         try {
-            consulClient.agentServiceDeregister(eventMeshUnRegisterInfo.getEventMeshClusterName() + "-" + eventMeshUnRegisterInfo.getEventMeshName());
+            consulClient.agentServiceDeregister(eventMeshUnRegisterInfo.getEventMeshClusterName() + "-" + eventMeshUnRegisterInfo.getEventMeshName(),
+                token);
         } catch (Exception e) {
             throw new RegistryException(e.getMessage());
         }
@@ -123,7 +126,7 @@ public boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo) throw
     @Override
     public List<EventMeshDataInfo> findEventMeshInfoByCluster(String clusterName) throws RegistryException {
         Map<String, Service> agentServices = consulClient.getAgentServices().getValue();
-        HealthServicesRequest request = HealthServicesRequest.newBuilder().setPassing(true).build();
+        HealthServicesRequest request = HealthServicesRequest.newBuilder().setPassing(true).setToken(token).build();
         consulClient.getHealthServices(clusterName, request);
         List<EventMeshDataInfo> eventMeshDataInfos = new ArrayList<>();
         agentServices.forEach((k, v) -> {
diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/HeatBeatScheduler.java b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/HeatBeatScheduler.java
new file mode 100644
index 0000000000..a61c751e07
--- /dev/null
+++ b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/HeatBeatScheduler.java
@@ -0,0 +1,89 @@
+package org.apache.eventmesh.registry.consul.service;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import com.ecwid.consul.v1.ConsulClient;
+import com.ecwid.consul.v1.agent.model.NewService;
+
+/**
+ * @author huyuanxin
+ */
+public class HeatBeatScheduler {
+
+    private final ConsulClient consulClient;
+
+    private final ConcurrentHashMap<String, NewService> heartBeatMap = new ConcurrentHashMap<>();
+
+    private final ScheduledExecutorService heartbeatServiceExecutor = new ScheduledThreadPoolExecutor(
+        Runtime.getRuntime().availableProcessors(),
+        r -> {
+            Thread thread = new Thread(r);
+            thread.setDaemon(true);
+            thread.setName("ConsulHeartbeatService");
+            return thread;
+        }
+    );
+
+    public HeatBeatScheduler(ConsulClient consulClient) {
+        this.consulClient = consulClient;
+    }
+
+    /**
+     * 开始服务的心跳
+     *
+     * @param newService 服务
+     * @param aclToken   token
+     */
+    protected void startHeartBeat(NewService newService, String aclToken) {
+        heartbeatServiceExecutor.execute(new HeartBeat(newService, aclToken));
+        heartBeatMap.put(newService.getName(), newService);
+    }
+
+    /**
+     * 停止服务的心跳
+     *
+     * @param newService 服务
+     */
+    private void stopHeartBeat(NewService newService) {
+        heartBeatMap.remove(newService.getName());
+    }
+
+    class HeartBeat implements Runnable {
+
+        private static final String CHECK_ID_PREFIX = "service:";
+
+        private String checkId;
+
+        private final String aclToken;
+
+        private final NewService instance;
+
+        public HeartBeat(NewService instance, String aclToken) {
+            this.instance = instance;
+            this.checkId = instance.getId();
+            this.aclToken = aclToken;
+            if (!checkId.startsWith(CHECK_ID_PREFIX)) {
+                checkId = CHECK_ID_PREFIX + checkId;
+            }
+        }
+
+        @Override
+        public void run() {
+            try {
+                if (aclToken != null) {
+                    consulClient.agentCheckPass(checkId, aclToken);
+                    return;
+                }
+                if (heartBeatMap.contains(instance)) {
+                    consulClient.agentCheckPass(checkId);
+                }
+            } finally {
+                heartbeatServiceExecutor.schedule(this, 3000, TimeUnit.SECONDS);
+            }
+        }
+
+    }
+}

From 21218ad60a5d56f778701f724db6f3ff515a1f25 Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Tue, 19 Jul 2022 10:18:14 +0800
Subject: [PATCH 14/17] feat: add new impl

---
 .../consul/service/ConsulRegistryService.java | 23 ++++++++++++++++---
 1 file changed, 20 insertions(+), 3 deletions(-)

diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
index e68664c42d..36b3c2814c 100644
--- a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
+++ b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/ConsulRegistryService.java
@@ -41,6 +41,7 @@
 import com.ecwid.consul.v1.agent.model.NewService;
 import com.ecwid.consul.v1.agent.model.Service;
 import com.ecwid.consul.v1.health.HealthServicesRequest;
+import com.ecwid.consul.v1.health.model.HealthService;
 
 public class ConsulRegistryService implements RegistryService {
 
@@ -125,13 +126,24 @@ public boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo) throw
 
     @Override
     public List<EventMeshDataInfo> findEventMeshInfoByCluster(String clusterName) throws RegistryException {
-        Map<String, Service> agentServices = consulClient.getAgentServices().getValue();
         HealthServicesRequest request = HealthServicesRequest.newBuilder().setPassing(true).setToken(token).build();
-        consulClient.getHealthServices(clusterName, request);
+        List<HealthService> healthServices = consulClient.getHealthServices(clusterName, request).getValue();
+        List<EventMeshDataInfo> eventMeshDataInfos = new ArrayList<>();
+        healthServices.forEach(healthService -> {
+            HealthService.Service service = healthService.getService();
+            String[] split = service.getId().split("-");
+            eventMeshDataInfos.add(new EventMeshDataInfo(split[0], split[1], service.getAddress() + ":" + service.getPort(), 0, service.getMeta()));
+        });
+        return eventMeshDataInfos;
+    }
+
+    @Override
+    public List<EventMeshDataInfo> findAllEventMeshInfo() throws RegistryException {
+        Map<String, Service> agentServices = consulClient.getAgentServices().getValue();
         List<EventMeshDataInfo> eventMeshDataInfos = new ArrayList<>();
         agentServices.forEach((k, v) -> {
             String[] split = v.getId().split("-");
-            eventMeshDataInfos.add(new EventMeshDataInfo(split[0], split[1], v.getAddress() + ":" + v.getPort(), 0));
+            eventMeshDataInfos.add(new EventMeshDataInfo(split[0], split[1], v.getAddress() + ":" + v.getPort(), 0, v.getMeta()));
         });
         return eventMeshDataInfos;
     }
@@ -142,6 +154,11 @@ public Map<String, Map<String, Integer>> findEventMeshClientDistributionData(Str
         return Collections.emptyMap();
     }
 
+    @Override
+    public void registerMetadata(Map<String, String> metadataMap) {
+
+    }
+
     public ConsulClient getConsulClient() {
         return consulClient;
     }

From d5dcc56255a0fca80055adf2c56d74a41100273a Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Tue, 19 Jul 2022 10:45:02 +0800
Subject: [PATCH 15/17] feat: add licence header

---
 .../consul/service/HeatBeatScheduler.java       | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/HeatBeatScheduler.java b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/HeatBeatScheduler.java
index a61c751e07..27dfbabe65 100644
--- a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/HeatBeatScheduler.java
+++ b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/HeatBeatScheduler.java
@@ -1,3 +1,20 @@
+/*
+ * 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.registry.consul.service;
 
 import java.util.concurrent.ConcurrentHashMap;

From f7d71306bf59780eb28bae68debd0c2538b5b7e5 Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Tue, 19 Jul 2022 10:48:02 +0800
Subject: [PATCH 16/17] refactor: move to java

---
 .../licenses/{ => java}/LICENSE-consul-api.txt                    | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 rename tools/third-party-licenses/licenses/{ => java}/LICENSE-consul-api.txt (100%)

diff --git a/tools/third-party-licenses/licenses/LICENSE-consul-api.txt b/tools/third-party-licenses/licenses/java/LICENSE-consul-api.txt
similarity index 100%
rename from tools/third-party-licenses/licenses/LICENSE-consul-api.txt
rename to tools/third-party-licenses/licenses/java/LICENSE-consul-api.txt

From 9340d67b6fbfdf1983314d986e52fce9bb5923e7 Mon Sep 17 00:00:00 2001
From: huyuanxin <huyuanxin1999@outlook.com>
Date: Wed, 20 Jul 2022 09:33:04 +0800
Subject: [PATCH 17/17] style: replace Chinese to English

---
 .../registry/consul/service/HeatBeatScheduler.java    | 11 ++++-------
 1 file changed, 4 insertions(+), 7 deletions(-)

diff --git a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/HeatBeatScheduler.java b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/HeatBeatScheduler.java
index 27dfbabe65..a62db12fd2 100644
--- a/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/HeatBeatScheduler.java
+++ b/eventmesh-registry-plugin/eventmesh-registry-consul/src/main/java/org/apache/eventmesh/registry/consul/service/HeatBeatScheduler.java
@@ -25,9 +25,6 @@
 import com.ecwid.consul.v1.ConsulClient;
 import com.ecwid.consul.v1.agent.model.NewService;
 
-/**
- * @author huyuanxin
- */
 public class HeatBeatScheduler {
 
     private final ConsulClient consulClient;
@@ -49,9 +46,9 @@ public HeatBeatScheduler(ConsulClient consulClient) {
     }
 
     /**
-     * 开始服务的心跳
+     * start service heartbeat
      *
-     * @param newService 服务
+     * @param newService service
      * @param aclToken   token
      */
     protected void startHeartBeat(NewService newService, String aclToken) {
@@ -60,9 +57,9 @@ protected void startHeartBeat(NewService newService, String aclToken) {
     }
 
     /**
-     * 停止服务的心跳
+     * stop service heartbeat
      *
-     * @param newService 服务
+     * @param newService service
      */
     private void stopHeartBeat(NewService newService) {
         heartBeatMap.remove(newService.getName());