This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch config-manage-improve
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/config-manage-improve by this
push:
new 3767ca4af EventMeshServer config improve
new d95cb9b08 Merge pull request #2631 from eight-nines/MR1
3767ca4af is described below
commit 3767ca4af606987923dbbe49f35ba1efce24b2ec
Author: eight-nines <[email protected]>
AuthorDate: Tue Dec 20 23:43:19 2022 +0800
EventMeshServer config improve
---
.../common/config/CommonConfiguration.java | 49 +++--
.../runtime/boot/EventMeshGrpcBootstrap.java | 9 +-
.../runtime/boot/EventMeshHttpBootstrap.java | 12 +-
.../eventmesh/runtime/boot/EventMeshServer.java | 22 +-
.../eventmesh/runtime/boot/EventMeshStartup.java | 13 +-
.../runtime/boot/EventMeshTcpBootstrap.java | 13 +-
.../configuration/EventMeshGrpcConfiguration.java | 28 ++-
.../configuration/EventMeshHTTPConfiguration.java | 152 +++++++------
.../configuration/EventMeshTCPConfiguration.java | 35 ++-
.../runtime/boot/EventMeshServerTest.java | 236 +++++++++++++++++++++
.../eventmesh/runtime/client/common/Server.java | 5 +-
.../resources/newConfiguration-runtime.properties | 105 +++++++++
.../admin/AdminWebHookConfigOperationManage.java | 1 +
13 files changed, 555 insertions(+), 125 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 d34d181ac..119078406 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
@@ -33,70 +33,69 @@ import com.google.common.base.Preconditions;
import lombok.Getter;
-@Config(prefix = "eventMesh")
public class CommonConfiguration {
- @ConfigFiled(field = "sysid")
+ @ConfigFiled(field = "eventMesh.sysid")
public String sysID = "5477";
- @ConfigFiled(field = "server.env")
+ @ConfigFiled(field = "eventMesh.server.env")
public String eventMeshEnv = "P";
- @ConfigFiled(field = "server.idc")
+ @ConfigFiled(field = "eventMesh.server.idc")
public String eventMeshIDC = "FT";
- @ConfigFiled(field = "server.name")
+ @ConfigFiled(field = "eventMesh.server.name")
public String eventMeshName = "";
- @ConfigFiled(field = "server.cluster")
+ @ConfigFiled(field = "eventMesh.server.cluster")
public String eventMeshCluster = "LS";
- @ConfigFiled(field = "server.hostIp")
+ @ConfigFiled(field = "eventMesh.server.hostIp")
public String eventMeshServerIp = null;
- @ConfigFiled(field = "registry.plugin.server-addr")
+ @ConfigFiled(field = "eventMesh.registry.plugin.server-addr")
public String namesrvAddr = "";
- @ConfigFiled(field = "trace.plugin")
+ @ConfigFiled(field = "eventMesh.trace.plugin")
public String eventMeshTracePluginType;
- @ConfigFiled(field = "metrics.plugin")
+ @ConfigFiled(field = "eventMesh.metrics.plugin")
public List<String> eventMeshMetricsPluginType;
- @ConfigFiled(field = "registry.plugin.type")
+ @ConfigFiled(field = "eventMesh.registry.plugin.type")
public String eventMeshRegistryPluginType = "namesrv";
- @ConfigFiled(field = "security.plugin.type")
+ @ConfigFiled(field = "eventMesh.security.plugin.type")
public String eventMeshSecurityPluginType = "security";
- @ConfigFiled(field = "connector.plugin.type")
+ @ConfigFiled(field = "eventMesh.connector.plugin.type")
public String eventMeshConnectorPluginType = "rocketmq";
- @ConfigFiled(field = "registry.plugin.username")
+ @ConfigFiled(field = "eventMesh.registry.plugin.username")
public String eventMeshRegistryPluginUsername = "";
- @ConfigFiled(field = "registry.plugin.password")
+ @ConfigFiled(field = "eventMesh.registry.plugin.password")
public String eventMeshRegistryPluginPassword = "";
- @ConfigFiled(field = "server.registry.registerIntervalInMills")
+ @ConfigFiled(field = "eventMesh.server.registry.registerIntervalInMills")
public Integer eventMeshRegisterIntervalInMills = 10 * 1000;
- @ConfigFiled(field = "server.registry.fetchRegistryAddrIntervalInMills")
+ @ConfigFiled(field =
"eventMesh.server.registry.fetchRegistryAddrIntervalInMills")
public Integer eventMeshFetchRegistryAddrInterval = 10 * 1000;
- @ConfigFiled(field = "server.trace.enabled")
+ @ConfigFiled(field = "eventMesh.server.trace.enabled")
public boolean eventMeshServerTraceEnable = false;
- @ConfigFiled(field = "server.security.enabled")
+ @ConfigFiled(field = "eventMesh.server.security.enabled")
public boolean eventMeshServerSecurityEnable = false;
- @ConfigFiled(field = "server.registry.enabled")
+ @ConfigFiled(field = "eventMesh.server.registry.enabled")
public boolean eventMeshServerRegistryEnable = false;
- @ConfigFiled(field = "server.provide.protocols")
+ @ConfigFiled(field = "eventMesh.server.provide.protocols")
public List<String> eventMeshProvideServerProtocols;
@@ -112,7 +111,7 @@ public class CommonConfiguration {
@Getter
protected ConfigurationWrapper configurationWrapper;
-
+
public CommonConfiguration(ConfigurationWrapper configurationWrapper) {
this.configurationWrapper = configurationWrapper;
}
@@ -197,9 +196,9 @@ public class CommonConfiguration {
String provideProtocols =
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_SERVER_PROVIDE_PROTOCOLS);
if (StringUtils.isNotEmpty(provideProtocols)) {
return Arrays.stream(provideProtocols.split(","))
- .filter(StringUtils::isNotBlank)
- .map(String::trim)
- .collect(Collectors.toList());
+ .filter(StringUtils::isNotBlank)
+ .map(String::trim)
+ .collect(Collectors.toList());
} else {
return Collections.singletonList(ConfigurationContextUtil.HTTP);
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshGrpcBootstrap.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshGrpcBootstrap.java
index f2d6a8116..138950335 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshGrpcBootstrap.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshGrpcBootstrap.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.runtime.boot;
+import org.apache.eventmesh.common.config.ConfigService;
import org.apache.eventmesh.common.config.ConfigurationWrapper;
import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
import org.apache.eventmesh.runtime.configuration.EventMeshGrpcConfiguration;
@@ -30,10 +31,12 @@ public class EventMeshGrpcBootstrap implements
EventMeshBootstrap {
private final Registry registry;
- public EventMeshGrpcBootstrap(ConfigurationWrapper configurationWrapper,
Registry registry) {
+ public EventMeshGrpcBootstrap(Registry registry) {
this.registry = registry;
- this.eventMeshGrpcConfiguration = new
EventMeshGrpcConfiguration(configurationWrapper);
- eventMeshGrpcConfiguration.init();
+
+ ConfigService configService = ConfigService.getInstance();
+ this.eventMeshGrpcConfiguration =
configService.getConfig(EventMeshGrpcConfiguration.class);
+
ConfigurationContextUtil.putIfAbsent(ConfigurationContextUtil.GRPC,
eventMeshGrpcConfiguration);
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHttpBootstrap.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHttpBootstrap.java
index 5e0323e70..6d80aa9a5 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHttpBootstrap.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHttpBootstrap.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.runtime.boot;
+import org.apache.eventmesh.common.config.ConfigService;
import org.apache.eventmesh.common.config.ConfigurationWrapper;
import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
@@ -32,15 +33,14 @@ public class EventMeshHttpBootstrap implements
EventMeshBootstrap {
private final Registry registry;
- public EventMeshHttpBootstrap(EventMeshServer eventMeshServer,
- ConfigurationWrapper configurationWrapper,
- Registry registry) {
+ public EventMeshHttpBootstrap(EventMeshServer eventMeshServer, Registry
registry) {
this.eventMeshServer = eventMeshServer;
this.registry = registry;
- this.eventMeshHttpConfiguration = new
EventMeshHTTPConfiguration(configurationWrapper);
- eventMeshHttpConfiguration.init();
- ConfigurationContextUtil.putIfAbsent(ConfigurationContextUtil.HTTP,
eventMeshHttpConfiguration);
+ ConfigService configService = ConfigService.getInstance();
+ this.eventMeshHttpConfiguration =
configService.getConfig(EventMeshHTTPConfiguration.class);
+
+ ConfigurationContextUtil.putIfAbsent(ConfigurationContextUtil.HTTP,
eventMeshHttpConfiguration);
}
@Override
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 602ca78ca..449fbeb9c 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
@@ -18,10 +18,13 @@
package org.apache.eventmesh.runtime.boot;
import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.common.config.ConfigService;
import org.apache.eventmesh.common.config.ConfigurationWrapper;
import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
import org.apache.eventmesh.runtime.acl.Acl;
import org.apache.eventmesh.runtime.common.ServiceState;
+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;
@@ -51,28 +54,25 @@ public class EventMeshServer {
private static final List<EventMeshBootstrap> BOOTSTRAP_LIST = new
CopyOnWriteArrayList<>();
- public EventMeshServer(ConfigurationWrapper configurationWrapper) {
- CommonConfiguration configuration = new
CommonConfiguration(configurationWrapper);
- configuration.init();
- this.configuration = configuration;
+ public EventMeshServer() {
+ ConfigService configService = ConfigService.getInstance();
+ this.configuration =
configService.getConfig(CommonConfiguration.class);
+
this.acl = new Acl();
this.registry = new Registry();
- trace = new Trace(configuration.eventMeshServerTraceEnable);
this.connectorResource = new ConnectorResource();
+ trace = new Trace(configuration.eventMeshServerTraceEnable);
List<String> provideServerProtocols =
configuration.eventMeshProvideServerProtocols;
for (String provideServerProtocol : provideServerProtocols) {
if (ConfigurationContextUtil.HTTP.equals(provideServerProtocol)) {
- BOOTSTRAP_LIST.add(new EventMeshHttpBootstrap(this,
- configurationWrapper, registry));
+ BOOTSTRAP_LIST.add(new EventMeshHttpBootstrap(this, registry));
}
if (ConfigurationContextUtil.TCP.equals(provideServerProtocol)) {
- BOOTSTRAP_LIST.add(new EventMeshTcpBootstrap(this,
- configurationWrapper, registry));
+ BOOTSTRAP_LIST.add(new EventMeshTcpBootstrap(this, registry));
}
if (ConfigurationContextUtil.GRPC.equals(provideServerProtocol)) {
- BOOTSTRAP_LIST.add(new
EventMeshGrpcBootstrap(configurationWrapper,
- registry));
+ BOOTSTRAP_LIST.add(new EventMeshGrpcBootstrap(registry));
}
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshStartup.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshStartup.java
index 34ec72201..72816fa96 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshStartup.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshStartup.java
@@ -17,9 +17,11 @@
package org.apache.eventmesh.runtime.boot;
-import org.apache.eventmesh.common.config.ConfigurationWrapper;
+import org.apache.eventmesh.common.config.ConfigService;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import java.io.File;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,10 +31,11 @@ public class EventMeshStartup {
public static void main(String[] args) throws Exception {
try {
- ConfigurationWrapper configurationWrapper =
- new
ConfigurationWrapper(EventMeshConstants.EVENTMESH_CONF_HOME,
- EventMeshConstants.EVENTMESH_CONF_FILE, false);
- EventMeshServer server = new EventMeshServer(configurationWrapper);
+ ConfigService.getInstance()
+ .setConfigPath(EventMeshConstants.EVENTMESH_CONF_HOME +
File.separator)
+ .setRootConfig(EventMeshConstants.EVENTMESH_CONF_FILE);
+
+ EventMeshServer server = new EventMeshServer();
server.init();
server.start();
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTcpBootstrap.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTcpBootstrap.java
index 743e2c805..f0fe9584b 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTcpBootstrap.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshTcpBootstrap.java
@@ -17,8 +17,10 @@
package org.apache.eventmesh.runtime.boot;
+import org.apache.eventmesh.common.config.ConfigService;
import org.apache.eventmesh.common.config.ConfigurationWrapper;
import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
+import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
import org.apache.eventmesh.runtime.configuration.EventMeshTCPConfiguration;
import org.apache.eventmesh.runtime.registry.Registry;
@@ -32,15 +34,14 @@ public class EventMeshTcpBootstrap implements
EventMeshBootstrap {
private final Registry registry;
- public EventMeshTcpBootstrap(EventMeshServer eventMeshServer,
- ConfigurationWrapper configurationWrapper,
- Registry registry) {
+ public EventMeshTcpBootstrap(EventMeshServer eventMeshServer, Registry
registry) {
this.eventMeshServer = eventMeshServer;
this.registry = registry;
- this.eventMeshTcpConfiguration = new
EventMeshTCPConfiguration(configurationWrapper);
- eventMeshTcpConfiguration.init();
- ConfigurationContextUtil.putIfAbsent(ConfigurationContextUtil.TCP,
eventMeshTcpConfiguration);
+ ConfigService configService = ConfigService.getInstance();
+ this.eventMeshTcpConfiguration =
configService.getConfig(EventMeshTCPConfiguration.class);
+
+ ConfigurationContextUtil.putIfAbsent(ConfigurationContextUtil.TCP,
eventMeshTcpConfiguration);
}
@Override
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshGrpcConfiguration.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshGrpcConfiguration.java
index 65142f632..c65212c0e 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshGrpcConfiguration.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshGrpcConfiguration.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.runtime.configuration;
import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.common.config.ConfigFiled;
import org.apache.eventmesh.common.config.ConfigurationWrapper;
import org.apache.eventmesh.common.utils.IPUtils;
@@ -27,56 +28,79 @@ import com.google.common.base.Preconditions;
public class EventMeshGrpcConfiguration extends CommonConfiguration {
+ @ConfigFiled(field = "eventMesh.server.grpc.port")
public int grpcServerPort = 10205;
+ @ConfigFiled(field = "eventMesh.server.session.expiredInMills")
public int eventMeshSessionExpiredInMills = 60000;
+ @ConfigFiled(field = "eventMesh.server.batchmsg.batch.enabled")
public boolean eventMeshServerBatchMsgBatchEnabled = Boolean.TRUE;
+ @ConfigFiled(field = "eventMesh.server.batchmsg.threads.num")
public int eventMeshServerBatchMsgThreadNum = 10;
+ @ConfigFiled(field = "eventMesh.server.sendmsg.threads.num")
public int eventMeshServerSendMsgThreadNum = 8;
+ @ConfigFiled(field = "eventMesh.server.pushmsg.threads.num")
public int eventMeshServerPushMsgThreadNum = 8;
+ @ConfigFiled(field = "eventMesh.server.replymsg.threads.num")
public int eventMeshServerReplyMsgThreadNum = 8;
+ @ConfigFiled(field = "eventMesh.server.clientmanage.threads.num")
public int eventMeshServerSubscribeMsgThreadNum = 4;
+ @ConfigFiled(field = "eventMesh.server.registry.threads.num")
public int eventMeshServerRegistryThreadNum = 10;
+ @ConfigFiled(field = "eventMesh.server.admin.threads.num")
public int eventMeshServerAdminThreadNum = 2;
+ @ConfigFiled(field = "eventMesh.server.retry.threads.num")
public int eventMeshServerRetryThreadNum = 2;
+ @ConfigFiled(field = "eventMesh.server.pull.registry.interval")
public int eventMeshServerPullRegistryInterval = 30000;
+ @ConfigFiled(field = "eventMesh.server.async.accumulation.threshold")
public int eventMeshServerAsyncAccumulationThreshold = 1000;
+ @ConfigFiled(field = "eventMesh.server.retry.blockQ.size")
public int eventMeshServerRetryBlockQueueSize = 10000;
+ @ConfigFiled(field = "eventMesh.server.batchmsg.blockQ.size")
public int eventMeshServerBatchBlockQueueSize = 1000;
+ @ConfigFiled(field = "eventMesh.server.sendmsg.blockQ.size")
public int eventMeshServerSendMsgBlockQueueSize = 1000;
+ @ConfigFiled(field = "eventMesh.server.pushmsg.blockQ.size")
public int eventMeshServerPushMsgBlockQueueSize = 1000;
+ @ConfigFiled(field = "eventMesh.server.clientM.blockQ.size")
public int eventMeshServerSubscribeMsgBlockQueueSize = 1000;
+ @ConfigFiled(field = "eventMesh.server.busy.check.interval")
public int eventMeshServerBusyCheckInterval = 1000;
+ @ConfigFiled(field = "eventMesh.server.consumer.enabled")
public boolean eventMeshServerConsumerEnabled = false;
+ @ConfigFiled(field = "eventMesh.server.useTls.enabled")
public boolean eventMeshServerUseTls = false;
+ @ConfigFiled(field = "eventMesh.server.batchmsg.reqNumPerSecond")
public int eventMeshBatchMsgRequestNumPerSecond = 20000;
+ @ConfigFiled(field = "eventMesh.server.http.msgReqnumPerSecond")
public int eventMeshMsgReqNumPerSecond = 15000;
public String eventMeshIp = IPUtils.getLocalAddress();
- public EventMeshGrpcConfiguration(ConfigurationWrapper
configurationWrapper) {
- super(configurationWrapper);
+ public EventMeshGrpcConfiguration() {
+
}
@Override
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshHTTPConfiguration.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshHTTPConfiguration.java
index f0415a7fa..30918b88f 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshHTTPConfiguration.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshHTTPConfiguration.java
@@ -18,6 +18,8 @@
package org.apache.eventmesh.runtime.configuration;
import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.common.config.Config;
+import org.apache.eventmesh.common.config.ConfigFiled;
import org.apache.eventmesh.common.config.ConfigurationWrapper;
import org.apache.commons.lang3.StringUtils;
@@ -39,72 +41,104 @@ public class EventMeshHTTPConfiguration extends
CommonConfiguration {
public static final Logger logger =
LoggerFactory.getLogger(EventMeshHTTPConfiguration.class);
+ @ConfigFiled(field = "eventMesh.server.http.port")
public int httpServerPort = 10105;
+ @ConfigFiled(field = "eventMesh.server.batchmsg.batch.enabled")
public boolean eventMeshServerBatchMsgBatchEnabled = Boolean.TRUE;
+ @ConfigFiled(field = "eventMesh.server.batchmsg.threads.num")
public int eventMeshServerBatchMsgThreadNum = 10;
+ @ConfigFiled(field = "eventMesh.server.sendmsg.threads.num")
public int eventMeshServerSendMsgThreadNum = 8;
+ @ConfigFiled(field = "eventMesh.server.remotemsg.threads.num")
public int eventMeshServerRemoteMsgThreadNum = 8;
+ @ConfigFiled(field = "eventMesh.server.pushmsg.threads.num")
public int eventMeshServerPushMsgThreadNum = 8;
+ @ConfigFiled(field = "eventMesh.server.replymsg.threads.num")
public int eventMeshServerReplyMsgThreadNum = 8;
+ @ConfigFiled(field = "eventMesh.server.clientmanage.threads.num")
public int eventMeshServerClientManageThreadNum = 4;
+ @ConfigFiled(field = "eventMesh.server.registry.threads.num")
public int eventMeshServerRegistryThreadNum = 10;
+ @ConfigFiled(field = "eventMesh.server.admin.threads.num")
public int eventMeshServerAdminThreadNum = 2;
+ @ConfigFiled(field = "eventMesh.server.retry.threads.num")
public int eventMeshServerRetryThreadNum = 2;
-
+
+ @ConfigFiled(field = "")
public int eventMeshServerWebhookThreadNum = 4;
+ @ConfigFiled(field = "eventMesh.server.pull.registry.interval")
public int eventMeshServerPullRegistryInterval = 30000;
+ @ConfigFiled(field = "eventMesh.server.async.accumulation.threshold")
public int eventMeshServerAsyncAccumulationThreshold = 1000;
+ @ConfigFiled(field = "eventMesh.server.retry.blockQ.size")
public int eventMeshServerRetryBlockQSize = 10000;
+ @ConfigFiled(field = "eventMesh.server.batchmsg.blockQ.size")
public int eventMeshServerBatchBlockQSize = 1000;
+ @ConfigFiled(field = "eventMesh.server.sendmsg.blockQ.size")
public int eventMeshServerSendMsgBlockQSize = 1000;
+ @ConfigFiled(field = "")
public int eventMeshServerRemoteMsgBlockQSize = 1000;
+ @ConfigFiled(field = "eventMesh.server.pushmsg.blockQ.size")
public int eventMeshServerPushMsgBlockQSize = 1000;
+ @ConfigFiled(field = "eventMesh.server.clientM.blockQ.size")
public int eventMeshServerClientManageBlockQSize = 1000;
+ @ConfigFiled(field = "eventMesh.server.busy.check.interval")
public int eventMeshServerBusyCheckInterval = 1000;
+ @ConfigFiled(field = "eventMesh.server.consumer.enabled")
public boolean eventMeshServerConsumerEnabled = false;
+ @ConfigFiled(field = "eventMesh.server.useTls.enabled")
public boolean eventMeshServerUseTls = false;
+ @ConfigFiled(field = "eventMesh.server.ssl.protocol")
public String eventMeshServerSSLProtocol = "TLSv1.1";
+ @ConfigFiled(field = "eventMesh.server.ssl.cer")
public String eventMeshServerSSLCer = "sChat2.jks";
+ @ConfigFiled(field = "eventMesh.server.ssl.pass")
public String eventMeshServerSSLPass = "sNetty";
+ @ConfigFiled(field = "eventMesh.server.http.msgReqnumPerSecond")
public int eventMeshHttpMsgReqNumPerSecond = 15000;
+ @ConfigFiled(field = "eventMesh.server.batchmsg.reqNumPerSecond")
public int eventMeshBatchMsgRequestNumPerSecond = 20000;
+ @ConfigFiled(field = "eventMesh.server.maxEventSize")
public int eventMeshEventSize = 1000;
+ @ConfigFiled(field = "eventMesh.server.maxEventBatchSize")
public int eventMeshEventBatchSize = 10;
+ @ConfigFiled(field = "eventMesh.server.blacklist.ipv4")
public List<IPAddress> eventMeshIpv4BlackList = Collections.emptyList();
+ @ConfigFiled(field = "eventMesh.server.blacklist.ipv6")
public List<IPAddress> eventMeshIpv6BlackList = Collections.emptyList();
- public EventMeshHTTPConfiguration(ConfigurationWrapper
configurationWrapper) {
- super(configurationWrapper);
+ public EventMeshHTTPConfiguration() {
+
}
@Override
@@ -114,44 +148,44 @@ public class EventMeshHTTPConfiguration extends
CommonConfiguration {
if (configurationWrapper != null) {
String httpServerPortStr =
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_SERVER_HTTP_PORT);
Preconditions.checkState(StringUtils.isNotEmpty(httpServerPortStr)
- && StringUtils.isNumeric(httpServerPortStr),
String.format("%s error", ConfKeys.KEYS_EVENTMESH_SERVER_HTTP_PORT));
+ && StringUtils.isNumeric(httpServerPortStr), String.format("%s
error", ConfKeys.KEYS_EVENTMESH_SERVER_HTTP_PORT));
httpServerPort =
Integer.parseInt(StringUtils.deleteWhitespace(httpServerPortStr));
String eventMeshServerBatchMsgThreadNumStr =
-
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_BATCHMSG_THREAD_NUM);
+
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_BATCHMSG_THREAD_NUM);
if (StringUtils.isNotEmpty(eventMeshServerBatchMsgThreadNumStr)
- &&
StringUtils.isNumeric(eventMeshServerBatchMsgThreadNumStr)) {
+ && StringUtils.isNumeric(eventMeshServerBatchMsgThreadNumStr))
{
eventMeshServerBatchMsgThreadNum =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerBatchMsgThreadNumStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerBatchMsgThreadNumStr));
}
String eventMeshServerBatchMsgReqNumPerSecondStr =
-
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_BATCHMSG_REQ_NUM_PER_SECOND);
+
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_BATCHMSG_REQ_NUM_PER_SECOND);
if
(StringUtils.isNotEmpty(eventMeshServerBatchMsgReqNumPerSecondStr)
- &&
StringUtils.isNumeric(eventMeshServerBatchMsgReqNumPerSecondStr)) {
+ &&
StringUtils.isNumeric(eventMeshServerBatchMsgReqNumPerSecondStr)) {
eventMeshBatchMsgRequestNumPerSecond =
Integer.parseInt(eventMeshServerBatchMsgReqNumPerSecondStr);
}
String eventMeshServerBatchMsgBatchEnableStr =
-
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_BATCHMSG_BATCH_ENABLED);
+
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_BATCHMSG_BATCH_ENABLED);
if (StringUtils.isNotBlank(eventMeshServerBatchMsgBatchEnableStr))
{
eventMeshServerBatchMsgBatchEnabled =
Boolean.parseBoolean(eventMeshServerBatchMsgBatchEnableStr);
}
String eventMeshServerAsyncAccumulationThresholdStr =
-
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_ASYNC_ACCUMULATION_THRESHOLD);
+
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_ASYNC_ACCUMULATION_THRESHOLD);
if
(StringUtils.isNotEmpty(eventMeshServerAsyncAccumulationThresholdStr)
- &&
StringUtils.isNumeric(eventMeshServerAsyncAccumulationThresholdStr)) {
+ &&
StringUtils.isNumeric(eventMeshServerAsyncAccumulationThresholdStr)) {
eventMeshServerAsyncAccumulationThreshold =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerAsyncAccumulationThresholdStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerAsyncAccumulationThresholdStr));
}
String eventMeshServerSendMsgThreadNumStr =
-
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_SENDMSG_THREAD_NUM);
+
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_SENDMSG_THREAD_NUM);
if (StringUtils.isNotEmpty(eventMeshServerSendMsgThreadNumStr)
- &&
StringUtils.isNumeric(eventMeshServerSendMsgThreadNumStr)) {
+ && StringUtils.isNumeric(eventMeshServerSendMsgThreadNumStr)) {
eventMeshServerSendMsgThreadNum =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerSendMsgThreadNumStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerSendMsgThreadNumStr));
}
String eventMeshServerRemoteMsgThreadNumStr =
@@ -163,116 +197,116 @@ public class EventMeshHTTPConfiguration extends
CommonConfiguration {
}
String eventMeshServerReplyMsgThreadNumStr =
-
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_REPLYMSG_THREAD_NUM);
+
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_REPLYMSG_THREAD_NUM);
if (StringUtils.isNotEmpty(eventMeshServerReplyMsgThreadNumStr)
- &&
StringUtils.isNumeric(eventMeshServerReplyMsgThreadNumStr)) {
+ && StringUtils.isNumeric(eventMeshServerReplyMsgThreadNumStr))
{
eventMeshServerReplyMsgThreadNum =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerReplyMsgThreadNumStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerReplyMsgThreadNumStr));
}
String eventMeshServerPushMsgThreadNumStr =
-
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_PUSHMSG_THREAD_NUM);
+
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_PUSHMSG_THREAD_NUM);
if (StringUtils.isNotEmpty(eventMeshServerPushMsgThreadNumStr)
- &&
StringUtils.isNumeric(eventMeshServerPushMsgThreadNumStr)) {
+ && StringUtils.isNumeric(eventMeshServerPushMsgThreadNumStr)) {
eventMeshServerPushMsgThreadNum =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerPushMsgThreadNumStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerPushMsgThreadNumStr));
}
String eventMeshServerRegistryThreadNumStr =
-
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_REGISTRY_THREAD_NUM);
+
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_REGISTRY_THREAD_NUM);
if (StringUtils.isNotEmpty(eventMeshServerRegistryThreadNumStr)
- &&
StringUtils.isNumeric(eventMeshServerRegistryThreadNumStr)) {
+ && StringUtils.isNumeric(eventMeshServerRegistryThreadNumStr))
{
eventMeshServerRegistryThreadNum =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerRegistryThreadNumStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerRegistryThreadNumStr));
}
String eventMeshServerClientManageThreadNumStr =
-
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_CLIENTMANAGE_THREAD_NUM);
+
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_CLIENTMANAGE_THREAD_NUM);
if (StringUtils.isNotEmpty(eventMeshServerClientManageThreadNumStr)
- &&
StringUtils.isNumeric(eventMeshServerClientManageThreadNumStr)) {
+ &&
StringUtils.isNumeric(eventMeshServerClientManageThreadNumStr)) {
eventMeshServerClientManageThreadNum =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerClientManageThreadNumStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerClientManageThreadNumStr));
}
String eventMeshServerPullRegistryIntervalStr =
-
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_PULL_REGISTRY_INTERVAL);
+
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_PULL_REGISTRY_INTERVAL);
if (StringUtils.isNotEmpty(eventMeshServerPullRegistryIntervalStr)
- &&
StringUtils.isNumeric(eventMeshServerPullRegistryIntervalStr)) {
+ &&
StringUtils.isNumeric(eventMeshServerPullRegistryIntervalStr)) {
eventMeshServerPullRegistryInterval =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerPullRegistryIntervalStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerPullRegistryIntervalStr));
}
String eventMeshServerAdminThreadNumStr =
-
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_ADMIN_THREAD_NUM);
+
configurationWrapper.getProp(ConfKeys.KEYS_EVENTMESH_ADMIN_THREAD_NUM);
if (StringUtils.isNotEmpty(eventMeshServerAdminThreadNumStr)
- &&
StringUtils.isNumeric(eventMeshServerAdminThreadNumStr)) {
+ && StringUtils.isNumeric(eventMeshServerAdminThreadNumStr)) {
eventMeshServerAdminThreadNum =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerAdminThreadNumStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerAdminThreadNumStr));
}
String eventMeshServerRetryBlockQSizeStr =
-
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_RETRY_BLOCKQ_SIZE);
+
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_RETRY_BLOCKQ_SIZE);
if (StringUtils.isNotEmpty(eventMeshServerRetryBlockQSizeStr)
- &&
StringUtils.isNumeric(eventMeshServerRetryBlockQSizeStr)) {
+ && StringUtils.isNumeric(eventMeshServerRetryBlockQSizeStr)) {
eventMeshServerRetryBlockQSize =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerRetryBlockQSizeStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerRetryBlockQSizeStr));
}
String eventMeshServerBatchBlockQSizeStr =
-
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_BATCHMSG_BLOCKQ_SIZE);
+
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_BATCHMSG_BLOCKQ_SIZE);
if (StringUtils.isNotEmpty(eventMeshServerBatchBlockQSizeStr)
- &&
StringUtils.isNumeric(eventMeshServerBatchBlockQSizeStr)) {
+ && StringUtils.isNumeric(eventMeshServerBatchBlockQSizeStr)) {
eventMeshServerBatchBlockQSize =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerBatchBlockQSizeStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerBatchBlockQSizeStr));
}
String eventMeshServerSendMsgBlockQSizeStr =
-
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_SENDMSG_BLOCKQ_SIZE);
+
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_SENDMSG_BLOCKQ_SIZE);
if (StringUtils.isNotEmpty(eventMeshServerSendMsgBlockQSizeStr)
- &&
StringUtils.isNumeric(eventMeshServerSendMsgBlockQSizeStr)) {
+ && StringUtils.isNumeric(eventMeshServerSendMsgBlockQSizeStr))
{
eventMeshServerSendMsgBlockQSize =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerSendMsgBlockQSizeStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerSendMsgBlockQSizeStr));
}
String eventMeshServerPushMsgBlockQSizeStr =
-
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_PUSHMSG_BLOCKQ_SIZE);
+
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_PUSHMSG_BLOCKQ_SIZE);
if (StringUtils.isNotEmpty(eventMeshServerPushMsgBlockQSizeStr)
- &&
StringUtils.isNumeric(eventMeshServerPushMsgBlockQSizeStr)) {
+ && StringUtils.isNumeric(eventMeshServerPushMsgBlockQSizeStr))
{
eventMeshServerPushMsgBlockQSize =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerPushMsgBlockQSizeStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerPushMsgBlockQSizeStr));
}
String eventMeshServerClientManageBlockQSizeStr =
-
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_CLIENTM_BLOCKQ_SIZE);
+
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_CLIENTM_BLOCKQ_SIZE);
if
(StringUtils.isNotEmpty(eventMeshServerClientManageBlockQSizeStr)
- &&
StringUtils.isNumeric(eventMeshServerClientManageBlockQSizeStr)) {
+ &&
StringUtils.isNumeric(eventMeshServerClientManageBlockQSizeStr)) {
eventMeshServerClientManageBlockQSize =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerClientManageBlockQSizeStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerClientManageBlockQSizeStr));
}
String eventMeshServerBusyCheckIntervalStr =
-
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_BUSY_CHECK_INTERVAL);
+
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_BUSY_CHECK_INTERVAL);
if (StringUtils.isNotEmpty(eventMeshServerBusyCheckIntervalStr)
- &&
StringUtils.isNumeric(eventMeshServerBusyCheckIntervalStr)) {
+ && StringUtils.isNumeric(eventMeshServerBusyCheckIntervalStr))
{
eventMeshServerBusyCheckInterval =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerBusyCheckIntervalStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerBusyCheckIntervalStr));
}
String eventMeshServerConsumerEnabledStr =
-
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_CONSUMER_ENABLED);
+
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_CONSUMER_ENABLED);
if (StringUtils.isNotEmpty(eventMeshServerConsumerEnabledStr)) {
eventMeshServerConsumerEnabled =
-
Boolean.parseBoolean(StringUtils.deleteWhitespace(eventMeshServerConsumerEnabledStr));
+
Boolean.parseBoolean(StringUtils.deleteWhitespace(eventMeshServerConsumerEnabledStr));
}
String eventMeshServerRetryThreadNumStr =
-
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_RETRY_THREAD_NUM);
+
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_RETRY_THREAD_NUM);
if (StringUtils.isNotEmpty(eventMeshServerRetryThreadNumStr)
- &&
StringUtils.isNumeric(eventMeshServerRetryThreadNumStr)) {
+ && StringUtils.isNumeric(eventMeshServerRetryThreadNumStr)) {
eventMeshServerRetryThreadNum =
-
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerRetryThreadNumStr));
+
Integer.parseInt(StringUtils.deleteWhitespace(eventMeshServerRetryThreadNumStr));
}
@@ -297,9 +331,9 @@ public class EventMeshHTTPConfiguration extends
CommonConfiguration {
}
String eventMeshHttpMsgReqNumPerSecondStr =
-
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_SERVER_MSG_REQ_NUM_PER_SECOND);
+
configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_SERVER_MSG_REQ_NUM_PER_SECOND);
if (StringUtils.isNotEmpty(eventMeshHttpMsgReqNumPerSecondStr)
- &&
StringUtils.isNumeric(eventMeshHttpMsgReqNumPerSecondStr)) {
+ && StringUtils.isNumeric(eventMeshHttpMsgReqNumPerSecondStr)) {
eventMeshHttpMsgReqNumPerSecond =
Integer.parseInt(eventMeshHttpMsgReqNumPerSecondStr);
}
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 04db23a90..0da7760a1 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
@@ -18,69 +18,96 @@
package org.apache.eventmesh.runtime.configuration;
import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.common.config.ConfigFiled;
import org.apache.eventmesh.common.config.ConfigurationWrapper;
+import org.apache.eventmesh.common.config.NotNull;
public class EventMeshTCPConfiguration extends CommonConfiguration {
+ @ConfigFiled(field = "eventMesh.server.tcp.port")
public int eventMeshTcpServerPort = 10000;
+ @ConfigFiled(field = "eventMesh.server.tcp.allIdleSeconds")
public int eventMeshTcpIdleAllSeconds = 60;
+ @ConfigFiled(field = "eventMesh.server.tcp.writerIdleSeconds")
public int eventMeshTcpIdleWriteSeconds = 60;
+ @ConfigFiled(field = "eventMesh.server.tcp.readerIdleSeconds")
public int eventMeshTcpIdleReadSeconds = 60;
+ @ConfigFiled(field = "eventMesh.server.tcp.msgReqnumPerSecond")
public Integer eventMeshTcpMsgReqnumPerSecond = 15000;
/**
* TCP Server allows max client num
*/
+ @ConfigFiled(field = "eventMesh.server.tcp.clientMaxNum")
public int eventMeshTcpClientMaxNum = 10000;
//======================================= New add config
=================================
+ @ConfigFiled(field = "eventMesh.server.global.scheduler")
public int eventMeshTcpGlobalScheduler = 5;
+ @ConfigFiled(field = "eventMesh.server.tcp.taskHandleExecutorPoolSize")
public int eventMeshTcpTaskHandleExecutorPoolSize =
Runtime.getRuntime().availableProcessors();
+ @ConfigFiled(field = "eventMesh.server.tcp.msgDownStreamExecutorPoolSize")
public int eventMeshTcpMsgDownStreamExecutorPoolSize =
Math.max(Runtime.getRuntime().availableProcessors(), 8);
+ @ConfigFiled(field = "eventMesh.server.session.expiredInMills")
public int eventMeshTcpSessionExpiredInMills = 60000;
+ @ConfigFiled(field = "eventMesh.server.session.upstreamBufferSize")
public int eventMeshTcpSessionUpstreamBufferSize = 100;
+ @ConfigFiled(field = "eventMesh.server.retry.async.pushRetryTimes")
public int eventMeshTcpMsgAsyncRetryTimes = 3;
+ @ConfigFiled(field = "eventMesh.server.retry.sync.pushRetryTimes")
public int eventMeshTcpMsgSyncRetryTimes = 1;
+ @ConfigFiled(field = "eventMesh.server.retry.sync.pushRetryDelayInMills")
public int eventMeshTcpMsgRetrySyncDelayInMills = 500;
+ @ConfigFiled(field = "eventMesh.server.retry.async.pushRetryDelayInMills")
public int eventMeshTcpMsgRetryAsyncDelayInMills = 500;
+ @ConfigFiled(field = "eventMesh.server.retry.pushRetryQueueSize")
public int eventMeshTcpMsgRetryQueueSize = 10000;
+ @ConfigFiled(field = "eventMesh.server.tcp.RebalanceIntervalInMills")
public Integer eventMeshTcpRebalanceIntervalInMills = 30 * 1000;
+ @ConfigFiled(field = "eventMesh.server.admin.http.port")
public int eventMeshServerAdminPort = 10106;
+ @ConfigFiled(field = "eventMesh.server.tcp.sendBack.enabled")
public boolean eventMeshTcpSendBackEnabled = Boolean.TRUE;
+ @ConfigFiled(field = "")
public int eventMeshTcpSendBackMaxTimes = 3;
+ @ConfigFiled(field = "eventMesh.server.tcp.pushFailIsolateTimeInMills")
public int eventMeshTcpPushFailIsolateTimeInMills = 30 * 1000;
+ @ConfigFiled(field =
"eventMesh.server.gracefulShutdown.sleepIntervalInMills")
public int gracefulShutdownSleepIntervalInMills = 1000;
+ @ConfigFiled(field = "eventMesh.server.rebalanceRedirect.sleepIntervalInM")
public int sleepIntervalInRebalanceRedirectMills = 200;
+ @ConfigFiled(field = "eventMesh.server.maxEventSize")
public int eventMeshEventSize = 1000;
+ @ConfigFiled(field = "eventMesh.server.maxEventBatchSize")
public int eventMeshEventBatchSize = 10;
- private TrafficShapingConfig gtc = new TrafficShapingConfig(0, 10_000,
1_000, 2000);
- private TrafficShapingConfig ctc = new TrafficShapingConfig(0, 2_000,
1_000, 10_000);
+ private final TrafficShapingConfig gtc = new TrafficShapingConfig(0,
10_000, 1_000, 2000);
+ private final TrafficShapingConfig ctc = new TrafficShapingConfig(0,
2_000, 1_000, 10_000);
+
+ public EventMeshTCPConfiguration() {
- public EventMeshTCPConfiguration(ConfigurationWrapper
configurationWrapper) {
- super(configurationWrapper);
}
@Override
diff --git
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/boot/EventMeshServerTest.java
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/boot/EventMeshServerTest.java
new file mode 100644
index 000000000..e869ed066
--- /dev/null
+++
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/boot/EventMeshServerTest.java
@@ -0,0 +1,236 @@
+/*
+ * 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.boot;
+
+import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.common.config.ConfigService;
+import org.apache.eventmesh.runtime.configuration.EventMeshGrpcConfiguration;
+import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
+import org.apache.eventmesh.runtime.configuration.EventMeshTCPConfiguration;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import inet.ipaddr.AddressStringException;
+import inet.ipaddr.IPAddress;
+import inet.ipaddr.IPAddressString;
+
+
+public class EventMeshServerTest {
+ public static Logger logger =
LoggerFactory.getLogger(EventMeshServerTest.class);
+
+ @Test
+ public void testGetConfigForEventMeshHTTPConfiguration() throws Exception {
+
+ ConfigService configService = ConfigService.getInstance();
+
configService.setRootConfig("classPath://newConfiguration-runtime.properties");
+
+ EventMeshHTTPConfiguration config =
configService.getConfig(EventMeshHTTPConfiguration.class);
+
+ assertCommonConfig(config);
+
+ assertHTTPConfig(config);
+ }
+
+ private void assertHTTPConfig(EventMeshHTTPConfiguration config) throws
AddressStringException {
+ Assert.assertEquals(config.httpServerPort, 1816);
+ Assert.assertEquals(config.eventMeshServerBatchMsgBatchEnabled,
Boolean.FALSE);
+ Assert.assertEquals(config.eventMeshServerBatchMsgThreadNum, 2816);
+ Assert.assertEquals(config.eventMeshServerSendMsgThreadNum, 3816);
+ Assert.assertEquals(config.eventMeshServerPushMsgThreadNum, 4816);
+ Assert.assertEquals(config.eventMeshServerReplyMsgThreadNum, 5816);
+ Assert.assertEquals(config.eventMeshServerClientManageThreadNum, 6816);
+ Assert.assertEquals(config.eventMeshServerRegistryThreadNum, 7816);
+ Assert.assertEquals(config.eventMeshServerAdminThreadNum, 8816);
+
+ Assert.assertEquals(config.eventMeshServerRetryThreadNum, 9816);
+ Assert.assertEquals(config.eventMeshServerPullRegistryInterval, 11816);
+ Assert.assertEquals(config.eventMeshServerAsyncAccumulationThreshold,
12816);
+ Assert.assertEquals(config.eventMeshServerRetryBlockQSize, 13816);
+ Assert.assertEquals(config.eventMeshServerBatchBlockQSize, 14816);
+ Assert.assertEquals(config.eventMeshServerSendMsgBlockQSize, 15816);
+ Assert.assertEquals(config.eventMeshServerPushMsgBlockQSize, 16816);
+ Assert.assertEquals(config.eventMeshServerClientManageBlockQSize,
17816);
+ Assert.assertEquals(config.eventMeshServerBusyCheckInterval, 18816);
+ Assert.assertEquals(config.eventMeshServerConsumerEnabled,
Boolean.TRUE);
+ Assert.assertEquals(config.eventMeshServerUseTls, Boolean.TRUE);
+ Assert.assertEquals(config.eventMeshHttpMsgReqNumPerSecond, 19816);
+ Assert.assertEquals(config.eventMeshBatchMsgRequestNumPerSecond,
21816);
+ Assert.assertEquals(config.eventMeshEventSize, 22816);
+ Assert.assertEquals(config.eventMeshEventBatchSize, 23816);
+
+ List<IPAddress> list4 = new ArrayList<>();
+ list4.add(new IPAddressString("127.0.0.1").toAddress());
+ list4.add(new IPAddressString("127.0.0.2").toAddress());
+ Assert.assertEquals(config.eventMeshIpv4BlackList, list4);
+ List<IPAddress> list6 = new ArrayList<>();
+ list6.add(new IPAddressString("0:0:0:0:0:0:7f00:01").toAddress());
+ list6.add(new IPAddressString("0:0:0:0:0:0:7f00:02").toAddress());
+ Assert.assertEquals(config.eventMeshIpv6BlackList, list6);
+ }
+
+ @Test
+ public void testGetConfigForEventMeshGrpcConfiguration() throws Exception {
+
+ ConfigService configService = ConfigService.getInstance();
+
configService.setRootConfig("classPath://newConfiguration-runtime.properties");
+
+ EventMeshGrpcConfiguration config =
configService.getConfig(EventMeshGrpcConfiguration.class);
+
+ assertCommonConfig(config);
+
+ assertGrpcConfig(config);
+ }
+
+ private void assertGrpcConfig(EventMeshGrpcConfiguration config) {
+ Assert.assertEquals(config.grpcServerPort, 816);
+ Assert.assertEquals(config.eventMeshSessionExpiredInMills, 1816);
+ Assert.assertEquals(config.eventMeshServerBatchMsgBatchEnabled,
Boolean.FALSE);
+ Assert.assertEquals(config.eventMeshServerBatchMsgThreadNum, 2816);
+ Assert.assertEquals(config.eventMeshServerSendMsgThreadNum, 3816);
+ Assert.assertEquals(config.eventMeshServerPushMsgThreadNum, 4816);
+ Assert.assertEquals(config.eventMeshServerReplyMsgThreadNum, 5816);
+ Assert.assertEquals(config.eventMeshServerSubscribeMsgThreadNum, 6816);
+ Assert.assertEquals(config.eventMeshServerRegistryThreadNum, 7816);
+ Assert.assertEquals(config.eventMeshServerAdminThreadNum, 8816);
+ Assert.assertEquals(config.eventMeshServerRetryThreadNum, 9816);
+ Assert.assertEquals(config.eventMeshServerPullRegistryInterval, 11816);
+ Assert.assertEquals(config.eventMeshServerAsyncAccumulationThreshold,
12816);
+ Assert.assertEquals(config.eventMeshServerRetryBlockQueueSize, 13816);
+ Assert.assertEquals(config.eventMeshServerBatchBlockQueueSize, 14816);
+ Assert.assertEquals(config.eventMeshServerSendMsgBlockQueueSize,
15816);
+ Assert.assertEquals(config.eventMeshServerPushMsgBlockQueueSize,
16816);
+ Assert.assertEquals(config.eventMeshServerSubscribeMsgBlockQueueSize,
17816);
+ Assert.assertEquals(config.eventMeshServerBusyCheckInterval, 18816);
+ Assert.assertEquals(config.eventMeshServerConsumerEnabled,
Boolean.TRUE);
+ Assert.assertEquals(config.eventMeshServerUseTls, Boolean.TRUE);
+ Assert.assertEquals(config.eventMeshBatchMsgRequestNumPerSecond,
21816);
+ Assert.assertEquals(config.eventMeshMsgReqNumPerSecond, 19816);
+ }
+
+ @Test
+ public void testGetConfigForEventMeshTCPConfiguration() throws Exception {
+
+ ConfigService configService = ConfigService.getInstance();
+
configService.setRootConfig("classPath://newConfiguration-runtime.properties");
+
+ EventMeshTCPConfiguration config =
configService.getConfig(EventMeshTCPConfiguration.class);
+
+ assertCommonConfig(config);
+
+ assertTCPConfig(config);
+ }
+
+ private void assertTCPConfig(EventMeshTCPConfiguration config) {
+ Assert.assertEquals(config.eventMeshTcpServerPort, 816);
+ Assert.assertEquals(config.eventMeshTcpIdleAllSeconds, 1816);
+ Assert.assertEquals(config.eventMeshTcpIdleWriteSeconds, 2816);
+ Assert.assertEquals(config.eventMeshTcpIdleReadSeconds, 3816);
+ Assert.assertEquals(config.eventMeshTcpMsgReqnumPerSecond,
Integer.valueOf(4816));
+ Assert.assertEquals(config.eventMeshTcpClientMaxNum, 5816);
+ Assert.assertEquals(config.eventMeshTcpGlobalScheduler, 6816);
+ Assert.assertEquals(config.eventMeshTcpTaskHandleExecutorPoolSize,
7816);
+ Assert.assertEquals(config.eventMeshTcpMsgDownStreamExecutorPoolSize,
8816);
+ Assert.assertEquals(config.eventMeshTcpSessionExpiredInMills, 1816);
+ Assert.assertEquals(config.eventMeshTcpSessionUpstreamBufferSize,
11816);
+ Assert.assertEquals(config.eventMeshTcpMsgAsyncRetryTimes, 12816);
+ Assert.assertEquals(config.eventMeshTcpMsgSyncRetryTimes, 13816);
+ Assert.assertEquals(config.eventMeshTcpMsgRetrySyncDelayInMills,
14816);
+ Assert.assertEquals(config.eventMeshTcpMsgRetryAsyncDelayInMills,
15816);
+ Assert.assertEquals(config.eventMeshTcpMsgRetryQueueSize, 16816);
+ Assert.assertEquals(config.eventMeshTcpRebalanceIntervalInMills,
Integer.valueOf(17816));
+ Assert.assertEquals(config.eventMeshServerAdminPort, 18816);
+ Assert.assertEquals(config.eventMeshTcpSendBackEnabled, Boolean.TRUE);
+ Assert.assertEquals(config.eventMeshTcpSendBackMaxTimes, 3);
+ Assert.assertEquals(config.eventMeshTcpPushFailIsolateTimeInMills,
21816);
+ Assert.assertEquals(config.gracefulShutdownSleepIntervalInMills,
22816);
+ Assert.assertEquals(config.sleepIntervalInRebalanceRedirectMills,
23816);
+ Assert.assertEquals(config.eventMeshEventSize, 22816);
+ Assert.assertEquals(config.eventMeshEventBatchSize, 23816);
+ }
+
+ private void assertCommonConfig(CommonConfiguration config) {
+ Assert.assertEquals(config.eventMeshEnv, "env-succeed!!!");
+ Assert.assertEquals(config.eventMeshIDC, "idc-succeed!!!");
+ Assert.assertEquals(config.eventMeshCluster, "cluster-succeed!!!");
+ Assert.assertEquals(config.eventMeshName, "name-succeed!!!");
+ Assert.assertEquals(config.sysID, "sysid-succeed!!!");
+ Assert.assertEquals(config.eventMeshConnectorPluginType,
"connector-succeed!!!");
+ Assert.assertEquals(config.eventMeshSecurityPluginType,
"security-succeed!!!");
+ Assert.assertEquals(config.eventMeshRegistryPluginType,
"registry-succeed!!!");
+ Assert.assertEquals(config.eventMeshTracePluginType,
"trace-succeed!!!");
+ Assert.assertEquals(config.eventMeshServerIp, "hostIp-succeed!!!");
+
+ List<String> list = new ArrayList<>();
+ list.add("metrics-succeed1!!!");
+ list.add("metrics-succeed2!!!");
+ list.add("metrics-succeed3!!!");
+ Assert.assertEquals(config.eventMeshMetricsPluginType, list);
+
+ Assert.assertEquals(config.eventMeshServerSecurityEnable,
Boolean.TRUE);
+ Assert.assertEquals(config.eventMeshServerRegistryEnable,
Boolean.TRUE);
+ Assert.assertEquals(config.eventMeshServerTraceEnable, Boolean.TRUE);
+
+ Assert.assertEquals(config.eventMeshWebhookOrigin,
"eventmesh.idc-succeed!!!");
+ }
+
+
+ /**
+ * True Environment variables need to be set during startup
+ */
+ @Test
+ public void testGetConfigWhenStartup() throws Exception {
+
+ testGetConfigWhenStartup(Boolean.FALSE);
+ }
+
+ private void testGetConfigWhenStartup(Boolean hasEnv) throws Exception {
+ String eventMeshConfFile = "newConfiguration-runtime.properties";
+
+ if (hasEnv) {
+ ConfigService.getInstance()
+ .setConfigPath(EventMeshConstants.EVENTMESH_CONF_HOME +
File.separator)
+ .setRootConfig(eventMeshConfFile);
+ } else {
+ eventMeshConfFile = "classPath://" + eventMeshConfFile;
+ ConfigService.getInstance().setRootConfig(eventMeshConfFile);
+ }
+
+ ConfigService configService = ConfigService.getInstance();
+ CommonConfiguration commonConfiguration =
configService.getConfig(CommonConfiguration.class);
+ EventMeshHTTPConfiguration eventMeshHttpConfiguration =
configService.getConfig(EventMeshHTTPConfiguration.class);
+ EventMeshTCPConfiguration eventMeshTCPConfiguration =
configService.getConfig(EventMeshTCPConfiguration.class);
+ EventMeshGrpcConfiguration eventMeshGrpcConfiguration =
configService.getConfig(EventMeshGrpcConfiguration.class);
+
+ assertCommonConfig(eventMeshTCPConfiguration);
+ assertCommonConfig(eventMeshHttpConfiguration);
+ assertCommonConfig(eventMeshGrpcConfiguration);
+
+ assertTCPConfig(eventMeshTCPConfiguration);
+ assertHTTPConfig(eventMeshHttpConfiguration);
+ assertGrpcConfig(eventMeshGrpcConfiguration);
+ }
+
+}
\ No newline at end of file
diff --git
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/common/Server.java
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/common/Server.java
index 8a5d9b68b..76f0fe550 100644
---
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/common/Server.java
+++
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/client/common/Server.java
@@ -33,10 +33,7 @@ public class Server {
}
public void startAccessServer() throws Exception {
- ConfigurationWrapper configurationWrapper =
- new
ConfigurationWrapper(EventMeshConstants.EVENTMESH_CONF_HOME,
- EventMeshConstants.EVENTMESH_CONF_FILE, false);
- server = new EventMeshServer(configurationWrapper);
+ server = new EventMeshServer();
server.init();
server.start();
}
diff --git
a/eventmesh-runtime/src/test/resources/newConfiguration-runtime.properties
b/eventmesh-runtime/src/test/resources/newConfiguration-runtime.properties
new file mode 100644
index 000000000..b132e4f96
--- /dev/null
+++ b/eventmesh-runtime/src/test/resources/newConfiguration-runtime.properties
@@ -0,0 +1,105 @@
+#
+# 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.
+#
+
+# CommonConfiguration config
+eventMesh.server.env=env-succeed!!!
+eventMesh.server.idc=idc-succeed!!!
+eventMesh.sysid=sysid-succeed!!!
+eventMesh.server.cluster=cluster-succeed!!!
+eventMesh.server.name=name-succeed!!!
+eventMesh.server.hostIp=hostIp-succeed!!!
+eventMesh.connector.plugin.type=connector-succeed!!!
+eventMesh.security.plugin.type=security-succeed!!!
+eventMesh.registry.plugin.type=registry-succeed!!!
+eventMesh.trace.plugin=trace-succeed!!!
+eventMesh.server.registry.registerIntervalInMills=816
+eventMesh.server.registry.fetchRegistryAddrIntervalInMills=1816
+eventMesh.metrics.plugin=metrics-succeed1!!!,metrics-succeed2!!!,metrics-succeed3!!!
+
+eventMesh.server.security.enabled=true
+eventMesh.server.registry.enabled=true
+eventMesh.server.trace.enabled=true
+
+eventMesh.server.provide.protocols=TCP,HTTP,GRPC
+eventMesh.registry.plugin.username=username-succeed!!!
+eventMesh.registry.plugin.password=password-succeed!!!
+
+# EventMeshHTTPConfiguration config
+eventMesh.server.http.port=1816
+eventMesh.server.batchmsg.batch.enabled=false
+eventMesh.server.batchmsg.threads.num=2816
+eventMesh.server.sendmsg.threads.num=3816
+eventMesh.server.pushmsg.threads.num=4816
+eventMesh.server.replymsg.threads.num=5816
+eventMesh.server.clientmanage.threads.num=6816
+eventMesh.server.registry.threads.num=7816
+eventMesh.server.admin.threads.num=8816
+eventMesh.server.retry.threads.num=9816
+eventMesh.server.pull.registry.interval=11816
+eventMesh.server.async.accumulation.threshold=12816
+eventMesh.server.retry.blockQ.size=13816
+eventMesh.server.batchmsg.blockQ.size=14816
+eventMesh.server.sendmsg.blockQ.size=15816
+eventMesh.server.pushmsg.blockQ.size=16816
+eventMesh.server.clientM.blockQ.size=17816
+eventMesh.server.busy.check.interval=18816
+eventMesh.server.consumer.enabled=true
+eventMesh.server.useTls.enabled=true
+eventMesh.server.http.msgReqnumPerSecond=19816
+eventMesh.server.batchmsg.reqNumPerSecond=21816
+eventMesh.server.maxEventSize=22816
+eventMesh.server.maxEventBatchSize=23816
+eventMesh.server.blacklist.ipv4=127.0.0.1,127.0.0.2
+eventMesh.server.blacklist.ipv6=0:0:0:0:0:0:7f00:01,0:0:0:0:0:0:7f00:02
+
+# EventMeshGrpcConfiguration config
+eventMesh.server.grpc.port=816
+eventMesh.server.session.expiredInMills=1816
+
+
+# EventMeshTCPConfiguration config
+eventMesh.server.tcp.port=816
+eventMesh.server.tcp.allIdleSeconds=1816
+eventMesh.server.tcp.writerIdleSeconds=2816
+eventMesh.server.tcp.readerIdleSeconds=3816
+eventMesh.server.tcp.msgReqnumPerSecond=4816
+eventMesh.server.tcp.clientMaxNum=5816
+eventMesh.server.tcp.enabled=true
+eventMesh.server.global.scheduler=6816
+eventMesh.server.tcp.taskHandleExecutorPoolSize=7816
+eventMesh.server.tcp.msgDownStreamExecutorPoolSize=8816
+eventMesh.server.session.upstreamBufferSize=11816
+eventMesh.server.retry.async.pushRetryTimes=12816
+eventMesh.server.retry.sync.pushRetryTimes=13816
+eventMesh.server.retry.sync.pushRetryDelayInMills=14816
+eventMesh.server.retry.async.pushRetryDelayInMills=15816
+eventMesh.server.retry.pushRetryQueueSize=16816
+eventMesh.server.tcp.RebalanceIntervalInMills=17816
+eventMesh.server.admin.http.port=18816
+eventMesh.server.tcp.sendBack.enabled=true
+eventMesh.server.tcp.pushFailIsolateTimeInMills=21816
+eventMesh.server.gracefulShutdown.sleepIntervalInMills=22816
+eventMesh.server.rebalanceRedirect.sleepIntervalInM=23816
+
+
+
+
+
+
+
+
+
diff --git
a/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/AdminWebHookConfigOperationManage.java
b/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/AdminWebHookConfigOperationManage.java
index 5148d19d4..51b74cf12 100644
---
a/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/AdminWebHookConfigOperationManage.java
+++
b/eventmesh-webhook/eventmesh-webhook-admin/src/main/java/org/apache/eventmesh/webhook/admin/AdminWebHookConfigOperationManage.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.webhook.admin;
+import org.apache.eventmesh.common.config.ConfigService;
import org.apache.eventmesh.common.config.ConfigurationWrapper;
import org.apache.eventmesh.webhook.api.WebHookConfigOperation;
import org.apache.eventmesh.webhook.api.WebHookOperationConstant;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
