This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/develop by this push:
new a62bf65 [ISSUE #488]Support registry adaptor plugin (#489)
a62bf65 is described below
commit a62bf6567d8e24406cad4eb65f4a0362f901bba7
Author: lrhkobe <[email protected]>
AuthorDate: Fri Aug 13 16:57:34 2021 +0800
[ISSUE #488]Support registry adaptor plugin (#489)
* modify:optimize flow control in downstreaming msg
* modify:optimize stategy of selecting session in downstream msg
* modify:optimize msg downstream,msg store in session
* modify:fix bug:not a @Sharable handler
* modify:downstream broadcast msg asynchronously
* modify:remove unneccessary interface in eventmesh-connector-api
* modify:fix conflict
* modify:add license in EventMeshAction
* modify:fix ack problem
* modify:fix exception handle when exception occured in
EventMeshTcpMessageDispatcher
* modify:fix log print
* modify:add eventmesh-registry-plugin module,add eventmesh-registry-api
design
* modify:add default empty registry plugin impl
* modify:add registry in eventmesh-runtime for rebalance or recommend
eventmesh to client
* modify:adjust security plugin config
* modify:add license,delete chinese characters
* modify:reactor eventmesh-registry-namesrv module name,add rebalance of
pub client
close #488
---
.../common/config/CommonConfiguration.java | 14 ++
.../src/test/resources/configuration.properties | 3 +-
.../build.gradle | 33 ++--
.../eventmesh-registry-api/build.gradle | 16 +-
.../eventmesh/api/exception/RegistryException.java | 22 +--
.../eventmesh/api/registry/RegistryService.java | 43 +++++
.../api/registry/dto/EventMeshDataInfo.java | 63 +++++++
.../api/registry/dto/EventMeshRegisterInfo.java | 58 ++++++
.../api/registry/dto/EventMeshUnRegisterInfo.java | 32 ++--
.../build.gradle | 17 +-
.../namesrv/RegistryServiceNamesrvImpl.java | 63 +++++++
...g.apache.eventmesh.api.registry.RegistryService | 11 +-
eventmesh-runtime/build.gradle | 5 +
eventmesh-runtime/conf/eventmesh.properties | 8 +
.../admin/controller/ClientManageController.java | 12 +-
.../handler/QueryRecommendEventMeshHandler.java | 91 +++++++++
.../eventmesh/runtime/boot/EventMeshServer.java | 17 +-
.../eventmesh/runtime/boot/EventMeshTCPServer.java | 93 ++++++++-
.../configuration/EventMeshTCPConfiguration.java | 12 +-
.../runtime/constants/EventMeshConstants.java | 2 +
.../tcp/client/EventMeshTcpMessageDispatcher.java | 15 +-
.../client/group/ClientSessionGroupMapping.java | 45 +++++
.../rebalance/EventMeshRebalanceService.java | 72 +++++++
.../rebalance/EventMeshRebalanceStrategy.java | 17 +-
.../client/rebalance/EventmeshRebalanceImpl.java | 208 +++++++++++++++++++++
.../client/recommend/EventMeshRecommendImpl.java | 195 +++++++++++++++++++
.../recommend/EventMeshRecommendStrategy.java | 20 +-
.../protocol/tcp/client/task/RecommendTask.java | 101 ++++++++++
.../runtime/metrics/tcp/EventMeshTcpMonitor.java | 1 +
.../eventmesh/runtime/registry/Registry.java | 66 +++++++
.../eventmesh/runtime/util/EventMeshUtil.java | 10 +
settings.gradle | 2 +
32 files changed, 1244 insertions(+), 123 deletions(-)
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
index b04109c..2c6b37d 100644
---
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,12 +31,14 @@ public class CommonConfiguration {
public String eventMeshConnectorPluginType = "rocketmq";
public String eventMeshSecurityPluginType = "security";
public int eventMeshPrometheusPort = 19090;
+ public String eventMeshRegistryPluginType = "namesrv";
public String namesrvAddr = "";
public Integer eventMeshRegisterIntervalInMills = 10 * 1000;
public Integer eventMeshFetchRegistryAddrInterval = 10 * 1000;
public String eventMeshServerIp = null;
public boolean eventMeshServerSecurityEnable = false;
+ public boolean eventMeshServerRegistryEnable = false;
protected ConfigurationWrapper configurationWrapper;
public CommonConfiguration(ConfigurationWrapper configurationWrapper) {
@@ -86,6 +88,14 @@ public class CommonConfiguration {
eventMeshSecurityPluginType =
configurationWrapper.getProp(ConfKeys.KEYS_ENENTMESH_SECURITY_PLUGIN_TYPE);
Preconditions.checkState(StringUtils.isNotEmpty(eventMeshSecurityPluginType),
String.format("%s error", ConfKeys.KEYS_ENENTMESH_SECURITY_PLUGIN_TYPE));
+
+ String eventMeshServerRegistryEnableStr =
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_REGISTRY_ENABLED);
+ if (StringUtils.isNotBlank(eventMeshServerRegistryEnableStr)) {
+ eventMeshServerRegistryEnable =
Boolean.valueOf(StringUtils.deleteWhitespace(eventMeshServerRegistryEnableStr));
+ }
+
+ eventMeshRegistryPluginType =
configurationWrapper.getProp(ConfKeys.KEYS_ENENTMESH_REGISTRY_PLUGIN_TYPE);
+
Preconditions.checkState(StringUtils.isNotEmpty(eventMeshRegistryPluginType),
String.format("%s error", ConfKeys.KEYS_ENENTMESH_REGISTRY_PLUGIN_TYPE));
}
}
@@ -113,5 +123,9 @@ public class CommonConfiguration {
public static String KEYS_ENENTMESH_SECURITY_PLUGIN_TYPE =
"eventMesh.security.plugin.type";
public static String KEY_EVENTMESH_METRICS_PROMETHEUS_PORT =
"eventMesh.metrics.prometheus.port";
+
+ public static String KEYS_EVENTMESH_REGISTRY_ENABLED =
"eventMesh.server.registry.enabled";
+
+ public static String KEYS_ENENTMESH_REGISTRY_PLUGIN_TYPE =
"eventMesh.registry.plugin.type";
}
}
\ No newline at end of file
diff --git a/eventmesh-common/src/test/resources/configuration.properties
b/eventmesh-common/src/test/resources/configuration.properties
index c9425d7..b8bf913 100644
--- a/eventmesh-common/src/test/resources/configuration.properties
+++ b/eventmesh-common/src/test/resources/configuration.properties
@@ -22,4 +22,5 @@ eventMesh.server.cluster=value4
eventMesh.server.name=value5
eventMesh.server.hostIp=value6
eventMesh.connector.plugin.type=rocketmq
-eventMesh.security.plugin.type=security
+eventMesh.security.plugin.type=acl
+eventMesh.registry.plugin.type=namesrv
\ No newline at end of file
diff --git a/settings.gradle b/eventmesh-registry-plugin/build.gradle
similarity index 53%
copy from settings.gradle
copy to eventmesh-registry-plugin/build.gradle
index 145df7e..87e4a9b 100644
--- a/settings.gradle
+++ b/eventmesh-registry-plugin/build.gradle
@@ -15,16 +15,23 @@
* limitations under the License.
*/
-rootProject.name = 'EventMesh'
-String jdkVersion = "${jdk}"
-include 'eventmesh-runtime'
-include 'eventmesh-sdk-java'
-include 'eventmesh-common'
-include 'eventmesh-starter'
-include 'eventmesh-test'
-include 'eventmesh-spi'
-include 'eventmesh-connector-plugin:eventmesh-connector-api'
-include 'eventmesh-connector-plugin:eventmesh-connector-rocketmq'
-include 'eventmesh-security-plugin:eventmesh-security-api'
-include 'eventmesh-security-plugin:eventmesh-security-acl'
-
+task copyRegistryPlugin(dependsOn: ['jar']) {
+ doFirst {
+ new File(projectDir, '../eventmesh-registry-plugin/dist/apps').mkdir()
+ new File(projectDir, '../dist/plugin/registry').mkdirs()
+ }
+ doLast {
+ copy {
+ into('../eventmesh-registry-plugin/dist/apps/')
+ from project.jar.getArchivePath()
+ exclude {
+ "eventmesh-registry-plugin-${version}.jar"
+ "eventmesh-registry-api-${version}.jar"
+ }
+ }
+ copy {
+ into '../dist/plugin/registry'
+ from
"../eventmesh-registry-plugin/dist/apps/eventmesh-registry-rocketmq-namesrv-${version}.jar"
+ }
+ }
+}
\ No newline at end of file
diff --git a/settings.gradle
b/eventmesh-registry-plugin/eventmesh-registry-api/build.gradle
similarity index 63%
copy from settings.gradle
copy to eventmesh-registry-plugin/eventmesh-registry-api/build.gradle
index 145df7e..0d41042 100644
--- a/settings.gradle
+++ b/eventmesh-registry-plugin/eventmesh-registry-api/build.gradle
@@ -15,16 +15,8 @@
* limitations under the License.
*/
-rootProject.name = 'EventMesh'
-String jdkVersion = "${jdk}"
-include 'eventmesh-runtime'
-include 'eventmesh-sdk-java'
-include 'eventmesh-common'
-include 'eventmesh-starter'
-include 'eventmesh-test'
-include 'eventmesh-spi'
-include 'eventmesh-connector-plugin:eventmesh-connector-api'
-include 'eventmesh-connector-plugin:eventmesh-connector-rocketmq'
-include 'eventmesh-security-plugin:eventmesh-security-api'
-include 'eventmesh-security-plugin:eventmesh-security-acl'
+dependencies {
+ api project(":eventmesh-spi")
+ testImplementation project(":eventmesh-spi")
+}
diff --git a/settings.gradle
b/eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/exception/RegistryException.java
similarity index 63%
copy from settings.gradle
copy to
eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/exception/RegistryException.java
index 145df7e..31453ef 100644
--- a/settings.gradle
+++
b/eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/exception/RegistryException.java
@@ -14,17 +14,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.eventmesh.api.exception;
-rootProject.name = 'EventMesh'
-String jdkVersion = "${jdk}"
-include 'eventmesh-runtime'
-include 'eventmesh-sdk-java'
-include 'eventmesh-common'
-include 'eventmesh-starter'
-include 'eventmesh-test'
-include 'eventmesh-spi'
-include 'eventmesh-connector-plugin:eventmesh-connector-api'
-include 'eventmesh-connector-plugin:eventmesh-connector-rocketmq'
-include 'eventmesh-security-plugin:eventmesh-security-api'
-include 'eventmesh-security-plugin:eventmesh-security-acl'
+public class RegistryException extends RuntimeException {
+ public RegistryException(String message) {
+ super(message);
+ }
+
+ public RegistryException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git
a/eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/registry/RegistryService.java
b/eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/registry/RegistryService.java
new file mode 100644
index 0000000..e161b71
--- /dev/null
+++
b/eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/registry/RegistryService.java
@@ -0,0 +1,43 @@
+/*
+ * 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.api.registry;
+
+import org.apache.eventmesh.api.exception.RegistryException;
+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.spi.EventMeshSPI;
+
+import java.util.List;
+import java.util.Map;
+
+@EventMeshSPI(isSingleton = true)
+public interface RegistryService {
+ void init() throws RegistryException;
+
+ void start() throws RegistryException;
+
+ void shutdown() throws RegistryException;
+
+ List<EventMeshDataInfo> findEventMeshInfoByCluster(String clusterName)
throws RegistryException;
+
+ Map<String/*eventMeshName*/, Map<String/*purpose*/, Integer/*num*/>>
findEventMeshClientDistributionData(String clusterName, String group, String
purpose) throws RegistryException;
+
+ boolean register(EventMeshRegisterInfo eventMeshRegisterInfo) throws
RegistryException;
+
+ boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo) throws
RegistryException;
+}
diff --git
a/eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/registry/dto/EventMeshDataInfo.java
b/eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/registry/dto/EventMeshDataInfo.java
new file mode 100644
index 0000000..389d1e4
--- /dev/null
+++
b/eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/registry/dto/EventMeshDataInfo.java
@@ -0,0 +1,63 @@
+/*
+ * 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.api.registry.dto;
+
+public class EventMeshDataInfo {
+ private String eventMeshClusterName;
+ private String eventMeshName;
+ private String endpoint;
+ private long lastUpdateTimestamp;
+
+ public EventMeshDataInfo(String eventMeshClusterName, String
eventMeshName, String endpoint, long lastUpdateTimestamp) {
+ this.eventMeshClusterName = eventMeshClusterName;
+ this.eventMeshName = eventMeshName;
+ this.endpoint = endpoint;
+ this.lastUpdateTimestamp = lastUpdateTimestamp;
+ }
+
+ public String getEventMeshClusterName() {
+ return eventMeshClusterName;
+ }
+
+ public void setEventMeshClusterName(String eventMeshClusterName) {
+ this.eventMeshClusterName = eventMeshClusterName;
+ }
+
+ public String getEventMeshName() {
+ return eventMeshName;
+ }
+
+ public void setEventMeshName(String eventMeshName) {
+ this.eventMeshName = eventMeshName;
+ }
+
+ public String getEndpoint() {
+ return endpoint;
+ }
+
+ public void setEndpoint(String endpoint) {
+ this.endpoint = endpoint;
+ }
+
+ public long getLastUpdateTimestamp() {
+ return lastUpdateTimestamp;
+ }
+
+ public void setLastUpdateTimestamp(long lastUpdateTimestamp) {
+ this.lastUpdateTimestamp = lastUpdateTimestamp;
+ }
+}
diff --git
a/eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/registry/dto/EventMeshRegisterInfo.java
b/eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/registry/dto/EventMeshRegisterInfo.java
new file mode 100644
index 0000000..98eaf5d
--- /dev/null
+++
b/eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/registry/dto/EventMeshRegisterInfo.java
@@ -0,0 +1,58 @@
+/*
+ * 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.api.registry.dto;
+
+import java.util.Map;
+
+public class EventMeshRegisterInfo {
+ private String eventMeshClusterName;
+ private String eventMeshName;
+ private String endPoint;
+ private Map<String, Map<String, Integer>> eventMeshInstanceNumMap;
+
+ public String getEventMeshClusterName() {
+ return eventMeshClusterName;
+ }
+
+ public void setEventMeshClusterName(String eventMeshClusterName) {
+ this.eventMeshClusterName = eventMeshClusterName;
+ }
+
+ public String getEventMeshName() {
+ return eventMeshName;
+ }
+
+ public void setEventMeshName(String eventMeshName) {
+ this.eventMeshName = eventMeshName;
+ }
+
+ public String getEndPoint() {
+ return endPoint;
+ }
+
+ public void setEndPoint(String endPoint) {
+ this.endPoint = endPoint;
+ }
+
+ public Map<String, Map<String, Integer>> getEventMeshInstanceNumMap() {
+ return eventMeshInstanceNumMap;
+ }
+
+ public void setEventMeshInstanceNumMap(Map<String, Map<String, Integer>>
eventMeshInstanceNumMap) {
+ this.eventMeshInstanceNumMap = eventMeshInstanceNumMap;
+ }
+}
diff --git a/settings.gradle
b/eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/registry/dto/EventMeshUnRegisterInfo.java
similarity index 58%
copy from settings.gradle
copy to
eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/registry/dto/EventMeshUnRegisterInfo.java
index 145df7e..06e54a0 100644
--- a/settings.gradle
+++
b/eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/registry/dto/EventMeshUnRegisterInfo.java
@@ -14,17 +14,25 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.eventmesh.api.registry.dto;
-rootProject.name = 'EventMesh'
-String jdkVersion = "${jdk}"
-include 'eventmesh-runtime'
-include 'eventmesh-sdk-java'
-include 'eventmesh-common'
-include 'eventmesh-starter'
-include 'eventmesh-test'
-include 'eventmesh-spi'
-include 'eventmesh-connector-plugin:eventmesh-connector-api'
-include 'eventmesh-connector-plugin:eventmesh-connector-rocketmq'
-include 'eventmesh-security-plugin:eventmesh-security-api'
-include 'eventmesh-security-plugin:eventmesh-security-acl'
+public class EventMeshUnRegisterInfo {
+ private String eventMeshClusterName;
+ private String eventMeshName;
+ public String getEventMeshClusterName() {
+ return eventMeshClusterName;
+ }
+
+ public void setEventMeshClusterName(String eventMeshClusterName) {
+ this.eventMeshClusterName = eventMeshClusterName;
+ }
+
+ public String getEventMeshName() {
+ return eventMeshName;
+ }
+
+ public void setEventMeshName(String eventMeshName) {
+ this.eventMeshName = eventMeshName;
+ }
+}
diff --git a/settings.gradle
b/eventmesh-registry-plugin/eventmesh-registry-rocketmq-namesrv/build.gradle
similarity index 63%
copy from settings.gradle
copy to
eventmesh-registry-plugin/eventmesh-registry-rocketmq-namesrv/build.gradle
index 145df7e..ec1fbb2 100644
--- a/settings.gradle
+++ b/eventmesh-registry-plugin/eventmesh-registry-rocketmq-namesrv/build.gradle
@@ -14,17 +14,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+dependencies {
+ api project(":eventmesh-registry-plugin:eventmesh-registry-api")
-rootProject.name = 'EventMesh'
-String jdkVersion = "${jdk}"
-include 'eventmesh-runtime'
-include 'eventmesh-sdk-java'
-include 'eventmesh-common'
-include 'eventmesh-starter'
-include 'eventmesh-test'
-include 'eventmesh-spi'
-include 'eventmesh-connector-plugin:eventmesh-connector-api'
-include 'eventmesh-connector-plugin:eventmesh-connector-rocketmq'
-include 'eventmesh-security-plugin:eventmesh-security-api'
-include 'eventmesh-security-plugin:eventmesh-security-acl'
-
+ testImplementation
project(":eventmesh-registry-plugin:eventmesh-registry-api")
+}
\ No newline at end of file
diff --git
a/eventmesh-registry-plugin/eventmesh-registry-rocketmq-namesrv/src/main/java/org/apache/eventmesh/registry/namesrv/RegistryServiceNamesrvImpl.java
b/eventmesh-registry-plugin/eventmesh-registry-rocketmq-namesrv/src/main/java/org/apache/eventmesh/registry/namesrv/RegistryServiceNamesrvImpl.java
new file mode 100644
index 0000000..7b11b51
--- /dev/null
+++
b/eventmesh-registry-plugin/eventmesh-registry-rocketmq-namesrv/src/main/java/org/apache/eventmesh/registry/namesrv/RegistryServiceNamesrvImpl.java
@@ -0,0 +1,63 @@
+/*
+ * 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.namesrv;
+
+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 java.util.List;
+import java.util.Map;
+
+public class RegistryServiceNamesrvImpl implements RegistryService {
+ @Override
+ public void init() throws RegistryException {
+
+ }
+
+ @Override
+ public void start() throws RegistryException {
+
+ }
+
+ @Override
+ public void shutdown() throws RegistryException {
+
+ }
+
+ @Override
+ public List<EventMeshDataInfo> findEventMeshInfoByCluster(String
clusterName) throws RegistryException {
+ return null;
+ }
+
+ @Override
+ public Map<String, Map<String, Integer>>
findEventMeshClientDistributionData(String clusterName, String group, String
purpose) throws RegistryException {
+ return null;
+ }
+
+ @Override
+ public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo)
throws RegistryException {
+ return true;
+ }
+
+ @Override
+ public boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo)
throws RegistryException {
+ return true;
+ }
+}
diff --git a/eventmesh-common/src/test/resources/configuration.properties
b/eventmesh-registry-plugin/eventmesh-registry-rocketmq-namesrv/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.registry.RegistryService
similarity index 75%
copy from eventmesh-common/src/test/resources/configuration.properties
copy to
eventmesh-registry-plugin/eventmesh-registry-rocketmq-namesrv/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.registry.RegistryService
index c9425d7..5e75096 100644
--- a/eventmesh-common/src/test/resources/configuration.properties
+++
b/eventmesh-registry-plugin/eventmesh-registry-rocketmq-namesrv/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.registry.RegistryService
@@ -1,4 +1,3 @@
-#
# 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.
@@ -13,13 +12,5 @@
# 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.
-#
-eventMesh.server.env=value1
-eventMesh.server.idc=value2
-eventMesh.sysid=3
-eventMesh.server.cluster=value4
-eventMesh.server.name=value5
-eventMesh.server.hostIp=value6
-eventMesh.connector.plugin.type=rocketmq
-eventMesh.security.plugin.type=security
+namesrv=org.apache.eventmesh.registry.namesrv.RegistryServiceNamesrvImpl
\ No newline at end of file
diff --git a/eventmesh-runtime/build.gradle b/eventmesh-runtime/build.gradle
index 63767bd..ffa13cc 100644
--- a/eventmesh-runtime/build.gradle
+++ b/eventmesh-runtime/build.gradle
@@ -26,8 +26,13 @@ dependencies {
implementation
project(":eventmesh-connector-plugin:eventmesh-connector-api")
implementation project(":eventmesh-security-plugin:eventmesh-security-api")
implementation project(":eventmesh-security-plugin:eventmesh-security-acl")
+ implementation project(":eventmesh-registry-plugin:eventmesh-registry-api")
+ implementation
project(":eventmesh-registry-plugin:eventmesh-registry-rocketmq-namesrv")
+
testImplementation
project(":eventmesh-connector-plugin:eventmesh-connector-api")
testImplementation
project(":eventmesh-security-plugin:eventmesh-security-api")
testImplementation
project(":eventmesh-security-plugin:eventmesh-security-acl")
+ testImplementation
project(":eventmesh-registry-plugin:eventmesh-registry-api")
+ testImplementation
project(":eventmesh-registry-plugin:eventmesh-registry-rocketmq-namesrv")
}
diff --git a/eventmesh-runtime/conf/eventmesh.properties
b/eventmesh-runtime/conf/eventmesh.properties
index 14123d4..cf09fb7 100644
--- a/eventmesh-runtime/conf/eventmesh.properties
+++ b/eventmesh-runtime/conf/eventmesh.properties
@@ -55,6 +55,10 @@
eventMesh.server.registry.fetchRegistryAddrIntervalInMills=20000
#auto-ack
#eventMesh.server.defibus.client.comsumeTimeoutInMin=5
+#sleep interval between closing client of different group in server graceful
shutdown
+eventMesh.server.gracefulShutdown.sleepIntervalInMills=1000
+eventMesh.server.rebalanceRedirect.sleepIntervalInMills=200
+
#connector plugin
eventMesh.connector.plugin.type=rocketmq
@@ -62,5 +66,9 @@ eventMesh.connector.plugin.type=rocketmq
eventMesh.server.security.enabled=false
eventMesh.security.plugin.type=security
+#registry plugin
+eventMesh.registry.plugin.type=namesrv
+eventMesh.server.registry.enabled=false
+
#prometheusPort
eventMesh.metrics.prometheus.port=19090
\ No newline at end of file
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/controller/ClientManageController.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/controller/ClientManageController.java
index 573eb02..e2cb042 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/controller/ClientManageController.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/controller/ClientManageController.java
@@ -22,15 +22,7 @@ import java.net.InetSocketAddress;
import com.sun.net.httpserver.HttpServer;
-import
org.apache.eventmesh.runtime.admin.handler.RedirectClientByIpPortHandler;
-import org.apache.eventmesh.runtime.admin.handler.RedirectClientByPathHandler;
-import
org.apache.eventmesh.runtime.admin.handler.RedirectClientBySubSystemHandler;
-import org.apache.eventmesh.runtime.admin.handler.RejectAllClientHandler;
-import org.apache.eventmesh.runtime.admin.handler.RejectClientByIpPortHandler;
-import
org.apache.eventmesh.runtime.admin.handler.RejectClientBySubSystemHandler;
-import org.apache.eventmesh.runtime.admin.handler.ShowClientBySystemHandler;
-import org.apache.eventmesh.runtime.admin.handler.ShowClientHandler;
-import
org.apache.eventmesh.runtime.admin.handler.ShowListenClientByTopicHandler;
+import org.apache.eventmesh.runtime.admin.handler.*;
import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -57,7 +49,7 @@ public class ClientManageController {
server.createContext("/clientManage/redirectClientByPath", new
RedirectClientByPathHandler(eventMeshTCPServer));
server.createContext("/clientManage/redirectClientByIpPort", new
RedirectClientByIpPortHandler(eventMeshTCPServer));
server.createContext("/clientManage/showListenClientByTopic", new
ShowListenClientByTopicHandler(eventMeshTCPServer));
-
+ server.createContext("/eventMesh/recommend", new
QueryRecommendEventMeshHandler(eventMeshTCPServer));
server.start();
logger.info("ClientManageController start success, port:{}", port);
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/QueryRecommendEventMeshHandler.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/QueryRecommendEventMeshHandler.java
new file mode 100644
index 0000000..41af41a
--- /dev/null
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/QueryRecommendEventMeshHandler.java
@@ -0,0 +1,91 @@
+/*
+ * 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.admin.handler;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import
org.apache.eventmesh.runtime.core.protocol.tcp.client.group.ClientGroupWrapper;
+import
org.apache.eventmesh.runtime.core.protocol.tcp.client.group.ClientSessionGroupMapping;
+import
org.apache.eventmesh.runtime.core.protocol.tcp.client.recommend.EventMeshRecommendImpl;
+import
org.apache.eventmesh.runtime.core.protocol.tcp.client.recommend.EventMeshRecommendStrategy;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
+import org.apache.eventmesh.runtime.util.NetUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * query recommend eventmesh
+ */
+public class QueryRecommendEventMeshHandler implements HttpHandler {
+
+ private Logger logger =
LoggerFactory.getLogger(QueryRecommendEventMeshHandler.class);
+
+ private final EventMeshTCPServer eventMeshTCPServer;
+
+ public QueryRecommendEventMeshHandler(EventMeshTCPServer
eventMeshTCPServer) {
+ this.eventMeshTCPServer = eventMeshTCPServer;
+ }
+
+ @Override
+ public void handle(HttpExchange httpExchange) throws IOException {
+ String result = "";
+ OutputStream out = httpExchange.getResponseBody();
+ try{
+
if(!eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshServerRegistryEnable)
{
+ throw new Exception("registry enable config is false, not
support");
+ }
+ String queryString = httpExchange.getRequestURI().getQuery();
+ Map<String,String> queryStringInfo =
NetUtils.formData2Dic(queryString);
+ String group =
queryStringInfo.get(EventMeshConstants.MANAGE_GROUP);
+ String purpose =
queryStringInfo.get(EventMeshConstants.MANAGE_PURPOSE);
+ if (StringUtils.isBlank(group) || StringUtils.isBlank(purpose)) {
+ httpExchange.sendResponseHeaders(200, 0);
+ result = "params illegal!";
+ out.write(result.getBytes());
+ return;
+ }
+
+ EventMeshRecommendStrategy eventMeshRecommendStrategy = new
EventMeshRecommendImpl(eventMeshTCPServer);
+ String recommendEventMeshResult =
eventMeshRecommendStrategy.calculateRecommendEventMesh(group, purpose);
+ result = (recommendEventMeshResult == null) ? "null" :
recommendEventMeshResult;
+ logger.info("recommend eventmesh:{},group:{},purpose:{}",result,
group, purpose);
+ httpExchange.sendResponseHeaders(200, 0);
+ out.write(result.getBytes());
+ }catch (Exception e){
+ logger.error("QueryRecommendEventMeshHandler fail...", e);
+ }finally {
+ if(out != null){
+ try {
+ out.close();
+ }catch (IOException e){
+ logger.warn("out close failed...", e);
+ }
+ }
+ }
+ }
+}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
index 39c2a0c..ca47338 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java
@@ -23,6 +23,7 @@ import
org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
import org.apache.eventmesh.runtime.configuration.EventMeshTCPConfiguration;
import org.apache.eventmesh.runtime.connector.ConnectorResource;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.registry.Registry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,6 +41,8 @@ public class EventMeshServer {
private Acl acl;
+ private Registry registry;
+
private ConnectorResource connectorResource;
private ServiceState serviceState;
@@ -49,6 +52,7 @@ public class EventMeshServer {
this.eventMeshHttpConfiguration = eventMeshHttpConfiguration;
this.eventMeshTCPConfiguration = eventMeshTCPConfiguration;
this.acl = new Acl();
+ this.registry = new Registry();
this.connectorResource = new ConnectorResource();
}
@@ -57,11 +61,15 @@ public class EventMeshServer {
acl.init(eventMeshHttpConfiguration.eventMeshSecurityPluginType);
}
+ if (eventMeshTCPConfiguration != null &&
eventMeshTCPConfiguration.eventMeshTcpServerEnabled &&
eventMeshTCPConfiguration.eventMeshServerRegistryEnable) {
+
registry.init(eventMeshTCPConfiguration.eventMeshRegistryPluginType);
+ }
+
connectorResource.init(eventMeshHttpConfiguration.eventMeshConnectorPluginType);
eventMeshHTTPServer = new EventMeshHTTPServer(this,
eventMeshHttpConfiguration);
eventMeshHTTPServer.init();
- eventMeshTCPServer = new EventMeshTCPServer(this,
eventMeshTCPConfiguration);
+ eventMeshTCPServer = new EventMeshTCPServer(this,
eventMeshTCPConfiguration, registry);
if (eventMeshTCPConfiguration != null &&
eventMeshTCPConfiguration.eventMeshTcpServerEnabled) {
eventMeshTCPServer.init();
}
@@ -78,6 +86,10 @@ public class EventMeshServer {
acl.start();
}
+ if (eventMeshTCPConfiguration != null &&
eventMeshTCPConfiguration.eventMeshTcpServerEnabled &&
eventMeshTCPConfiguration.eventMeshServerRegistryEnable) {
+ registry.start();
+ }
+
eventMeshHTTPServer.start();
if (eventMeshTCPConfiguration != null &&
eventMeshTCPConfiguration.eventMeshTcpServerEnabled) {
eventMeshTCPServer.start();
@@ -92,6 +104,9 @@ public class EventMeshServer {
eventMeshHTTPServer.shutdown();
if (eventMeshTCPConfiguration != null &&
eventMeshTCPConfiguration.eventMeshTcpServerEnabled) {
eventMeshTCPServer.shutdown();
+ if(eventMeshTCPConfiguration.eventMeshServerRegistryEnable) {
+ registry.shutdown();
+ }
}
connectorResource.release();
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java
index c24caa4..4620ce7 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTCPServer.java
@@ -17,10 +17,7 @@
package org.apache.eventmesh.runtime.boot;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.*;
import com.google.common.util.concurrent.RateLimiter;
@@ -32,16 +29,24 @@ import io.netty.handler.timeout.IdleStateHandler;
import io.netty.handler.traffic.ChannelTrafficShapingHandler;
import io.netty.handler.traffic.GlobalTrafficShapingHandler;
+import org.apache.eventmesh.api.registry.dto.EventMeshRegisterInfo;
+import org.apache.eventmesh.api.registry.dto.EventMeshUnRegisterInfo;
+import org.apache.eventmesh.common.EventMeshException;
+import org.apache.eventmesh.common.IPUtil;
import org.apache.eventmesh.common.ThreadPoolFactory;
import org.apache.eventmesh.common.protocol.tcp.codec.Codec;
import org.apache.eventmesh.runtime.admin.controller.ClientManageController;
import org.apache.eventmesh.runtime.configuration.EventMeshTCPConfiguration;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
import
org.apache.eventmesh.runtime.core.protocol.tcp.client.EventMeshTcpConnectionHandler;
import
org.apache.eventmesh.runtime.core.protocol.tcp.client.EventMeshTcpExceptionHandler;
import
org.apache.eventmesh.runtime.core.protocol.tcp.client.EventMeshTcpMessageDispatcher;
import
org.apache.eventmesh.runtime.core.protocol.tcp.client.group.ClientSessionGroupMapping;
+import
org.apache.eventmesh.runtime.core.protocol.tcp.client.rebalance.EventMeshRebalanceService;
+import
org.apache.eventmesh.runtime.core.protocol.tcp.client.rebalance.EventmeshRebalanceImpl;
import
org.apache.eventmesh.runtime.core.protocol.tcp.client.session.push.retry.EventMeshTcpRetryer;
import org.apache.eventmesh.runtime.metrics.tcp.EventMeshTcpMonitor;
+import org.apache.eventmesh.runtime.registry.Registry;
import org.apache.eventmesh.runtime.util.EventMeshThreadFactoryImpl;
public class EventMeshTCPServer extends AbstractRemotingServer {
@@ -66,6 +71,10 @@ public class EventMeshTCPServer extends
AbstractRemotingServer {
private ExecutorService broadcastMsgDownstreamExecutorService;
+ private Registry registry;
+
+ private EventMeshRebalanceService eventMeshRebalanceService;
+
public void setClientSessionGroupMapping(ClientSessionGroupMapping
clientSessionGroupMapping) {
this.clientSessionGroupMapping = clientSessionGroupMapping;
}
@@ -111,10 +120,11 @@ public class EventMeshTCPServer extends
AbstractRemotingServer {
private RateLimiter rateLimiter;
public EventMeshTCPServer(EventMeshServer eventMeshServer,
- EventMeshTCPConfiguration
eventMeshTCPConfiguration) {
+ EventMeshTCPConfiguration
eventMeshTCPConfiguration, Registry registry) {
super();
this.eventMeshServer = eventMeshServer;
this.eventMeshTCPConfiguration = eventMeshTCPConfiguration;
+ this.registry = registry;
}
private void startServer() throws Exception {
@@ -189,6 +199,10 @@ public class EventMeshTCPServer extends
AbstractRemotingServer {
eventMeshTcpMonitor = new EventMeshTcpMonitor(this);
eventMeshTcpMonitor.init();
+ if(eventMeshTCPConfiguration.eventMeshServerRegistryEnable) {
+ eventMeshRebalanceService = new EventMeshRebalanceService(this,
new EventmeshRebalanceImpl(this));
+ eventMeshRebalanceService.init();
+ }
logger.info("--------------------------EventMeshTCPServer Inited");
}
@@ -202,6 +216,11 @@ public class EventMeshTCPServer extends
AbstractRemotingServer {
eventMeshTcpMonitor.start();
+ if(eventMeshTCPConfiguration.eventMeshServerRegistryEnable) {
+ eventMeshRebalanceService.start();
+ selfRegisterToRegistry();
+ }
+
clientManageController.start();
logger.info("--------------------------EventMeshTCPServer Started");
@@ -214,6 +233,12 @@ public class EventMeshTCPServer extends
AbstractRemotingServer {
logger.info("shutdown bossGroup, no client is allowed to connect
access server");
}
+ if(eventMeshTCPConfiguration.eventMeshServerRegistryEnable) {
+ eventMeshRebalanceService.shutdown();
+
+ selfUnRegisterToRegistry();
+ }
+
clientSessionGroupMapping.shutdown();
try {
Thread.sleep(40 * 1000);
@@ -240,6 +265,56 @@ public class EventMeshTCPServer extends
AbstractRemotingServer {
logger.info("--------------------------EventMeshTCPServer Shutdown");
}
+ private void selfRegisterToRegistry() throws Exception {
+
+ boolean registerResult = registerToRegistry();
+ if (!registerResult) {
+ throw new EventMeshException("eventMesh fail to register");
+ }
+
+ tcpRegisterTask = scheduler.scheduleAtFixedRate(() -> {
+ try {
+ boolean heartbeatResult = registerToRegistry();
+ if (!heartbeatResult) {
+ logger.error("selfRegisterToRegistry fail");
+ }
+ } catch (Exception ex) {
+ logger.error("selfRegisterToRegistry fail", ex);
+ }
+ }, eventMeshTCPConfiguration.eventMeshRegisterIntervalInMills,
eventMeshTCPConfiguration.eventMeshRegisterIntervalInMills,
TimeUnit.MILLISECONDS);
+ }
+
+ public boolean registerToRegistry() {
+ boolean registerResult = false;
+ try{
+ String endPoints = IPUtil.getLocalAddress()
+ + EventMeshConstants.IP_PORT_SEPARATOR +
eventMeshTCPConfiguration.eventMeshTcpServerPort;
+ EventMeshRegisterInfo self = new EventMeshRegisterInfo();
+
self.setEventMeshClusterName(eventMeshTCPConfiguration.eventMeshCluster);
+ self.setEventMeshName(eventMeshTCPConfiguration.eventMeshName);
+ self.setEndPoint(endPoints);
+
self.setEventMeshInstanceNumMap(clientSessionGroupMapping.prepareProxyClientDistributionData());
+ registerResult = registry.register(self);
+ }catch (Exception e){
+ logger.warn("eventMesh register to registry failed", e);
+ }
+
+ return registerResult;
+ }
+
+ private void selfUnRegisterToRegistry() throws Exception {
+ EventMeshUnRegisterInfo eventMeshUnRegisterInfo = new
EventMeshUnRegisterInfo();
+
eventMeshUnRegisterInfo.setEventMeshClusterName(eventMeshTCPConfiguration.eventMeshCluster);
+
eventMeshUnRegisterInfo.setEventMeshName(eventMeshTCPConfiguration.eventMeshName);
+ boolean registerResult = registry.unRegister(eventMeshUnRegisterInfo);
+ if (!registerResult) {
+ throw new EventMeshException("eventMesh fail to unRegister");
+ }
+
+ //cancel task
+ tcpRegisterTask.cancel(true);
+ }
+
private void initThreadPool() throws Exception {
super.init("eventMesh-tcp");
@@ -296,4 +371,12 @@ public class EventMeshTCPServer extends
AbstractRemotingServer {
public EventMeshTCPConfiguration getEventMeshTCPConfiguration() {
return eventMeshTCPConfiguration;
}
+
+ public Registry getRegistry() {
+ return registry;
+ }
+
+ public EventMeshRebalanceService getEventMeshRebalanceService() {
+ return eventMeshRebalanceService;
+ }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshTCPConfiguration.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshTCPConfiguration.java
index 77021a8..fee44b4 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshTCPConfiguration.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshTCPConfiguration.java
@@ -72,6 +72,10 @@ public class EventMeshTCPConfiguration extends
CommonConfiguration {
public int eventMeshTcpPushFailIsolateTimeInMills = 30 * 1000;
+ public int gracefulShutdownSleepIntervalInMills = 1000;
+
+ public int sleepIntervalInRebalanceRedirectMills = 200;
+
private TrafficShapingConfig gtc = new TrafficShapingConfig(0, 10_000,
1_000, 2000);
private TrafficShapingConfig ctc = new TrafficShapingConfig(0, 2_000,
1_000, 10_000);
@@ -122,6 +126,11 @@ public class EventMeshTCPConfiguration extends
CommonConfiguration {
eventMeshTcpSendBackEnabled =
configurationWrapper.getBoolProp(ConfKeys.KEYS_EVENTMESH_TCP_SEND_BACK_ENABLED,
eventMeshTcpSendBackEnabled);
eventMeshTcpPushFailIsolateTimeInMills =
configurationWrapper.getIntProp(ConfKeys.KEYS_EVENTMESH_SERVER_PUSH_FAIL_ISOLATE_TIME,
eventMeshTcpPushFailIsolateTimeInMills);
+
+ gracefulShutdownSleepIntervalInMills =
configurationWrapper.getIntProp(ConfKeys.KEYS_EVENTMESH_SERVER_GRACEFUL_SHUTDOWN_SLEEP_TIME,
gracefulShutdownSleepIntervalInMills);
+
+ sleepIntervalInRebalanceRedirectMills =
configurationWrapper.getIntProp(ConfKeys.KEYS_EVENTMESH_SERVER_REBALANCE_REDIRECT_SLEEP_TIME,
sleepIntervalInRebalanceRedirectMills);
+
}
public TrafficShapingConfig getGtc() {
@@ -156,7 +165,8 @@ public class EventMeshTCPConfiguration extends
CommonConfiguration {
public static String KEYS_EVENTMESH_TCP_SERVER_ENABLED =
"eventMesh.server.tcp.enabled";
public static String KEYS_EVENTMESH_TCP_SEND_BACK_ENABLED =
"eventMesh.server.tcp.sendBack.enabled";
public static String KEYS_EVENTMESH_SERVER_PUSH_FAIL_ISOLATE_TIME =
"eventMesh.server.tcp.pushFailIsolateTimeInMills";
- public static String KEYS_EVENTMESH_TCP_DOWNSTREAM_MAP_SIZE =
"eventMesh.server.tcp.downstreamMapSize";
+ public static String
KEYS_EVENTMESH_SERVER_GRACEFUL_SHUTDOWN_SLEEP_TIME =
"eventMesh.server.gracefulShutdown.sleepIntervalInMills";
+ public static String
KEYS_EVENTMESH_SERVER_REBALANCE_REDIRECT_SLEEP_TIME =
"eventMesh.server.rebalanceRedirect.sleepIntervalInM";
}
public static class TrafficShapingConfig {
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/constants/EventMeshConstants.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/constants/EventMeshConstants.java
index 6a5d35d..810b3f2 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/constants/EventMeshConstants.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/constants/EventMeshConstants.java
@@ -29,6 +29,8 @@ public class EventMeshConstants {
public static final String DEFAULT_CHARSET = "UTF-8";
+ public static final String IP_PORT_SEPARATOR = ":";
+
public static final String EVENTMESH_CONF_HOME =
System.getProperty("confPath", System.getenv("confPath"));
public static final String EVENTMESH_CONF_FILE = "eventmesh.properties";
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/EventMeshTcpMessageDispatcher.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/EventMeshTcpMessageDispatcher.java
index 4d68ba8..df09394 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/EventMeshTcpMessageDispatcher.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/EventMeshTcpMessageDispatcher.java
@@ -24,14 +24,7 @@ import org.apache.eventmesh.common.protocol.tcp.*;
import org.apache.eventmesh.common.protocol.tcp.Package;
import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
import
org.apache.eventmesh.runtime.core.protocol.tcp.client.session.SessionState;
-import org.apache.eventmesh.runtime.core.protocol.tcp.client.task.GoodbyeTask;
-import
org.apache.eventmesh.runtime.core.protocol.tcp.client.task.HeartBeatTask;
-import org.apache.eventmesh.runtime.core.protocol.tcp.client.task.HelloTask;
-import org.apache.eventmesh.runtime.core.protocol.tcp.client.task.ListenTask;
-import
org.apache.eventmesh.runtime.core.protocol.tcp.client.task.MessageAckTask;
-import
org.apache.eventmesh.runtime.core.protocol.tcp.client.task.MessageTransferTask;
-import
org.apache.eventmesh.runtime.core.protocol.tcp.client.task.SubscribeTask;
-import
org.apache.eventmesh.runtime.core.protocol.tcp.client.task.UnSubscribeTask;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.task.*;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -55,6 +48,12 @@ public class EventMeshTcpMessageDispatcher extends
SimpleChannelInboundHandler<P
try {
Runnable task;
cmd = pkg.getHeader().getCommand();
+ if (cmd.equals(Command.RECOMMEND_REQUEST)) {
+ messageLogger.info("pkg|c2eventMesh|cmd={}|pkg={}", cmd, pkg);
+ task = new RecommendTask(pkg, ctx, startTime,
eventMeshTCPServer);
+ eventMeshTCPServer.getTaskHandleExecutorService().submit(task);
+ return;
+ }
if (cmd.equals(Command.HELLO_REQUEST)) {
messageLogger.info("pkg|c2eventMesh|cmd={}|pkg={}", cmd, pkg);
task = new HelloTask(pkg, ctx, startTime, eventMeshTCPServer);
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
index f7b2893..11aa1d8 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java
@@ -407,6 +407,35 @@ public class ClientSessionGroupMapping {
public void shutdown() throws Exception {
logger.info("begin to close sessions gracefully");
+ for(ClientGroupWrapper clientGroupWrapper : clientGroupMap.values()){
+ for(Session subSession :
clientGroupWrapper.getGroupConsumerSessions()){
+ try {
+
EventMeshTcp2Client.serverGoodby2Client(eventMeshTCPServer, subSession, this);
+ } catch (Exception e) {
+ logger.error("say goodbye to subSession error! {}",
subSession, e);
+ }
+ }
+
+ for(Session pubSession :
clientGroupWrapper.getGroupProducerSessions()){
+ try {
+
EventMeshTcp2Client.serverGoodby2Client(eventMeshTCPServer, pubSession, this);
+ } catch (Exception e) {
+ logger.error("say goodbye to pubSession error! {}",
pubSession, e);
+ }
+ }
+ try {
+
Thread.sleep(eventMeshTCPServer.getEventMeshTCPConfiguration().gracefulShutdownSleepIntervalInMills);
+ } catch (InterruptedException e) {
+ logger.warn("Thread.sleep occur InterruptedException", e);
+ }
+ }
+
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ logger.warn("Thread.sleep occur InterruptedException", e);
+ }
+
sessionTable.values().parallelStream().forEach(itr -> {
try {
EventMeshTcp2Client.serverGoodby2Client(this.eventMeshTCPServer,itr, this);
@@ -441,4 +470,20 @@ public class ClientSessionGroupMapping {
return result;
}
+
+ public Map<String, Map<String, Integer>>
prepareProxyClientDistributionData(){
+ Map<String, Map<String, Integer>> result = null;
+
+ if(!clientGroupMap.isEmpty()){
+ result = new HashMap<>();
+ for(Map.Entry<String, ClientGroupWrapper> entry :
clientGroupMap.entrySet()){
+ Map<String, Integer> map = new HashMap();
+
map.put(EventMeshConstants.PURPOSE_SUB,entry.getValue().getGroupConsumerSessions().size());
+
map.put(EventMeshConstants.PURPOSE_PUB,entry.getValue().getGroupProducerSessions().size());
+ result.put(entry.getKey(), map);
+ }
+ }
+
+ return result;
+ }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/rebalance/EventMeshRebalanceService.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/rebalance/EventMeshRebalanceService.java
new file mode 100644
index 0000000..516afa3
--- /dev/null
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/rebalance/EventMeshRebalanceService.java
@@ -0,0 +1,72 @@
+/*
+ * 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.protocol.tcp.client.rebalance;
+
+import org.apache.eventmesh.common.ThreadPoolFactory;
+import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
+import org.apache.eventmesh.runtime.util.EventMeshThreadFactoryImpl;
+import org.apache.eventmesh.runtime.util.EventMeshUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+public class EventMeshRebalanceService {
+ protected final Logger logger =
LoggerFactory.getLogger(EventMeshRebalanceService.class);
+
+ private EventMeshTCPServer eventMeshTCPServer;
+
+ private Integer rebalanceIntervalMills;
+
+ private EventMeshRebalanceStrategy rebalanceStrategy;
+
+ private ScheduledExecutorService serviceRebalanceScheduler;
+
+ public EventMeshRebalanceService(EventMeshTCPServer eventMeshTCPServer,
EventMeshRebalanceStrategy rebalanceStrategy) {
+ this.eventMeshTCPServer = eventMeshTCPServer;
+ this.rebalanceStrategy = rebalanceStrategy;
+ this.rebalanceIntervalMills =
eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshTcpRebalanceIntervalInMills;
+ }
+
+ public void init(){
+ this.serviceRebalanceScheduler =
ThreadPoolFactory.createScheduledExecutor(5, new
EventMeshThreadFactoryImpl("proxy-rebalance-sch",true));
+ logger.info("rebalance service inited......");
+ }
+
+ public void start() throws Exception {
+ rebalanceStrategy.doRebalance();
+ serviceRebalanceScheduler.scheduleAtFixedRate(() -> {
+ try {
+ rebalanceStrategy.doRebalance();
+ } catch (Exception ex) {
+ logger.error("RebalanceByService failed", ex);
+ }
+ }, rebalanceIntervalMills, rebalanceIntervalMills,
TimeUnit.MILLISECONDS);
+ logger.info("rebalance service started......");
+ }
+
+ public void shutdown(){
+ this.serviceRebalanceScheduler.shutdown();
+ logger.info("rebalance service shutdown......");
+ }
+
+ public void printRebalanceThreadPoolState(){
+ EventMeshUtil.printState((ThreadPoolExecutor)
serviceRebalanceScheduler);
+ }
+}
diff --git a/settings.gradle
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/rebalance/EventMeshRebalanceStrategy.java
similarity index 63%
copy from settings.gradle
copy to
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/rebalance/EventMeshRebalanceStrategy.java
index 145df7e..2bc32d1 100644
--- a/settings.gradle
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/rebalance/EventMeshRebalanceStrategy.java
@@ -14,17 +14,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.eventmesh.runtime.core.protocol.tcp.client.rebalance;
-rootProject.name = 'EventMesh'
-String jdkVersion = "${jdk}"
-include 'eventmesh-runtime'
-include 'eventmesh-sdk-java'
-include 'eventmesh-common'
-include 'eventmesh-starter'
-include 'eventmesh-test'
-include 'eventmesh-spi'
-include 'eventmesh-connector-plugin:eventmesh-connector-api'
-include 'eventmesh-connector-plugin:eventmesh-connector-rocketmq'
-include 'eventmesh-security-plugin:eventmesh-security-api'
-include 'eventmesh-security-plugin:eventmesh-security-acl'
-
+public interface EventMeshRebalanceStrategy {
+ void doRebalance() throws Exception;
+}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/rebalance/EventmeshRebalanceImpl.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/rebalance/EventmeshRebalanceImpl.java
new file mode 100644
index 0000000..204f9ed
--- /dev/null
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/rebalance/EventmeshRebalanceImpl.java
@@ -0,0 +1,208 @@
+/*
+ * 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.protocol.tcp.client.rebalance;
+
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo;
+import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import
org.apache.eventmesh.runtime.core.protocol.tcp.client.EventMeshTcp2Client;
+import
org.apache.eventmesh.runtime.core.protocol.tcp.client.recommend.EventMeshRecommendImpl;
+import
org.apache.eventmesh.runtime.core.protocol.tcp.client.recommend.EventMeshRecommendStrategy;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+
+public class EventmeshRebalanceImpl implements EventMeshRebalanceStrategy {
+
+ protected final Logger logger =
LoggerFactory.getLogger(EventmeshRebalanceImpl.class);
+
+ private EventMeshTCPServer eventMeshTCPServer;
+
+ public EventmeshRebalanceImpl(EventMeshTCPServer eventMeshTCPServer){
+ this.eventMeshTCPServer = eventMeshTCPServer;
+ }
+
+ @Override
+ public void doRebalance() throws Exception {
+ long startTime = System.currentTimeMillis();
+ logger.info("doRebalance start===========startTime:{}",startTime);
+
+ Set<String> groupSet =
eventMeshTCPServer.getClientSessionGroupMapping().getClientGroupMap().keySet();
+ if(CollectionUtils.isEmpty(groupSet)){
+ logger.warn("doRebalance failed,eventmesh has no group, please
check eventmeshData");
+ return;
+ }
+
+ final String cluster =
eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshCluster;
+ //get eventmesh of local idc
+ Map<String, String> localEventMeshMap =
queryLocalEventMeshMap(cluster);
+ if(localEventMeshMap == null || localEventMeshMap.size() == 0){
+ return;
+ }
+
+ for(String group : groupSet){
+ doRebalanceByGroup(cluster, group, EventMeshConstants.PURPOSE_SUB,
localEventMeshMap);
+ doRebalanceByGroup(cluster, group, EventMeshConstants.PURPOSE_PUB,
localEventMeshMap);
+ }
+ logger.info("doRebalance end===========startTime:{}, cost:{}",
startTime, System.currentTimeMillis() - startTime);
+ }
+
+ private Map<String, String> queryLocalEventMeshMap(String cluster){
+ Map<String, String> localEventMeshMap = null;
+ List<EventMeshDataInfo> eventMeshDataInfoList = null;
+ try{
+ eventMeshDataInfoList =
eventMeshTCPServer.getRegistry().findEventMeshInfoByCluster(cluster);
+
+ if(eventMeshDataInfoList == null ||
CollectionUtils.isEmpty(eventMeshDataInfoList)){
+ logger.warn("doRebalance failed,query eventmesh instances is
null from registry,cluster:{}", cluster);
+ return null;
+ }
+ localEventMeshMap = new HashMap<>();
+ String localIdc =
eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshIDC;
+ for(EventMeshDataInfo eventMeshDataInfo : eventMeshDataInfoList){
+ String idc =
eventMeshDataInfo.getEventMeshName().split("-")[0];
+ if(StringUtils.isNotBlank(idc) && StringUtils.equals(idc,
localIdc)){
+
localEventMeshMap.put(eventMeshDataInfo.getEventMeshName(),
eventMeshDataInfo.getEndpoint());
+ }
+ }
+
+ if(0 == localEventMeshMap.size()){
+ logger.warn("doRebalance failed,query eventmesh instances of
localIDC is null from registry,localIDC:{},cluster:{}", localIdc,cluster);
+ return null;
+ }
+ }catch (Exception e){
+ logger.warn("doRebalance failed,findEventMeshInfoByCluster
failed,cluster:{},errMsg:{}", cluster, e);
+ return null;
+ }
+
+ return localEventMeshMap;
+ }
+
+ private void doRebalanceByGroup(String cluster, String group, String
purpose, Map<String, String> eventMeshMap) throws Exception{
+ //query distribute data of loacl idc
+ Map<String, Integer> clientDistributionMap =
queryLocalEventMeshDistributeData(cluster, group, purpose, eventMeshMap);
+ if(clientDistributionMap == null || clientDistributionMap.size() == 0){
+ return;
+ }
+
+ int sum = 0;
+ for(Integer item : clientDistributionMap.values()){
+ sum += item.intValue();
+ }
+ int currentNum = 0;
+
if(clientDistributionMap.get(eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshName)
!= null){
+ currentNum =
clientDistributionMap.get(eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshName);
+ }
+ int avgNum = sum / clientDistributionMap.size();
+ int judge = avgNum >= 2 ? avgNum/2 : 1;
+
+ if(currentNum - avgNum > judge) {
+ Set<Session> sessionSet = null;
+ if(EventMeshConstants.PURPOSE_PUB.equals(purpose)){
+ sessionSet =
eventMeshTCPServer.getClientSessionGroupMapping().getClientGroupMap().get(group).getGroupProducerSessions();
+ }else if(EventMeshConstants.PURPOSE_SUB.equals(purpose)){
+ sessionSet =
eventMeshTCPServer.getClientSessionGroupMapping().getClientGroupMap().get(group).getGroupConsumerSessions();
+ }else{
+ logger.warn("doRebalance failed,purpose is not
support,purpose:{}", purpose);
+ return;
+ }
+
+ List<Session> sessionList = new ArrayList<>(sessionSet);
+ Collections.shuffle(new ArrayList<>(sessionList));
+ EventMeshRecommendStrategy eventMeshRecommendStrategy = new
EventMeshRecommendImpl(eventMeshTCPServer);
+ List<String> eventMeshRecommendResult =
eventMeshRecommendStrategy.calculateRedirectRecommendEventMesh(eventMeshMap,
clientDistributionMap, group, judge);
+ if(eventMeshRecommendResult == null ||
eventMeshRecommendResult.size() != judge){
+ logger.warn("doRebalance failed,recommendProxyNum is not
consistent,recommendResult:{},judge:{}", eventMeshRecommendResult, judge);
+ return;
+ }
+ logger.info("doRebalance redirect
start---------------------group:{},purpose:{},judge:{}", group, purpose, judge);
+ for(int i= 0; i<judge; i++){
+ //String redirectSessionAddr =
ProxyTcp2Client.redirectClientForRebalance(sessionList.get(i),
eventMeshTCPServer.getClientSessionGroupMapping());
+ String newProxyIp =
eventMeshRecommendResult.get(i).split(":")[0];
+ String newProxyPort =
eventMeshRecommendResult.get(i).split(":")[1];
+ String redirectSessionAddr =
EventMeshTcp2Client.redirectClient2NewEventMesh(eventMeshTCPServer,newProxyIp,Integer.valueOf(newProxyPort),sessionList.get(i),
eventMeshTCPServer.getClientSessionGroupMapping());
+ logger.info("doRebalance,redirect sessionAddr:{}",
redirectSessionAddr);
+ try {
+
Thread.sleep(eventMeshTCPServer.getEventMeshTCPConfiguration().sleepIntervalInRebalanceRedirectMills);
+ } catch (InterruptedException e) {
+ logger.warn("Thread.sleep occur InterruptedException", e);
+ }
+ }
+ logger.info("doRebalance redirect
end---------------------group:{}, purpose:{}", group, purpose);
+ }else{
+ logger.info("rebalance condition not
satisfy,group:{},sum:{},currentNum:{},avgNum:{},judge:{}", group, sum,
currentNum, avgNum, judge);
+ }
+ }
+
+ private Map<String, Integer> queryLocalEventMeshDistributeData(String
cluster, String group, String purpose, Map<String, String> eventMeshMap){
+ Map<String, Integer> localEventMeshDistributeData = null;
+ Map<String, Map<String, Integer>> eventMeshClientDistributionDataMap =
null;
+ try{
+ eventMeshClientDistributionDataMap =
eventMeshTCPServer.getRegistry().findEventMeshClientDistributionData(cluster,
group, purpose);
+
+ if(eventMeshClientDistributionDataMap == null ||
eventMeshClientDistributionDataMap.size() == 0){
+ logger.warn("doRebalance failed,found no distribute data in
regitry, cluster:{}, group:{}, purpose:{}", cluster, group, purpose);
+ return null;
+ }
+
+ localEventMeshDistributeData = new HashMap<>();
+ String localIdc =
eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshIDC;
+ for(Map.Entry<String, Map<String, Integer>> entry :
eventMeshClientDistributionDataMap.entrySet()){
+ String idc = entry.getKey().split("-")[0];
+ if(StringUtils.isNotBlank(idc) && StringUtils.equals(idc,
localIdc)){
+ localEventMeshDistributeData.put(entry.getKey(),
entry.getValue().get(purpose));
+ }
+ }
+
+ if(0 == localEventMeshDistributeData.size()){
+ logger.warn("doRebalance failed,found no distribute data of
localIDC in regitry,cluster:{},group:{}, purpose:{},localIDC:{}", cluster,
group, purpose, localIdc);
+ return null;
+ }
+
+ logger.info("before revert clientDistributionMap:{}, group:{},
purpose:{}", localEventMeshDistributeData, group, purpose);
+ for(String eventMeshName : localEventMeshDistributeData.keySet()){
+ if(!eventMeshMap.keySet().contains(eventMeshName)){
+ logger.warn("doRebalance failed,exist eventMesh not
register but exist in
distributionMap,cluster:{},grpup:{},purpose:{},eventMeshName:{}", cluster,
group, purpose, eventMeshName);
+ return null;
+ }
+ }
+ for(String eventMesh : eventMeshMap.keySet()){
+ if(!localEventMeshDistributeData.keySet().contains(eventMesh)){
+ localEventMeshDistributeData.put(eventMesh, 0);
+ }
+ }
+ logger.info("after revert clientDistributionMap:{}, group:{},
purpose:{}", localEventMeshDistributeData, group, purpose);
+ }catch (Exception e){
+ logger.warn("doRebalance
failed,cluster:{},group:{},purpose:{},findProxyClientDistributionData failed,
errMsg:{}", cluster, group, purpose, e);
+ return null;
+ }
+
+ return localEventMeshDistributeData;
+ }
+
+
+ private class ValueComparator implements Comparator<Map.Entry<String,
Integer>> {
+ @Override
+ public int compare(Map.Entry<String, Integer> x, Map.Entry<String,
Integer> y) {
+ return x.getValue().intValue() - y.getValue().intValue();
+ }
+ }
+}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/recommend/EventMeshRecommendImpl.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/recommend/EventMeshRecommendImpl.java
new file mode 100644
index 0000000..6c8b4a3
--- /dev/null
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/recommend/EventMeshRecommendImpl.java
@@ -0,0 +1,195 @@
+/*
+ * 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.protocol.tcp.client.recommend;
+
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.collections4.MapUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo;
+import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
+import org.apache.eventmesh.runtime.util.ValueComparator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+
+public class EventMeshRecommendImpl implements EventMeshRecommendStrategy {
+
+ protected final Logger logger =
LoggerFactory.getLogger(EventMeshRecommendImpl.class);
+
+ private EventMeshTCPServer eventMeshTCPServer;
+
+ public EventMeshRecommendImpl(EventMeshTCPServer eventMeshTCPServer) {
+ this.eventMeshTCPServer = eventMeshTCPServer;
+ }
+
+ @Override
+ public String calculateRecommendEventMesh(String group, String purpose)
throws Exception {
+ List<EventMeshDataInfo> eventMeshDataInfoList = null;
+ if(StringUtils.isBlank(group) || StringUtils.isBlank(purpose)){
+ logger.warn("EventMeshRecommend failed,params
illegal,group:{},purpose:{}", group, purpose);
+ return null;
+ }
+ final String cluster =
eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshCluster;
+ try{
+ eventMeshDataInfoList =
eventMeshTCPServer.getRegistry().findEventMeshInfoByCluster(cluster);
+ }catch (Exception e){
+ logger.warn("EventMeshRecommend failed, findEventMeshInfoByCluster
failed, cluster:{}, group:{}, purpose:{}, errMsg:{}", cluster, group , purpose,
e);
+ return null;
+ }
+
+ if(eventMeshDataInfoList == null ||
CollectionUtils.isEmpty(eventMeshDataInfoList)){
+ logger.warn("EventMeshRecommend failed,not find eventMesh
instances from registry,cluster:{},group:{},purpose:{}", cluster, group,
purpose);
+ return null;
+ }
+
+ Map<String, String> localEventMeshMap = new HashMap<>();
+ Map<String, String> remoteEventMeshMap = new HashMap<>();
+ String localIdc =
eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshIDC;
+ for(EventMeshDataInfo eventMeshDataInfo : eventMeshDataInfoList){
+ String idc = eventMeshDataInfo.getEventMeshName().split("-")[0];
+ if(StringUtils.isNotBlank(idc)){
+ if(StringUtils.equals(idc, localIdc)){
+
localEventMeshMap.put(eventMeshDataInfo.getEventMeshName(),
eventMeshDataInfo.getEndpoint());
+ }else{
+
remoteEventMeshMap.put(eventMeshDataInfo.getEventMeshName(),
eventMeshDataInfo.getEndpoint());
+ }
+ }else{
+ logger.error("EventMeshName may be illegal,idc is
null,eventMeshName:{}", eventMeshDataInfo.getEventMeshName());
+ }
+ }
+
+ if(localEventMeshMap.size() ==0 && remoteEventMeshMap.size()==0){
+ logger.warn("EventMeshRecommend failed,find no legal eventMesh
instances from registry,localIDC:{}", localIdc);
+ return null;
+ }
+ if(localEventMeshMap.size() > 0){
+ //recommend eventmesh of local idc
+ return
recommendProxyByDistributeData(cluster,group,purpose,localEventMeshMap,true);
+ }else if(remoteEventMeshMap.size() > 0){
+ //recommend eventmesh of other idc
+ return
recommendProxyByDistributeData(cluster,group,purpose,remoteEventMeshMap,false);
+ }else {
+ logger.error("localEventMeshMap or remoteEventMeshMap size error");
+ return null;
+ }
+ }
+
+ @Override
+ public List<String> calculateRedirectRecommendEventMesh(Map<String,
String> eventMeshMap, Map<String, Integer> clientDistributeMap, String group,
int recommendProxyNum) throws Exception {
+
logger.info("eventMeshMap:{},clientDistributionMap:{},group:{},recommendNum:{}",
eventMeshMap,clientDistributeMap,group,recommendProxyNum);
+ List<String> recommendProxyList = null;
+
+ //find eventmesh with least client
+ List<Map.Entry<String, Integer>> list = new ArrayList<>();
+ ValueComparator vc = new ValueComparator();
+ for (Map.Entry<String, Integer> entry :
clientDistributeMap.entrySet()) {
+ list.add(entry);
+ }
+ Collections.sort(list, vc);
+ logger.info("clientDistributionMap after sort:{}", list);
+
+ recommendProxyList = new ArrayList<>(recommendProxyNum);
+ while(recommendProxyList.size() < recommendProxyNum){
+ Map.Entry<String, Integer> minProxyItem = list.get(0);
+ int currProxyNum =
clientDistributeMap.get(eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshName);
+ recommendProxyList.add(eventMeshMap.get(minProxyItem.getKey()));
+
clientDistributeMap.put(minProxyItem.getKey(),minProxyItem.getValue() + 1);
+
clientDistributeMap.put(eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshName,currProxyNum
- 1);
+ Collections.sort(list, vc);
+ logger.info("clientDistributionMap after sort:{}", list);
+ }
+ logger.info("choose proxys with min instance num, group:{},
recommendProxyNum:{}, recommendProxyList:{}", group, recommendProxyNum,
recommendProxyList);
+ return recommendProxyList;
+ }
+
+ private String recommendProxyByDistributeData(String cluster, String
group, String purpose, Map<String,String> eventMeshMap, boolean caculateLocal){
+
logger.info("eventMeshMap:{},cluster:{},group:{},purpose:{},caculateLocal:{}",
eventMeshMap,cluster,group,purpose,caculateLocal);
+
+ String recommendProxyAddr = null;
+ List<String> tmpProxyAddrList = null;
+ Map<String, Map<String, Integer>> eventMeshClientDistributionDataMap =
null;
+ try{
+ eventMeshClientDistributionDataMap =
eventMeshTCPServer.getRegistry().findEventMeshClientDistributionData(cluster,
group, purpose);
+ }catch (Exception e){
+ logger.warn("EventMeshRecommend
failed,findEventMeshClientDistributionData
failed,cluster:{},group:{},purpose:{}, errMsg:{}",cluster,group,purpose, e);
+ }
+
+ if(eventMeshClientDistributionDataMap == null ||
MapUtils.isEmpty(eventMeshClientDistributionDataMap)) {
+ tmpProxyAddrList = new ArrayList<>(eventMeshMap.values());
+ Collections.shuffle(tmpProxyAddrList);
+ recommendProxyAddr = tmpProxyAddrList.get(0);
+ logger.info("No distribute data in registry,cluster:{},
group:{},purpose:{}, recommendProxyAddr:{}", cluster, group, purpose,
recommendProxyAddr);
+ return recommendProxyAddr;
+ }
+
+ Map<String, Integer> localClientDistributionMap = new HashMap<>();
+ Map<String, Integer> remoteClientDistributionMap = new HashMap<>();
+ for(Map.Entry<String, Map<String, Integer>> entry :
eventMeshClientDistributionDataMap.entrySet()){
+ String idc = entry.getKey().split("-")[0];
+ if(StringUtils.isNotBlank(idc)) {
+ if (StringUtils.equals(idc,
eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshIDC)) {
+ localClientDistributionMap.put(entry.getKey(),
entry.getValue().get(purpose));
+ } else {
+ remoteClientDistributionMap.put(entry.getKey(),
entry.getValue().get(purpose));
+ }
+ }else {
+ logger.error("eventMeshName may be illegal,idc is
null,eventMeshName:{}",entry.getKey());
+ }
+ }
+ recommendProxyAddr = recommendProxy(eventMeshMap, (caculateLocal ==
true) ? localClientDistributionMap : remoteClientDistributionMap, group);
+
logger.info("eventMeshMap:{},group:{},purpose:{},caculateLocal:{},recommendProxyAddr:{}",eventMeshMap,group,purpose,caculateLocal,recommendProxyAddr);
+ return recommendProxyAddr;
+ }
+
+ private String recommendProxy(Map<String, String> eventMeshMap,Map<String,
Integer> clientDistributionMap,String group){
+ logger.info("eventMeshMap:{},clientDistributionMap:{},group:{}",
eventMeshMap,clientDistributionMap,group);
+ String recommendProxy = null;
+
+ for(String proxyName : clientDistributionMap.keySet()){
+ if(!eventMeshMap.keySet().contains(proxyName)){
+ logger.warn("exist proxy not register but exist in
distributionMap,proxy:{}", proxyName);
+ return null;
+ }
+ }
+ for(String proxy : eventMeshMap.keySet()){
+ if(!clientDistributionMap.keySet().contains(proxy)){
+ clientDistributionMap.put(proxy, 0);
+ }
+ }
+
+ //select the eventmesh with least instances
+ List<Map.Entry<String, Integer>> list = new ArrayList<>();
+ ValueComparator vc = new ValueComparator();
+ for (Map.Entry<String, Integer> entry :
clientDistributionMap.entrySet()) {
+ list.add(entry);
+ }
+ if(list.size() == 0){
+ logger.error("no legal distribute data,check eventMeshMap and
distributeData, group:{}", group);
+ return null;
+ }else{
+ Collections.sort(list, vc);
+ logger.info("clientDistributionMap after sort:{}", list);
+ recommendProxy = eventMeshMap.get(list.get(0).getKey());
+ return recommendProxy;
+ }
+ }
+
+ private List<String> calculate(Map<String, String> proxyMap,Map<String,
Integer> clientDistributionMap,String group, int recommendProxyNum){
+ return null;
+ }
+}
diff --git a/settings.gradle
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/recommend/EventMeshRecommendStrategy.java
similarity index 63%
copy from settings.gradle
copy to
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/recommend/EventMeshRecommendStrategy.java
index 145df7e..3cc5044 100644
--- a/settings.gradle
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/recommend/EventMeshRecommendStrategy.java
@@ -14,17 +14,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.eventmesh.runtime.core.protocol.tcp.client.recommend;
-rootProject.name = 'EventMesh'
-String jdkVersion = "${jdk}"
-include 'eventmesh-runtime'
-include 'eventmesh-sdk-java'
-include 'eventmesh-common'
-include 'eventmesh-starter'
-include 'eventmesh-test'
-include 'eventmesh-spi'
-include 'eventmesh-connector-plugin:eventmesh-connector-api'
-include 'eventmesh-connector-plugin:eventmesh-connector-rocketmq'
-include 'eventmesh-security-plugin:eventmesh-security-api'
-include 'eventmesh-security-plugin:eventmesh-security-acl'
+import java.util.List;
+import java.util.Map;
+public interface EventMeshRecommendStrategy {
+ String calculateRecommendEventMesh(String group, String purpose) throws
Exception;
+
+ List<String> calculateRedirectRecommendEventMesh(Map<String, String>
eventMeshMap, Map<String, Integer> clientDistributeMap, String group, int
recommendNum) throws Exception;
+}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/RecommendTask.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/RecommendTask.java
new file mode 100644
index 0000000..ca626a8
--- /dev/null
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/RecommendTask.java
@@ -0,0 +1,101 @@
+/*
+ * 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.protocol.tcp.client.task;
+
+import io.netty.channel.ChannelHandlerContext;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.eventmesh.common.protocol.tcp.*;
+import org.apache.eventmesh.common.protocol.tcp.Package;
+import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import
org.apache.eventmesh.runtime.core.protocol.tcp.client.recommend.EventMeshRecommendImpl;
+import
org.apache.eventmesh.runtime.core.protocol.tcp.client.recommend.EventMeshRecommendStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static
org.apache.eventmesh.common.protocol.tcp.Command.RECOMMEND_RESPONSE;
+import static org.apache.eventmesh.runtime.util.Utils.writeAndFlush;
+
+public class RecommendTask extends AbstractTask {
+
+ private final Logger messageLogger = LoggerFactory.getLogger("message");
+
+ public RecommendTask(Package pkg, ChannelHandlerContext ctx, long
startTime, EventMeshTCPServer eventMeshTCPServer) {
+ super(pkg, ctx, startTime, eventMeshTCPServer);
+ }
+
+ @Override
+ public void run() {
+ long taskExecuteTime = System.currentTimeMillis();
+ Package res = new Package();
+ try {
+
if(!eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshServerRegistryEnable)
{
+ throw new Exception("registry enable config is false, not
support");
+ }
+ UserAgent user = (UserAgent) pkg.getBody();
+ validateUserAgent(user);
+ String group = getGroupOfClient(user);
+ EventMeshRecommendStrategy eventMeshRecommendStrategy = new
EventMeshRecommendImpl(eventMeshTCPServer);
+ String eventMeshRecommendResult =
eventMeshRecommendStrategy.calculateRecommendEventMesh(group,
user.getPurpose());
+ res.setHeader(new Header(RECOMMEND_RESPONSE,
OPStatus.SUCCESS.getCode(), OPStatus.SUCCESS.getDesc(),
pkg.getHeader().getSeq()));
+ res.setBody(eventMeshRecommendResult);
+ } catch (Exception e) {
+ messageLogger.error("RecommendTask failed|address={}|errMsg={}",
ctx.channel().remoteAddress(), e);
+ res.setHeader(new Header(RECOMMEND_RESPONSE,
OPStatus.FAIL.getCode(), e.toString(), pkg
+ .getHeader().getSeq()));
+
+ } finally {
+ writeAndFlush(res, startTime, taskExecuteTime,
session.getContext(), session);
+ //session.write2Client(res);
+ }
+ }
+
+ private void validateUserAgent(UserAgent user) throws Exception {
+ if (user == null) {
+ throw new Exception("client info cannot be null");
+ }
+
+ if (user.getVersion() == null) {
+ throw new Exception("client version cannot be null");
+ }
+
+ if (user.getUsername() == null) {
+ throw new Exception("client wemqUser cannot be null");
+ }
+
+ if (user.getPassword() == null) {
+ throw new Exception("client wemqPasswd cannot be null");
+ }
+
+ if (!(StringUtils.equals(EventMeshConstants.PURPOSE_PUB,
user.getPurpose()) || StringUtils.equals(EventMeshConstants.PURPOSE_SUB,
user.getPurpose()))) {
+ throw new Exception("client purpose config is error");
+ }
+ }
+
+ private String getGroupOfClient(UserAgent userAgent){
+ if(userAgent == null){
+ return null;
+ }
+ if(EventMeshConstants.PURPOSE_PUB.equals(userAgent.getPurpose())){
+ return userAgent.getProducerGroup();
+ }else
if(EventMeshConstants.PURPOSE_SUB.equals(userAgent.getPurpose())){
+ return userAgent.getConsumerGroup();
+ }else{
+ return null;
+ }
+ }
+}
\ No newline at end of file
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/tcp/EventMeshTcpMonitor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/tcp/EventMeshTcpMonitor.java
index 05a70b8..99b0f05 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/tcp/EventMeshTcpMonitor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/tcp/EventMeshTcpMonitor.java
@@ -144,6 +144,7 @@ public class EventMeshTcpMonitor {
@Override
public void run() {
// ThreadPoolHelper.printThreadPoolState();
+
eventMeshTCPServer.getEventMeshRebalanceService().printRebalanceThreadPoolState();
eventMeshTCPServer.getEventMeshTcpRetryer().printRetryThreadPoolState();
//monitor retry queue size
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/registry/Registry.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/registry/Registry.java
new file mode 100644
index 0000000..c8fe7ad
--- /dev/null
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/registry/Registry.java
@@ -0,0 +1,66 @@
+/*
+ * 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.registry;
+
+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.spi.EventMeshExtensionFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+
+public class Registry {
+ private static final Logger logger =
LoggerFactory.getLogger(Registry.class);
+ private static RegistryService registryService;
+
+ public void init(String registryPluginType) throws Exception {
+ registryService =
EventMeshExtensionFactory.getExtension(RegistryService.class,
registryPluginType);
+ if (registryService == null) {
+ logger.error("can't load the registryService plugin, please
check.");
+ throw new RuntimeException("doesn't load the registryService
plugin, please check.");
+ }
+ registryService.init();
+ }
+
+ public void start() throws Exception{
+ registryService.start();
+ }
+
+ public void shutdown() throws Exception{
+ registryService.shutdown();
+ }
+
+ public List<EventMeshDataInfo> findEventMeshInfoByCluster(String
clusterName) throws Exception{
+ return registryService.findEventMeshInfoByCluster(clusterName);
+ }
+
+ public Map<String, Map<String, Integer>>
findEventMeshClientDistributionData(String clusterName, String group, String
purpose) throws Exception{
+ return
registryService.findEventMeshClientDistributionData(clusterName, group,
purpose);
+ }
+
+ public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo)
throws Exception{
+ return registryService.register(eventMeshRegisterInfo);
+ }
+
+ public boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo)
throws Exception{
+ return registryService.unRegister(eventMeshUnRegisterInfo);
+ }
+}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/EventMeshUtil.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/EventMeshUtil.java
index 11542a2..970ef82 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/EventMeshUtil.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/EventMeshUtil.java
@@ -32,6 +32,7 @@ import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.TimeZone;
+import java.util.concurrent.ThreadPoolExecutor;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.databind.DeserializationFeature;
@@ -57,6 +58,8 @@ public class EventMeshUtil {
private static final RandomStringGenerator RANDOM_GENERATOR = new
RandomStringGenerator.Builder().withinRange('0', '9').build();
+ private final static Logger tcpLogger =
LoggerFactory.getLogger("tcpMonitor");
+
public static String buildPushMsgSeqNo() {
return
StringUtils.rightPad(String.valueOf(System.currentTimeMillis()), 6) +
RANDOM_GENERATOR.generate(4);
}
@@ -332,4 +335,11 @@ public class EventMeshUtil {
.append(client.getHost()).append(":").append(client.getPort());
return sb.toString();
}
+
+ public static void printState(ThreadPoolExecutor scheduledExecutorService)
{
+ tcpLogger.info("{} [{} {} {} {}]", ((EventMeshThreadFactoryImpl)
scheduledExecutorService.getThreadFactory())
+ .getThreadNamePrefix(),
scheduledExecutorService.getQueue().size(), scheduledExecutorService
+ .getPoolSize(), scheduledExecutorService.getActiveCount(),
scheduledExecutorService
+ .getCompletedTaskCount());
+ }
}
diff --git a/settings.gradle b/settings.gradle
index 145df7e..7dfce61 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -27,4 +27,6 @@ include 'eventmesh-connector-plugin:eventmesh-connector-api'
include 'eventmesh-connector-plugin:eventmesh-connector-rocketmq'
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-rocketmq-namesrv'
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
