You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by mi...@apache.org on 2021/11/19 02:14:38 UTC
[incubator-eventmesh] branch cloudevents updated: Remove some unused code (#591)
This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch cloudevents
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/cloudevents by this push:
new db02781 Remove some unused code (#591)
db02781 is described below
commit db0278134793a1777e2efa785d7f66b0f2b0823a
Author: Wenjun Ruan <we...@apache.org>
AuthorDate: Fri Nov 19 10:14:31 2021 +0800
Remove some unused code (#591)
* Remove some unused code in sdk module
* Remove some unused code in sdk module
---
.../eventmesh/common/EventMeshException.java | 7 +-
.../eventmesh/common/ProtocolTransportObject.java | 12 +
.../eventmesh/common/command/HttpCommand.java | 15 +-
.../eventmesh/common/protocol/tcp/Package.java | 4 +-
.../eventmesh-protocol-api/build.gradle | 6 +
.../eventmesh/protocol/api/ProtocolAdaptor.java | 5 +-
.../protocol/api/ProtocolPluginFactory.java | 14 +-
.../cloudevents/CloudEventsProtocolAdaptor.java | 16 +-
.../openmessage/OpenMessageProtocolAdaptor.java | 9 +-
.../eventmesh/runtime/boot/AbstractHTTPServer.java | 321 +++++++++-------
.../runtime/boot/EventMeshHTTPServer.java | 88 +++--
.../http/processor/AdminMetricsProcessor.java | 2 -
.../http/processor/AdminShutdownProcessor.java | 15 +-
.../processor/BatchSendMessageV2Processor.java | 215 +++++++----
.../http/processor/HeartBeatProcessor.java | 83 ++--
.../http/processor/ReplyMessageProcessor.java | 3 +-
.../http/processor/SendAsyncMessageProcessor.java | 3 +-
.../http/processor/SendSyncMessageProcessor.java | 76 ++--
.../http/processor/SubscribeProcessor.java | 27 +-
.../http/processor/UnSubscribeProcessor.java | 10 +-
.../protocol/http/push/AsyncHTTPPushRequest.java | 11 +-
.../tcp/client/task/MessageTransferTask.java | 13 +-
eventmesh-sdk-java/build.gradle | 6 +
.../eventmesh/client/http/AbstractLiteClient.java | 9 +-
.../eventmesh/client/http/RemotingServer.java | 424 ---------------------
.../client/http/consumer/LiteConsumer.java | 123 +++---
.../eventmesh/client/http/http/RequestParam.java | 10 +-
.../client/http/producer/LiteProducer.java | 64 ++--
style/checkStyle.xml | 2 +-
29 files changed, 651 insertions(+), 942 deletions(-)
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/EventMeshException.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/EventMeshException.java
index bffa57e..620bdf9 100644
--- a/eventmesh-common/src/main/java/org/apache/eventmesh/common/EventMeshException.java
+++ b/eventmesh-common/src/main/java/org/apache/eventmesh/common/EventMeshException.java
@@ -34,13 +34,12 @@ public class EventMeshException extends Exception {
super(cause);
}
- public EventMeshException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+ public EventMeshException(String message, Throwable cause, boolean enableSuppression,
+ boolean writableStackTrace) {
super(message, cause, enableSuppression, writableStackTrace);
}
public EventMeshException(Integer errCode, String errMsg) {
- super((new StringBuilder()).append(errCode)
- .append("|")
- .append(errMsg).toString());
+ super(String.format("errorCode: %s, errorMessage: %s", errCode, errMsg));
}
}
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/ProtocolTransportObject.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/ProtocolTransportObject.java
new file mode 100644
index 0000000..cf660a3
--- /dev/null
+++ b/eventmesh-common/src/main/java/org/apache/eventmesh/common/ProtocolTransportObject.java
@@ -0,0 +1,12 @@
+package org.apache.eventmesh.common;
+
+import java.io.Serializable;
+
+/**
+ * <ul>
+ * <li>Tcp transport object{@link org.apache.eventmesh.common.protocol.tcp.Package}</li>
+ * <li>Http transport object{@link org.apache.eventmesh.common.command.HttpCommand}</li>
+ * </ul>
+ */
+public interface ProtocolTransportObject extends Serializable {
+}
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/command/HttpCommand.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/command/HttpCommand.java
index b6c61d6..031b993 100644
--- a/eventmesh-common/src/main/java/org/apache/eventmesh/common/command/HttpCommand.java
+++ b/eventmesh-common/src/main/java/org/apache/eventmesh/common/command/HttpCommand.java
@@ -18,8 +18,10 @@
package org.apache.eventmesh.common.command;
import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.ProtocolTransportObject;
import org.apache.eventmesh.common.protocol.http.body.BaseResponseBody;
import org.apache.eventmesh.common.protocol.http.body.Body;
+import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
import org.apache.eventmesh.common.protocol.http.header.BaseResponseHeader;
import org.apache.eventmesh.common.protocol.http.header.Header;
import org.apache.eventmesh.common.utils.JsonUtils;
@@ -38,9 +40,9 @@ import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
-public class HttpCommand {
+public class HttpCommand implements ProtocolTransportObject {
- private static AtomicLong requestId = new AtomicLong(0);
+ private static final AtomicLong requestId = new AtomicLong(0);
private long opaque;
@@ -63,8 +65,7 @@ public class HttpCommand {
public CmdType cmdType = CmdType.REQ;
public HttpCommand() {
- this.reqTime = System.currentTimeMillis();
- this.opaque = requestId.incrementAndGet();
+ this(null, null, null);
}
public HttpCommand(String httpMethod, String httpVersion, String requestCode) {
@@ -90,7 +91,7 @@ public class HttpCommand {
return response;
}
- public HttpCommand createHttpCommandResponse(Integer retCode, String retMsg) {
+ public HttpCommand createHttpCommandResponse(EventMeshRetCode eventMeshRetCode) {
if (StringUtils.isBlank(requestCode)) {
return null;
}
@@ -101,8 +102,8 @@ public class HttpCommand {
baseResponseHeader.setCode(requestCode);
response.setHeader(baseResponseHeader);
BaseResponseBody baseResponseBody = new BaseResponseBody();
- baseResponseBody.setRetCode(retCode);
- baseResponseBody.setRetMsg(retMsg);
+ baseResponseBody.setRetCode(eventMeshRetCode.getRetCode());
+ baseResponseBody.setRetMsg(eventMeshRetCode.getErrMsg());
response.setBody(baseResponseBody);
response.setCmdType(CmdType.RES);
response.setResTime(System.currentTimeMillis());
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/tcp/Package.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/tcp/Package.java
index d2cde02..d108817 100644
--- a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/tcp/Package.java
+++ b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/tcp/Package.java
@@ -17,7 +17,9 @@
package org.apache.eventmesh.common.protocol.tcp;
-public class Package {
+import org.apache.eventmesh.common.ProtocolTransportObject;
+
+public class Package implements ProtocolTransportObject {
private Header header;
private Object body;
diff --git a/eventmesh-protocol-plugin/eventmesh-protocol-api/build.gradle b/eventmesh-protocol-plugin/eventmesh-protocol-api/build.gradle
index 8eb049d..9445967 100644
--- a/eventmesh-protocol-plugin/eventmesh-protocol-api/build.gradle
+++ b/eventmesh-protocol-plugin/eventmesh-protocol-api/build.gradle
@@ -26,4 +26,10 @@ dependencies {
testImplementation "io.cloudevents:cloudevents-core"
testImplementation "junit:junit"
+
+ compileOnly 'org.projectlombok:lombok:1.18.22'
+ annotationProcessor 'org.projectlombok:lombok:1.18.22'
+
+ testCompileOnly 'org.projectlombok:lombok:1.18.22'
+ testAnnotationProcessor 'org.projectlombok:lombok:1.18.22'
}
\ No newline at end of file
diff --git a/eventmesh-protocol-plugin/eventmesh-protocol-api/src/main/java/org/apache/eventmesh/protocol/api/ProtocolAdaptor.java b/eventmesh-protocol-plugin/eventmesh-protocol-api/src/main/java/org/apache/eventmesh/protocol/api/ProtocolAdaptor.java
index 9505df9..e8b854e 100644
--- a/eventmesh-protocol-plugin/eventmesh-protocol-api/src/main/java/org/apache/eventmesh/protocol/api/ProtocolAdaptor.java
+++ b/eventmesh-protocol-plugin/eventmesh-protocol-api/src/main/java/org/apache/eventmesh/protocol/api/ProtocolAdaptor.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.protocol.api;
+import org.apache.eventmesh.common.ProtocolTransportObject;
import org.apache.eventmesh.common.protocol.tcp.Package;
import org.apache.eventmesh.protocol.api.exception.ProtocolHandleException;
import org.apache.eventmesh.spi.EventMeshExtensionType;
@@ -35,7 +36,7 @@ import java.util.List;
* @since 1.3.0
*/
@EventMeshSPI(isSingleton = true, eventMeshExtensionType = EventMeshExtensionType.PROTOCOL)
-public interface ProtocolAdaptor<T> {
+public interface ProtocolAdaptor<T extends ProtocolTransportObject> {
/**
* transform protocol to {@link CloudEvent}.
@@ -59,7 +60,7 @@ public interface ProtocolAdaptor<T> {
* @param cloudEvent clout event
* @return target protocol
*/
- Object fromCloudEvent(CloudEvent cloudEvent) throws ProtocolHandleException;
+ ProtocolTransportObject fromCloudEvent(CloudEvent cloudEvent) throws ProtocolHandleException;
/**
* Get protocol type.
diff --git a/eventmesh-protocol-plugin/eventmesh-protocol-api/src/main/java/org/apache/eventmesh/protocol/api/ProtocolPluginFactory.java b/eventmesh-protocol-plugin/eventmesh-protocol-api/src/main/java/org/apache/eventmesh/protocol/api/ProtocolPluginFactory.java
index f9208ee..fd0de78 100644
--- a/eventmesh-protocol-plugin/eventmesh-protocol-api/src/main/java/org/apache/eventmesh/protocol/api/ProtocolPluginFactory.java
+++ b/eventmesh-protocol-plugin/eventmesh-protocol-api/src/main/java/org/apache/eventmesh/protocol/api/ProtocolPluginFactory.java
@@ -17,20 +17,23 @@
package org.apache.eventmesh.protocol.api;
+import org.apache.eventmesh.common.ProtocolTransportObject;
import org.apache.eventmesh.spi.EventMeshExtensionFactory;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import lombok.experimental.UtilityClass;
+
/**
* A factory to get Protocol plugin instance.
*
* @since 1.3.0
*/
-public enum ProtocolPluginFactory {
- ;
+@UtilityClass
+public class ProtocolPluginFactory {
- private static final Map<String, ProtocolAdaptor> PROTOCOL_ADAPTOR_MAP =
+ private static final Map<String, ProtocolAdaptor<ProtocolTransportObject>> PROTOCOL_ADAPTOR_MAP =
new ConcurrentHashMap<>(16);
/**
@@ -40,8 +43,9 @@ public enum ProtocolPluginFactory {
* @return protocol adaptor
* @throws IllegalArgumentException if protocol not found
*/
- public static ProtocolAdaptor getProtocolAdaptor(String protocolType) {
- ProtocolAdaptor protocolAdaptor = PROTOCOL_ADAPTOR_MAP.computeIfAbsent(
+ @SuppressWarnings("unchecked")
+ public static ProtocolAdaptor<ProtocolTransportObject> getProtocolAdaptor(String protocolType) {
+ ProtocolAdaptor<ProtocolTransportObject> protocolAdaptor = PROTOCOL_ADAPTOR_MAP.computeIfAbsent(
protocolType,
(type) -> EventMeshExtensionFactory.getExtension(ProtocolAdaptor.class, type)
);
diff --git a/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/CloudEventsProtocolAdaptor.java b/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/CloudEventsProtocolAdaptor.java
index 52c5321..b9bb8e2 100644
--- a/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/CloudEventsProtocolAdaptor.java
+++ b/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/CloudEventsProtocolAdaptor.java
@@ -20,13 +20,13 @@ package org.apache.eventmesh.protocol.cloudevents;
import io.cloudevents.CloudEvent;
import org.apache.commons.lang3.StringUtils;
import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.ProtocolTransportObject;
import org.apache.eventmesh.common.command.HttpCommand;
import org.apache.eventmesh.common.protocol.http.body.Body;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.protocol.tcp.Header;
import org.apache.eventmesh.common.protocol.tcp.Package;
import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
-
import org.apache.eventmesh.protocol.api.exception.ProtocolHandleException;
import org.apache.eventmesh.protocol.cloudevents.resolver.http.SendMessageBatchProtocolResolver;
import org.apache.eventmesh.protocol.cloudevents.resolver.http.SendMessageBatchV2ProtocolResolver;
@@ -41,10 +41,11 @@ import java.util.List;
*
* @since 1.3.0
*/
-public class CloudEventsProtocolAdaptor<T> implements ProtocolAdaptor<T> {
+public class CloudEventsProtocolAdaptor<T extends ProtocolTransportObject>
+ implements ProtocolAdaptor<ProtocolTransportObject> {
@Override
- public CloudEvent toCloudEvent(T cloudEvent) throws ProtocolHandleException {
+ public CloudEvent toCloudEvent(ProtocolTransportObject cloudEvent) throws ProtocolHandleException {
if (cloudEvent instanceof Package) {
Header header = ((Package) cloudEvent).getHeader();
@@ -84,15 +85,18 @@ public class CloudEventsProtocolAdaptor<T> implements ProtocolAdaptor<T> {
}
@Override
- public List<CloudEvent> toBatchCloudEvent(T protocol) throws ProtocolHandleException {
+ public List<CloudEvent> toBatchCloudEvent(ProtocolTransportObject protocol)
+ throws ProtocolHandleException {
return null;
}
@Override
- public Object fromCloudEvent(CloudEvent cloudEvent) throws ProtocolHandleException {
+ public ProtocolTransportObject fromCloudEvent(CloudEvent cloudEvent) throws ProtocolHandleException {
String protocolDesc = cloudEvent.getExtension(Constants.PROTOCOL_DESC).toString();
if (StringUtils.equals("http", protocolDesc)) {
- return new String(cloudEvent.getData().toBytes(), StandardCharsets.UTF_8);
+ // todo: return command, set cloudEvent.getData() to content?
+ return null;
+// return new String(cloudEvent.getData().toBytes(), StandardCharsets.UTF_8);
} else if (StringUtils.equals("tcp", protocolDesc)) {
Package pkg = new Package();
pkg.setBody(cloudEvent);
diff --git a/eventmesh-protocol-plugin/eventmesh-protocol-openmessage/src/main/java/org/apache/eventmesh/protocol/openmessage/OpenMessageProtocolAdaptor.java b/eventmesh-protocol-plugin/eventmesh-protocol-openmessage/src/main/java/org/apache/eventmesh/protocol/openmessage/OpenMessageProtocolAdaptor.java
index 6e00275..1492a28 100644
--- a/eventmesh-protocol-plugin/eventmesh-protocol-openmessage/src/main/java/org/apache/eventmesh/protocol/openmessage/OpenMessageProtocolAdaptor.java
+++ b/eventmesh-protocol-plugin/eventmesh-protocol-openmessage/src/main/java/org/apache/eventmesh/protocol/openmessage/OpenMessageProtocolAdaptor.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.protocol.openmessage;
+import org.apache.eventmesh.common.ProtocolTransportObject;
import org.apache.eventmesh.common.protocol.tcp.Package;
import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
@@ -33,20 +34,20 @@ import java.util.List;
*
* @since 1.3.0
*/
-public class OpenMessageProtocolAdaptor<T> implements ProtocolAdaptor<T> {
+public class OpenMessageProtocolAdaptor<T extends ProtocolTransportObject> implements ProtocolAdaptor<ProtocolTransportObject> {
@Override
- public CloudEvent toCloudEvent(T message) {
+ public CloudEvent toCloudEvent(ProtocolTransportObject message) {
return null;
}
@Override
- public List<CloudEvent> toBatchCloudEvent(T protocol) throws ProtocolHandleException {
+ public List<CloudEvent> toBatchCloudEvent(ProtocolTransportObject protocol) throws ProtocolHandleException {
return null;
}
@Override
- public Object fromCloudEvent(CloudEvent cloudEvent) {
+ public ProtocolTransportObject fromCloudEvent(CloudEvent cloudEvent) {
return null;
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
index 83cba95..9b08724 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractHTTPServer.java
@@ -17,16 +17,39 @@
package org.apache.eventmesh.runtime.boot;
+import org.apache.eventmesh.common.ThreadPoolFactory;
+import org.apache.eventmesh.common.command.HttpCommand;
+import org.apache.eventmesh.common.protocol.http.body.Body;
+import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolVersion;
+import org.apache.eventmesh.common.protocol.http.common.RequestCode;
+import org.apache.eventmesh.common.protocol.http.header.Header;
+import org.apache.eventmesh.runtime.common.Pair;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
+import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.HttpRequestProcessor;
+import org.apache.eventmesh.runtime.metrics.http.HTTPMetricsServer;
+import org.apache.eventmesh.runtime.util.RemotingHelper;
+
+import org.apache.commons.collections4.MapUtils;
+import org.apache.commons.lang3.ObjectUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.IOException;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import com.google.common.base.Preconditions;
import io.netty.bootstrap.ServerBootstrap;
@@ -44,6 +67,7 @@ import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpRequest;
@@ -61,27 +85,6 @@ import io.netty.handler.ssl.SslHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
-import org.apache.commons.collections4.MapUtils;
-import org.apache.commons.lang3.ObjectUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.eventmesh.common.ThreadPoolFactory;
-import org.apache.eventmesh.common.command.HttpCommand;
-import org.apache.eventmesh.common.protocol.http.body.Body;
-import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
-import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
-import org.apache.eventmesh.common.protocol.http.common.ProtocolVersion;
-import org.apache.eventmesh.common.protocol.http.common.RequestCode;
-import org.apache.eventmesh.common.protocol.http.header.Header;
-import org.apache.eventmesh.runtime.common.Pair;
-import org.apache.eventmesh.runtime.constants.EventMeshConstants;
-import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
-import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.HttpRequestProcessor;
-import org.apache.eventmesh.runtime.metrics.http.HTTPMetricsServer;
-import org.apache.eventmesh.runtime.util.EventMeshUtil;
-import org.apache.eventmesh.runtime.util.RemotingHelper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
public abstract class AbstractHTTPServer extends AbstractRemotingServer {
public Logger httpServerLogger = LoggerFactory.getLogger(this.getClass());
@@ -97,55 +100,41 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
private boolean useTLS;
public ThreadPoolExecutor asyncContextCompleteHandler =
- ThreadPoolFactory.createThreadPoolExecutor(10, 10, "eventMesh-http-asyncContext-");
+ ThreadPoolFactory.createThreadPoolExecutor(10, 10, "EventMesh-http-asyncContext-");
static {
DiskAttribute.deleteOnExitTemporaryFile = false;
}
- protected HashMap<Integer/* request code */, Pair<HttpRequestProcessor, ThreadPoolExecutor>> processorTable =
- new HashMap<Integer, Pair<HttpRequestProcessor, ThreadPoolExecutor>>(64);
+ protected final Map<String/* request code */, Pair<HttpRequestProcessor, ThreadPoolExecutor>>
+ processorTable = new HashMap<>(64);
public AbstractHTTPServer(int port, boolean useTLS) {
this.port = port;
this.useTLS = useTLS;
}
- public Map<String, Object> parseHTTPHeader(HttpRequest fullReq) {
- Map<String, Object> headerParam = new HashMap<>();
- for (String key : fullReq.headers().names()) {
- if (StringUtils.equalsIgnoreCase(HttpHeaderNames.CONTENT_TYPE.toString(), key)
- || StringUtils.equalsIgnoreCase(HttpHeaderNames.ACCEPT_ENCODING.toString(), key)
- || StringUtils.equalsIgnoreCase(HttpHeaderNames.CONTENT_LENGTH.toString(), key)) {
- continue;
- }
- headerParam.put(key, fullReq.headers().get(key));
- }
- return headerParam;
- }
-
- public void sendError(ChannelHandlerContext ctx,
- HttpResponseStatus status) {
- FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
- status);
- response.headers().add(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN +
- "; charset=" + EventMeshConstants.DEFAULT_CHARSET);
- response.headers().add(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
- response.headers().add(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
+ public void sendError(ChannelHandlerContext ctx, HttpResponseStatus status) {
+ FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status);
+ HttpHeaders responseHeaders = response.headers();
+ responseHeaders.add(
+ HttpHeaderNames.CONTENT_TYPE, String.format("text/plain; charset=%s", EventMeshConstants.DEFAULT_CHARSET)
+ );
+ responseHeaders.add(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
+ responseHeaders.add(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
// todo server span end with error, record status, we should get channel here to get span in channel's context in async call..
ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}
- public void sendResponse(ChannelHandlerContext ctx,
- DefaultFullHttpResponse response) {
+ public void sendResponse(ChannelHandlerContext ctx, DefaultFullHttpResponse response) {
// todo end server span, we should get channel here to get span in channel's context in async call.
ctx.writeAndFlush(response).addListener(new ChannelFutureListener() {
@Override
- public void operationComplete(ChannelFuture f) throws Exception {
+ public void operationComplete(ChannelFuture f) {
if (!f.isSuccess()) {
- httpLogger.warn("send response to [{}] fail, will close this channel", RemotingHelper.parseChannelRemoteAddr(f.channel()));
+ httpLogger.warn("send response to [{}] fail, will close this channel",
+ RemotingHelper.parseChannelRemoteAddr(f.channel()));
f.channel().close();
- return;
}
}
});
@@ -158,9 +147,9 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
ServerBootstrap b = new ServerBootstrap();
SSLContext sslContext = useTLS ? SSLContextFactory.getSslContext() : null;
b.group(this.bossGroup, this.workerGroup)
- .channel(NioServerSocketChannel.class)
- .childHandler(new HttpsServerInitializer(sslContext))
- .childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE);
+ .channel(NioServerSocketChannel.class)
+ .childHandler(new HttpsServerInitializer(sslContext))
+ .childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE);
try {
httpServerLogger.info("HTTPServer[port={}] started......", this.port);
ChannelFuture future = b.bind(this.port).sync();
@@ -172,11 +161,10 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
} catch (Exception e1) {
httpServerLogger.error("HTTPServer shutdown Err!", e);
}
- return;
}
};
- Thread t = new Thread(r, "eventMesh-http-server");
+ Thread t = new Thread(r, "EventMesh-http-server");
t.start();
started.compareAndSet(false, true);
}
@@ -196,62 +184,34 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
Preconditions.checkState(ObjectUtils.allNotNull(requestCode), "requestCode can't be null");
Preconditions.checkState(ObjectUtils.allNotNull(processor), "processor can't be null");
Preconditions.checkState(ObjectUtils.allNotNull(executor), "executor can't be null");
- Pair<HttpRequestProcessor, ThreadPoolExecutor> pair = new Pair<HttpRequestProcessor, ThreadPoolExecutor>(processor, executor);
- this.processorTable.put(requestCode, pair);
+ Pair<HttpRequestProcessor, ThreadPoolExecutor> pair = new Pair<>(processor, executor);
+ this.processorTable.put(requestCode.toString(), pair);
}
class HTTPHandler extends SimpleChannelInboundHandler<HttpRequest> {
@Override
- protected void channelRead0(ChannelHandlerContext ctx, HttpRequest httpRequest) throws Exception {
- HttpPostRequestDecoder decoder = null;
+ protected void channelRead0(ChannelHandlerContext ctx, HttpRequest httpRequest) {
// todo start server span, we should get channel here to put span in channel's context in async call.
try {
- if (!httpRequest.decoderResult().isSuccess()) {
- sendError(ctx, HttpResponseStatus.BAD_REQUEST);
+ preProcessHTTPRequestHeader(ctx, httpRequest);
+ final HttpResponseStatus errorStatus = validateHTTPRequest(httpRequest);
+ if (errorStatus != null) {
+ sendError(ctx, errorStatus);
return;
}
+ metrics.summaryMetrics.recordHTTPRequest();
final HttpCommand requestCommand = new HttpCommand();
// todo record command opaque in span.
- httpRequest.headers().set(ProtocolKey.ClientInstanceKey.IP, RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
-
- String protocolVersion = StringUtils.deleteWhitespace(httpRequest.headers().get(ProtocolKey.VERSION));
- if (StringUtils.isBlank(protocolVersion)) {
- protocolVersion = ProtocolVersion.V1.getVersion();
- httpRequest.headers().set(ProtocolKey.VERSION, ProtocolVersion.V1.getVersion());
- }
-
- metrics.summaryMetrics.recordHTTPRequest();
-
- long bodyDecodeStart = System.currentTimeMillis();
-
- Map<String, Object> bodyMap = new HashMap<>();
-
- if (httpRequest.method() == HttpMethod.GET) {
- QueryStringDecoder getDecoder = new QueryStringDecoder(httpRequest.uri());
- getDecoder.parameters().entrySet().forEach(entry -> {
- bodyMap.put(entry.getKey(), entry.getValue().get(0));
- });
- } else if (httpRequest.method() == HttpMethod.POST) {
- decoder = new HttpPostRequestDecoder(defaultHttpDataFactory, httpRequest);
- List<InterfaceHttpData> parmList = decoder.getBodyHttpDatas();
- for (InterfaceHttpData parm : parmList) {
- if (parm.getHttpDataType() == InterfaceHttpData.HttpDataType.Attribute) {
- Attribute data = (Attribute) parm;
- bodyMap.put(data.getName(), data.getValue());
- }
- }
- } else {
- sendError(ctx, HttpResponseStatus.METHOD_NOT_ALLOWED);
- return;
- }
+ final Map<String, Object> bodyMap = parseHttpRequestBody(httpRequest);
- metrics.summaryMetrics.recordDecodeTimeCost(System.currentTimeMillis() - bodyDecodeStart);
+ // todo: split get and post, use different submethod to process
String requestCode =
- (httpRequest.method() == HttpMethod.POST) ? StringUtils.deleteWhitespace(httpRequest.headers().get(ProtocolKey.REQUEST_CODE))
- : MapUtils.getString(bodyMap, StringUtils.lowerCase(ProtocolKey.REQUEST_CODE), "");
+ (httpRequest.method() == HttpMethod.POST)
+ ? httpRequest.headers().get(ProtocolKey.REQUEST_CODE)
+ : MapUtils.getString(bodyMap, StringUtils.lowerCase(ProtocolKey.REQUEST_CODE), "");
requestCommand.setHttpMethod(httpRequest.method().name());
requestCommand.setHttpVersion(httpRequest.protocolVersion().protocolName());
@@ -260,23 +220,11 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
HttpCommand responseCommand = null;
- if (!ProtocolVersion.contains(protocolVersion)) {
- responseCommand = requestCommand.createHttpCommandResponse(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(), EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg());
- sendResponse(ctx, responseCommand.httpResponse());
- return;
- }
-
if (StringUtils.isBlank(requestCode)
- || !StringUtils.isNumeric(requestCode)
- || !RequestCode.contains(Integer.valueOf(requestCode))
- || !processorTable.containsKey(Integer.valueOf(requestCode))) {
- responseCommand = requestCommand.createHttpCommandResponse(EventMeshRetCode.EVENTMESH_REQUESTCODE_INVALID.getRetCode(), EventMeshRetCode.EVENTMESH_REQUESTCODE_INVALID.getErrMsg());
- sendResponse(ctx, responseCommand.httpResponse());
- return;
- }
-
- if (!started.get()) {
- responseCommand = requestCommand.createHttpCommandResponse(EventMeshRetCode.EVENTMESH_STOP.getRetCode(), EventMeshRetCode.EVENTMESH_STOP.getErrMsg());
+ || !processorTable.containsKey(requestCode)
+ || !RequestCode.contains(Integer.valueOf(requestCode))) {
+ responseCommand =
+ requestCommand.createHttpCommandResponse(EventMeshRetCode.EVENTMESH_REQUESTCODE_INVALID);
sendResponse(ctx, responseCommand.httpResponse());
return;
}
@@ -285,7 +233,7 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
requestCommand.setHeader(Header.buildHeader(requestCode, parseHTTPHeader(httpRequest)));
requestCommand.setBody(Body.buildBody(requestCode, bodyMap));
} catch (Exception e) {
- responseCommand = requestCommand.createHttpCommandResponse(EventMeshRetCode.EVENTMESH_RUNTIME_ERR.getRetCode(), EventMeshRetCode.EVENTMESH_RUNTIME_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 3));
+ responseCommand = requestCommand.createHttpCommandResponse(EventMeshRetCode.EVENTMESH_RUNTIME_ERR);
sendResponse(ctx, responseCommand.httpResponse());
return;
}
@@ -294,27 +242,25 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
httpLogger.debug("{}", requestCommand);
}
- AsyncContext<HttpCommand> asyncContext = new AsyncContext<HttpCommand>(requestCommand, responseCommand, asyncContextCompleteHandler);
+ AsyncContext<HttpCommand> asyncContext =
+ new AsyncContext<>(requestCommand, responseCommand, asyncContextCompleteHandler);
processEventMeshRequest(ctx, asyncContext);
} catch (Exception ex) {
- httpServerLogger.error("AbrstractHTTPServer.HTTPHandler.channelRead0 err", ex);
+ httpServerLogger.error("AbstractHTTPServer.HTTPHandler.channelRead0 err", ex);
// todo span end with exception.
- } finally {
- try {
- decoder.destroy();
- } catch (Exception e) {
- }
}
}
public void processEventMeshRequest(final ChannelHandlerContext ctx,
final AsyncContext<HttpCommand> asyncContext) {
- final Pair<HttpRequestProcessor, ThreadPoolExecutor> choosed = processorTable.get(Integer.valueOf(asyncContext.getRequest().getRequestCode()));
+ final HttpCommand request = asyncContext.getRequest();
+ final Pair<HttpRequestProcessor, ThreadPoolExecutor> choosed = processorTable.get(request.getRequestCode());
try {
choosed.getObject2().submit(() -> {
try {
if (choosed.getObject1().rejectRequest()) {
- HttpCommand responseCommand = asyncContext.getRequest().createHttpCommandResponse(EventMeshRetCode.EVENTMESH_REJECT_BY_PROCESSOR_ERROR.getRetCode(), EventMeshRetCode.EVENTMESH_REJECT_BY_PROCESSOR_ERROR.getErrMsg());
+ HttpCommand responseCommand =
+ request.createHttpCommandResponse(EventMeshRetCode.EVENTMESH_REJECT_BY_PROCESSOR_ERROR);
asyncContext.onComplete(responseCommand);
if (asyncContext.isComplete()) {
if (httpLogger.isDebugEnabled()) {
@@ -326,11 +272,12 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
}
choosed.getObject1().processRequest(ctx, asyncContext);
- if (asyncContext == null || !asyncContext.isComplete()) {
+ if (!asyncContext.isComplete()) {
return;
}
- metrics.summaryMetrics.recordHTTPReqResTimeCost(System.currentTimeMillis() - asyncContext.getRequest().getReqTime());
+ metrics.summaryMetrics.recordHTTPReqResTimeCost(
+ System.currentTimeMillis() - request.getReqTime());
if (httpLogger.isDebugEnabled()) {
httpLogger.debug("{}", asyncContext.getResponse());
@@ -342,13 +289,14 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
}
});
} catch (RejectedExecutionException re) {
- HttpCommand responseCommand = asyncContext.getRequest().createHttpCommandResponse(EventMeshRetCode.OVERLOAD.getRetCode(), EventMeshRetCode.OVERLOAD.getErrMsg());
+ HttpCommand responseCommand = request.createHttpCommandResponse(EventMeshRetCode.OVERLOAD);
asyncContext.onComplete(responseCommand);
metrics.summaryMetrics.recordHTTPDiscard();
- metrics.summaryMetrics.recordHTTPReqResTimeCost(System.currentTimeMillis() - responseCommand.getReqTime());
+ metrics.summaryMetrics.recordHTTPReqResTimeCost(System.currentTimeMillis() - request.getReqTime());
try {
sendResponse(ctx, asyncContext.getResponse().httpResponse());
} catch (Exception e) {
+ // ignore
}
}
}
@@ -360,14 +308,18 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
}
@Override
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
- if (null != cause) cause.printStackTrace();
- if (null != ctx) ctx.close();
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ if (null != cause) {
+ logger.error("", cause);
+ }
+ if (null != ctx) {
+ ctx.close();
+ }
}
}
class HttpConnectionHandler extends ChannelDuplexHandler {
- public AtomicInteger connections = new AtomicInteger(0);
+ public final AtomicInteger connections = new AtomicInteger(0);
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
@@ -384,8 +336,9 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
int c = connections.incrementAndGet();
if (c > 20000) {
- httpServerLogger.warn("client|http|channelActive|remoteAddress={}|msg={}", remoteAddress, "too many client(20000) connect " +
- "this eventMesh server");
+ httpServerLogger
+ .warn("client|http|channelActive|remoteAddress={}|msg={}", remoteAddress,
+ "too many client(20000) connect this eventMesh server");
ctx.close();
return;
}
@@ -396,19 +349,17 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
connections.decrementAndGet();
- final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
super.channelInactive(ctx);
}
-
@Override
- public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
if (evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) evt;
if (event.state().equals(IdleState.ALL_IDLE)) {
final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
- httpServerLogger.info("client|http|userEventTriggered|remoteAddress={}|msg={}", remoteAddress, evt.getClass()
- .getName());
+ httpServerLogger.info("client|http|userEventTriggered|remoteAddress={}|msg={}",
+ remoteAddress, evt.getClass().getName());
ctx.close();
}
}
@@ -419,14 +370,14 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
class HttpsServerInitializer extends ChannelInitializer<SocketChannel> {
- private SSLContext sslContext;
+ private final SSLContext sslContext;
public HttpsServerInitializer(SSLContext sslContext) {
this.sslContext = sslContext;
}
@Override
- protected void initChannel(SocketChannel channel) throws Exception {
+ protected void initChannel(SocketChannel channel) {
ChannelPipeline pipeline = channel.pipeline();
@@ -436,10 +387,92 @@ public abstract class AbstractHTTPServer extends AbstractRemotingServer {
pipeline.addFirst("ssl", new SslHandler(sslEngine));
}
pipeline.addLast(new HttpRequestDecoder(),
- new HttpResponseEncoder(),
- new HttpConnectionHandler(),
- new HttpObjectAggregator(Integer.MAX_VALUE),
- new HTTPHandler());
+ new HttpResponseEncoder(),
+ new HttpConnectionHandler(),
+ new HttpObjectAggregator(Integer.MAX_VALUE),
+ new HTTPHandler());
+ }
+ }
+
+ private Map<String, Object> parseHTTPHeader(HttpRequest fullReq) {
+ Map<String, Object> headerParam = new HashMap<>();
+ for (String key : fullReq.headers().names()) {
+ if (StringUtils.equalsIgnoreCase(HttpHeaderNames.CONTENT_TYPE.toString(), key)
+ || StringUtils.equalsIgnoreCase(HttpHeaderNames.ACCEPT_ENCODING.toString(), key)
+ || StringUtils.equalsIgnoreCase(HttpHeaderNames.CONTENT_LENGTH.toString(), key)) {
+ continue;
+ }
+ headerParam.put(key, fullReq.headers().get(key));
+ }
+ return headerParam;
+ }
+
+ /**
+ * Validate request, return error status.
+ *
+ * @param httpRequest
+ * @return if request is validated return null else return error status
+ */
+ private HttpResponseStatus validateHTTPRequest(HttpRequest httpRequest) {
+ if (!started.get()) {
+ return HttpResponseStatus.SERVICE_UNAVAILABLE;
+ }
+ if (!httpRequest.decoderResult().isSuccess()) {
+ return HttpResponseStatus.BAD_REQUEST;
+ }
+ if (!HttpMethod.GET.equals(httpRequest.method()) && !HttpMethod.POST.equals(httpRequest.method())) {
+ return HttpResponseStatus.METHOD_NOT_ALLOWED;
+ }
+ final String protocolVersion = httpRequest.headers().get(ProtocolKey.VERSION);
+ if (!ProtocolVersion.contains(protocolVersion)) {
+ return HttpResponseStatus.BAD_REQUEST;
+ }
+ return null;
+ }
+
+ /**
+ * Inject ip and protocol version, if the protocol version is empty, set default to {@link ProtocolVersion#V1}.
+ *
+ * @param ctx
+ * @param httpRequest
+ */
+ private void preProcessHTTPRequestHeader(ChannelHandlerContext ctx, HttpRequest httpRequest) {
+ HttpHeaders requestHeaders = httpRequest.headers();
+ requestHeaders.set(ProtocolKey.ClientInstanceKey.IP,
+ RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
+
+ String protocolVersion = httpRequest.headers().get(ProtocolKey.VERSION);
+ if (StringUtils.isBlank(protocolVersion)) {
+ requestHeaders.set(ProtocolKey.VERSION, ProtocolVersion.V1.getVersion());
}
}
+
+ /**
+ * Parse request body to map
+ *
+ * @param httpRequest
+ * @return
+ */
+ private Map<String, Object> parseHttpRequestBody(HttpRequest httpRequest) throws IOException {
+ final long bodyDecodeStart = System.currentTimeMillis();
+ Map<String, Object> httpRequestBody = new HashMap<>();
+
+ if (HttpMethod.GET.equals(httpRequest.method())) {
+ QueryStringDecoder getDecoder = new QueryStringDecoder(httpRequest.uri());
+ getDecoder.parameters().forEach((key, value) -> httpRequestBody.put(key, value.get(0)));
+ } else if (HttpMethod.POST.equals(httpRequest.method())) {
+ HttpPostRequestDecoder decoder =
+ new HttpPostRequestDecoder(defaultHttpDataFactory, httpRequest);
+ for (InterfaceHttpData parm : decoder.getBodyHttpDatas()) {
+ if (parm.getHttpDataType() == InterfaceHttpData.HttpDataType.Attribute) {
+ Attribute data = (Attribute) parm;
+ httpRequestBody.put(data.getName(), data.getValue());
+ }
+ }
+ decoder.destroy();
+ }
+ metrics.summaryMetrics.recordDecodeTimeCost(System.currentTimeMillis() - bodyDecodeStart);
+ return httpRequestBody;
+ }
+
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
index f5055ba..ce3e2cc 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java
@@ -17,15 +17,21 @@
package org.apache.eventmesh.runtime.boot;
-import com.google.common.eventbus.EventBus;
-import com.google.common.util.concurrent.RateLimiter;
import org.apache.eventmesh.common.ThreadPoolFactory;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.runtime.common.ServiceState;
import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
import org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupConf;
import org.apache.eventmesh.runtime.core.protocol.http.consumer.ConsumerManager;
-import org.apache.eventmesh.runtime.core.protocol.http.processor.*;
+import org.apache.eventmesh.runtime.core.protocol.http.processor.AdminMetricsProcessor;
+import org.apache.eventmesh.runtime.core.protocol.http.processor.BatchSendMessageProcessor;
+import org.apache.eventmesh.runtime.core.protocol.http.processor.BatchSendMessageV2Processor;
+import org.apache.eventmesh.runtime.core.protocol.http.processor.HeartBeatProcessor;
+import org.apache.eventmesh.runtime.core.protocol.http.processor.ReplyMessageProcessor;
+import org.apache.eventmesh.runtime.core.protocol.http.processor.SendAsyncMessageProcessor;
+import org.apache.eventmesh.runtime.core.protocol.http.processor.SendSyncMessageProcessor;
+import org.apache.eventmesh.runtime.core.protocol.http.processor.SubscribeProcessor;
+import org.apache.eventmesh.runtime.core.protocol.http.processor.UnSubscribeProcessor;
import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.Client;
import org.apache.eventmesh.runtime.core.protocol.http.producer.ProducerManager;
import org.apache.eventmesh.runtime.core.protocol.http.push.AbstractHTTPPushRequest;
@@ -38,6 +44,9 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
+import com.google.common.eventbus.EventBus;
+import com.google.common.util.concurrent.RateLimiter;
+
public class EventMeshHTTPServer extends AbstractHTTPServer {
private EventMeshServer eventMeshServer;
@@ -46,9 +55,11 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
private EventMeshHTTPConfiguration eventMeshHttpConfiguration;
- public final ConcurrentHashMap<String /**group*/, ConsumerGroupConf> localConsumerGroupMapping = new ConcurrentHashMap<>();
+ public final ConcurrentHashMap<String /**group*/, ConsumerGroupConf> localConsumerGroupMapping =
+ new ConcurrentHashMap<>();
- public final ConcurrentHashMap<String /**group@topic*/, List<Client>> localClientInfoMapping = new ConcurrentHashMap<>();
+ public final ConcurrentHashMap<String /**group@topic*/, List<Client>> localClientInfoMapping =
+ new ConcurrentHashMap<>();
public EventMeshHTTPServer(EventMeshServer eventMeshServer,
EventMeshHTTPConfiguration eventMeshHttpConfiguration) {
@@ -96,29 +107,45 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
public void initThreadPool() throws Exception {
- BlockingQueue<Runnable> batchMsgThreadPoolQueue = new LinkedBlockingQueue<Runnable>(eventMeshHttpConfiguration.eventMeshServerBatchBlockQSize);
- batchMsgExecutor = ThreadPoolFactory.createThreadPoolExecutor(eventMeshHttpConfiguration.eventMeshServerBatchMsgThreadNum,
- eventMeshHttpConfiguration.eventMeshServerBatchMsgThreadNum, batchMsgThreadPoolQueue, "eventMesh-batchMsg-", true);
-
- BlockingQueue<Runnable> sendMsgThreadPoolQueue = new LinkedBlockingQueue<Runnable>(eventMeshHttpConfiguration.eventMeshServerSendMsgBlockQSize);
- sendMsgExecutor = ThreadPoolFactory.createThreadPoolExecutor(eventMeshHttpConfiguration.eventMeshServerSendMsgThreadNum,
- eventMeshHttpConfiguration.eventMeshServerSendMsgThreadNum, sendMsgThreadPoolQueue, "eventMesh-sendMsg-", true);
-
- BlockingQueue<Runnable> pushMsgThreadPoolQueue = new LinkedBlockingQueue<Runnable>(eventMeshHttpConfiguration.eventMeshServerPushMsgBlockQSize);
- pushMsgExecutor = ThreadPoolFactory.createThreadPoolExecutor(eventMeshHttpConfiguration.eventMeshServerPushMsgThreadNum,
- eventMeshHttpConfiguration.eventMeshServerPushMsgThreadNum, pushMsgThreadPoolQueue, "eventMesh-pushMsg-", true);
-
- BlockingQueue<Runnable> clientManageThreadPoolQueue = new LinkedBlockingQueue<Runnable>(eventMeshHttpConfiguration.eventMeshServerClientManageBlockQSize);
- clientManageExecutor = ThreadPoolFactory.createThreadPoolExecutor(eventMeshHttpConfiguration.eventMeshServerClientManageThreadNum,
- eventMeshHttpConfiguration.eventMeshServerClientManageThreadNum, clientManageThreadPoolQueue, "eventMesh-clientManage-", true);
+ BlockingQueue<Runnable> batchMsgThreadPoolQueue =
+ new LinkedBlockingQueue<Runnable>(eventMeshHttpConfiguration.eventMeshServerBatchBlockQSize);
+ batchMsgExecutor =
+ ThreadPoolFactory.createThreadPoolExecutor(eventMeshHttpConfiguration.eventMeshServerBatchMsgThreadNum,
+ eventMeshHttpConfiguration.eventMeshServerBatchMsgThreadNum, batchMsgThreadPoolQueue,
+ "eventMesh-batchMsg-", true);
+
+ BlockingQueue<Runnable> sendMsgThreadPoolQueue =
+ new LinkedBlockingQueue<Runnable>(eventMeshHttpConfiguration.eventMeshServerSendMsgBlockQSize);
+ sendMsgExecutor =
+ ThreadPoolFactory.createThreadPoolExecutor(eventMeshHttpConfiguration.eventMeshServerSendMsgThreadNum,
+ eventMeshHttpConfiguration.eventMeshServerSendMsgThreadNum, sendMsgThreadPoolQueue,
+ "eventMesh-sendMsg-", true);
+
+ BlockingQueue<Runnable> pushMsgThreadPoolQueue =
+ new LinkedBlockingQueue<Runnable>(eventMeshHttpConfiguration.eventMeshServerPushMsgBlockQSize);
+ pushMsgExecutor =
+ ThreadPoolFactory.createThreadPoolExecutor(eventMeshHttpConfiguration.eventMeshServerPushMsgThreadNum,
+ eventMeshHttpConfiguration.eventMeshServerPushMsgThreadNum, pushMsgThreadPoolQueue,
+ "eventMesh-pushMsg-", true);
+
+ BlockingQueue<Runnable> clientManageThreadPoolQueue =
+ new LinkedBlockingQueue<Runnable>(eventMeshHttpConfiguration.eventMeshServerClientManageBlockQSize);
+ clientManageExecutor =
+ ThreadPoolFactory.createThreadPoolExecutor(eventMeshHttpConfiguration.eventMeshServerClientManageThreadNum,
+ eventMeshHttpConfiguration.eventMeshServerClientManageThreadNum, clientManageThreadPoolQueue,
+ "eventMesh-clientManage-", true);
BlockingQueue<Runnable> adminThreadPoolQueue = new LinkedBlockingQueue<Runnable>(50);
- adminExecutor = ThreadPoolFactory.createThreadPoolExecutor(eventMeshHttpConfiguration.eventMeshServerAdminThreadNum,
- eventMeshHttpConfiguration.eventMeshServerAdminThreadNum, adminThreadPoolQueue, "eventMesh-admin-", true);
+ adminExecutor =
+ ThreadPoolFactory.createThreadPoolExecutor(eventMeshHttpConfiguration.eventMeshServerAdminThreadNum,
+ eventMeshHttpConfiguration.eventMeshServerAdminThreadNum, adminThreadPoolQueue, "eventMesh-admin-",
+ true);
BlockingQueue<Runnable> replyMessageThreadPoolQueue = new LinkedBlockingQueue<Runnable>(100);
- replyMsgExecutor = ThreadPoolFactory.createThreadPoolExecutor(eventMeshHttpConfiguration.eventMeshServerReplyMsgThreadNum,
- eventMeshHttpConfiguration.eventMeshServerReplyMsgThreadNum, replyMessageThreadPoolQueue, "eventMesh-replyMsg-", true);
+ replyMsgExecutor =
+ ThreadPoolFactory.createThreadPoolExecutor(eventMeshHttpConfiguration.eventMeshServerReplyMsgThreadNum,
+ eventMeshHttpConfiguration.eventMeshServerReplyMsgThreadNum, replyMessageThreadPoolQueue,
+ "eventMesh-replyMsg-", true);
}
public ThreadPoolExecutor getBatchMsgExecutor() {
@@ -210,20 +237,17 @@ public class EventMeshHTTPServer extends AbstractHTTPServer {
public void registerHTTPRequestProcessor() {
BatchSendMessageProcessor batchSendMessageProcessor = new BatchSendMessageProcessor(this);
- registerProcessor(RequestCode.MSG_BATCH_SEND.getRequestCode(),
- batchSendMessageProcessor, batchMsgExecutor);
+ registerProcessor(RequestCode.MSG_BATCH_SEND.getRequestCode(), batchSendMessageProcessor, batchMsgExecutor);
BatchSendMessageV2Processor batchSendMessageV2Processor = new BatchSendMessageV2Processor(this);
- registerProcessor(RequestCode.MSG_BATCH_SEND_V2.getRequestCode(),
- batchSendMessageV2Processor, batchMsgExecutor);
+ registerProcessor(RequestCode.MSG_BATCH_SEND_V2.getRequestCode(), batchSendMessageV2Processor,
+ batchMsgExecutor);
SendSyncMessageProcessor sendSyncMessageProcessor = new SendSyncMessageProcessor(this);
- registerProcessor(RequestCode.MSG_SEND_SYNC.getRequestCode(),
- sendSyncMessageProcessor, sendMsgExecutor);
+ registerProcessor(RequestCode.MSG_SEND_SYNC.getRequestCode(), sendSyncMessageProcessor, sendMsgExecutor);
SendAsyncMessageProcessor sendAsyncMessageProcessor = new SendAsyncMessageProcessor(this);
- registerProcessor(RequestCode.MSG_SEND_ASYNC.getRequestCode(),
- sendAsyncMessageProcessor, sendMsgExecutor);
+ registerProcessor(RequestCode.MSG_SEND_ASYNC.getRequestCode(), sendAsyncMessageProcessor, sendMsgExecutor);
AdminMetricsProcessor adminMetricsProcessor = new AdminMetricsProcessor(this);
registerProcessor(RequestCode.ADMIN_METRICS.getRequestCode(), adminMetricsProcessor, adminExecutor);
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminMetricsProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminMetricsProcessor.java
index dac2746..6feaf9a 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminMetricsProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminMetricsProcessor.java
@@ -28,8 +28,6 @@ import org.slf4j.LoggerFactory;
public class AdminMetricsProcessor implements HttpRequestProcessor {
- public Logger cmdLogger = LoggerFactory.getLogger("cmd");
-
private EventMeshHTTPServer eventMeshHTTPServer;
public AdminMetricsProcessor(EventMeshHTTPServer eventMeshHTTPServer) {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminShutdownProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminShutdownProcessor.java
index 775a7c1..ca6212a 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminShutdownProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/AdminShutdownProcessor.java
@@ -17,8 +17,6 @@
package org.apache.eventmesh.runtime.core.protocol.http.processor;
-import io.netty.channel.ChannelHandlerContext;
-
import org.apache.eventmesh.common.IPUtil;
import org.apache.eventmesh.common.command.HttpCommand;
import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
@@ -28,9 +26,12 @@ import org.apache.eventmesh.runtime.constants.EventMeshConstants;
import org.apache.eventmesh.runtime.core.protocol.http.async.AsyncContext;
import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.HttpRequestProcessor;
import org.apache.eventmesh.runtime.util.RemotingHelper;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import io.netty.channel.ChannelHandlerContext;
+
public class AdminShutdownProcessor implements HttpRequestProcessor {
public Logger cmdLogger = LoggerFactory.getLogger("cmd");
@@ -45,14 +46,14 @@ public class AdminShutdownProcessor implements HttpRequestProcessor {
public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpCommand> asyncContext) throws Exception {
HttpCommand responseEventMeshCommand;
- cmdLogger.info("cmd={}|{}|client2eventMesh|from={}|to={}", RequestCode.get(Integer.valueOf(asyncContext.getRequest().getRequestCode())),
- EventMeshConstants.PROTOCOL_HTTP,
- RemotingHelper.parseChannelRemoteAddr(ctx.channel()), IPUtil.getLocalAddress());
+ cmdLogger.info("cmd={}|{}|client2eventMesh|from={}|to={}",
+ RequestCode.get(Integer.valueOf(asyncContext.getRequest().getRequestCode())),
+ EventMeshConstants.PROTOCOL_HTTP,
+ RemotingHelper.parseChannelRemoteAddr(ctx.channel()), IPUtil.getLocalAddress());
eventMeshServer.shutdown();
- responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- EventMeshRetCode.SUCCESS.getRetCode(), EventMeshRetCode.SUCCESS.getErrMsg());
+ responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(EventMeshRetCode.SUCCESS);
asyncContext.onComplete(responseEventMeshCommand);
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
index c38ff5d..2787142 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
@@ -17,17 +17,16 @@
package org.apache.eventmesh.runtime.core.protocol.http.processor;
-import io.cloudevents.CloudEvent;
-import io.cloudevents.core.builder.CloudEventBuilder;
-import io.netty.channel.ChannelHandlerContext;
-import org.apache.commons.lang3.StringUtils;
import org.apache.eventmesh.api.SendCallback;
import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.exception.OnExceptionContext;
-import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.IPUtil;
+import org.apache.eventmesh.common.ProtocolTransportObject;
import org.apache.eventmesh.common.command.HttpCommand;
-import org.apache.eventmesh.common.protocol.http.body.message.*;
+import org.apache.eventmesh.common.protocol.http.body.message.SendMessageBatchV2RequestBody;
+import org.apache.eventmesh.common.protocol.http.body.message.SendMessageBatchV2ResponseBody;
+import org.apache.eventmesh.common.protocol.http.body.message.SendMessageRequestBody;
+import org.apache.eventmesh.common.protocol.http.body.message.SendMessageResponseBody;
import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
@@ -44,13 +43,19 @@ import org.apache.eventmesh.runtime.core.protocol.http.producer.EventMeshProduce
import org.apache.eventmesh.runtime.core.protocol.http.producer.SendMessageContext;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.eventmesh.runtime.util.RemotingHelper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.util.List;
+import org.apache.commons.lang3.StringUtils;
+
import java.util.Objects;
import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import io.netty.channel.ChannelHandlerContext;
+
public class BatchSendMessageV2Processor implements HttpRequestProcessor {
public Logger cmdLogger = LoggerFactory.getLogger("cmd");
@@ -66,66 +71,89 @@ public class BatchSendMessageV2Processor implements HttpRequestProcessor {
public Logger batchMessageLogger = LoggerFactory.getLogger("batchMessage");
@Override
- public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpCommand> asyncContext) throws Exception {
+ public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpCommand> asyncContext)
+ throws Exception {
HttpCommand responseEventMeshCommand;
+ final HttpCommand request = asyncContext.getRequest();
+ final Integer requestCode = Integer.valueOf(request.getRequestCode());
- cmdLogger.info("cmd={}|{}|client2eventMesh|from={}|to={}", RequestCode.get(Integer.valueOf(asyncContext.getRequest().getRequestCode())),
- EventMeshConstants.PROTOCOL_HTTP,
- RemotingHelper.parseChannelRemoteAddr(ctx.channel()), IPUtil.getLocalAddress());
+ cmdLogger.info("cmd={}|{}|client2eventMesh|from={}|to={}",
+ RequestCode.get(requestCode),
+ EventMeshConstants.PROTOCOL_HTTP,
+ RemotingHelper.parseChannelRemoteAddr(ctx.channel()), IPUtil.getLocalAddress());
- SendMessageBatchV2RequestHeader sendMessageBatchV2RequestHeader = (SendMessageBatchV2RequestHeader) asyncContext.getRequest().getHeader();
+ SendMessageBatchV2RequestHeader sendMessageBatchV2RequestHeader =
+ (SendMessageBatchV2RequestHeader) asyncContext.getRequest().getHeader();
String protocolType = sendMessageBatchV2RequestHeader.getProtocolType();
- ProtocolAdaptor httpCommandProtocolAdaptor = ProtocolPluginFactory.getProtocolAdaptor(protocolType);
+ ProtocolAdaptor<ProtocolTransportObject> httpCommandProtocolAdaptor =
+ ProtocolPluginFactory.getProtocolAdaptor(protocolType);
CloudEvent event = httpCommandProtocolAdaptor.toCloudEvent(asyncContext.getRequest());
SendMessageBatchV2ResponseHeader sendMessageBatchV2ResponseHeader =
- SendMessageBatchV2ResponseHeader.buildHeader(Integer.valueOf(asyncContext.getRequest().getRequestCode()), eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshCluster,
- IPUtil.getLocalAddress(), eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEnv,
- eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshIDC);
-
+ SendMessageBatchV2ResponseHeader.buildHeader(
+ requestCode,
+ eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshCluster,
+ IPUtil.getLocalAddress(),
+ eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEnv,
+ eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshIDC
+ );
+
+ // todo: use validate processor to check
//validate event
- if (event != null
- || StringUtils.isBlank(event.getId())
- || event.getSource() != null
- || event.getSpecVersion() != null
- || StringUtils.isBlank(event.getType())
- || StringUtils.isBlank(event.getSubject())) {
- responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(), EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
+ if (StringUtils.isBlank(event.getId())
+ || event.getSource() != null
+ || event.getSpecVersion() != null
+ || StringUtils.isBlank(event.getType())
+ || StringUtils.isBlank(event.getSubject())) {
+ responseEventMeshCommand = request.createHttpCommandResponse(
+ sendMessageBatchV2ResponseHeader,
+ SendMessageBatchV2ResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
asyncContext.onComplete(responseEventMeshCommand);
return;
}
- String idc = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.IDC)).toString();
- String pid = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.PID)).toString();
- String sys = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.SYS)).toString();
+ String idc = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.IDC))
+ .toString();
+ String pid = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.PID))
+ .toString();
+ String sys = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.SYS))
+ .toString();
//validate event-extension
if (StringUtils.isBlank(idc)
- || StringUtils.isBlank(pid)
- || !StringUtils.isNumeric(pid)
- || StringUtils.isBlank(sys)) {
- responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(), EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
+ || StringUtils.isBlank(pid)
+ || !StringUtils.isNumeric(pid)
+ || StringUtils.isBlank(sys)) {
+ responseEventMeshCommand = request.createHttpCommandResponse(
+ sendMessageBatchV2ResponseHeader,
+ SendMessageBatchV2ResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
asyncContext.onComplete(responseEventMeshCommand);
return;
}
- String bizNo = Objects.requireNonNull(event.getExtension(SendMessageBatchV2RequestBody.BIZSEQNO)).toString();
- String producerGroup = Objects.requireNonNull(event.getExtension(SendMessageBatchV2RequestBody.PRODUCERGROUP)).toString();
+ String bizNo =
+ Objects.requireNonNull(event.getExtension(SendMessageBatchV2RequestBody.BIZSEQNO))
+ .toString();
+ String producerGroup =
+ Objects.requireNonNull(event.getExtension(SendMessageBatchV2RequestBody.PRODUCERGROUP))
+ .toString();
String topic = event.getSubject();
if (StringUtils.isBlank(bizNo)
- || StringUtils.isBlank(topic)
- || StringUtils.isBlank(producerGroup)
- || event.getData() != null) {
- responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(), EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
+ || StringUtils.isBlank(topic)
+ || StringUtils.isBlank(producerGroup)
+ || event.getData() != null) {
+ responseEventMeshCommand = request.createHttpCommandResponse(
+ sendMessageBatchV2ResponseHeader,
+ SendMessageBatchV2ResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
asyncContext.onComplete(responseEventMeshCommand);
return;
}
@@ -136,38 +164,46 @@ public class BatchSendMessageV2Processor implements HttpRequestProcessor {
String user = event.getExtension(ProtocolKey.ClientInstanceKey.USERNAME).toString();
String pass = event.getExtension(ProtocolKey.ClientInstanceKey.PASSWD).toString();
String subsystem = event.getExtension(ProtocolKey.ClientInstanceKey.SYS).toString();
- int requestCode = Integer.parseInt(asyncContext.getRequest().getRequestCode());
try {
Acl.doAclCheckInHttpSend(remoteAddr, user, pass, subsystem, topic, requestCode);
} catch (Exception e) {
//String errorMsg = String.format("CLIENT HAS NO PERMISSION,send failed, topic:%s, subsys:%s, realIp:%s", topic, subsys, realIp);
responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_ACL_ERR.getRetCode(), e.getMessage()));
+ sendMessageBatchV2ResponseHeader,
+ SendMessageResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_ACL_ERR.getRetCode(),
+ e.getMessage()));
asyncContext.onComplete(responseEventMeshCommand);
- aclLogger.warn("CLIENT HAS NO PERMISSION,BatchSendMessageV2Processor send failed", e);
+ aclLogger
+ .warn("CLIENT HAS NO PERMISSION,BatchSendMessageV2Processor send failed", e);
return;
}
}
if (!eventMeshHTTPServer.getBatchRateLimiter()
- .tryAcquire(EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS, TimeUnit.MILLISECONDS)) {
- responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody.buildBody(EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getRetCode(), EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getErrMsg()));
+ .tryAcquire(EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS,
+ TimeUnit.MILLISECONDS)) {
+ responseEventMeshCommand = request.createHttpCommandResponse(
+ sendMessageBatchV2ResponseHeader,
+ SendMessageBatchV2ResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_BATCH_SPEED_OVER_LIMIT_ERR.getErrMsg()));
eventMeshHTTPServer.metrics.summaryMetrics
- .recordSendBatchMsgDiscard(1);
+ .recordSendBatchMsgDiscard(1);
asyncContext.onComplete(responseEventMeshCommand);
return;
}
- EventMeshProducer batchEventMeshProducer = eventMeshHTTPServer.getProducerManager().getEventMeshProducer(producerGroup);
+ EventMeshProducer batchEventMeshProducer =
+ eventMeshHTTPServer.getProducerManager().getEventMeshProducer(producerGroup);
batchEventMeshProducer.getMqProducerWrapper().getMeshMQProducer().setExtFields();
if (!batchEventMeshProducer.getStarted().get()) {
- responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody.buildBody(EventMeshRetCode.EVENTMESH_BATCH_PRODUCER_STOPED_ERR.getRetCode(), EventMeshRetCode.EVENTMESH_BATCH_PRODUCER_STOPED_ERR.getErrMsg()));
+ responseEventMeshCommand = request.createHttpCommandResponse(
+ sendMessageBatchV2ResponseHeader,
+ SendMessageBatchV2ResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_BATCH_PRODUCER_STOPED_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_BATCH_PRODUCER_STOPED_ERR.getErrMsg()));
asyncContext.onComplete(responseEventMeshCommand);
return;
}
@@ -175,18 +211,22 @@ public class BatchSendMessageV2Processor implements HttpRequestProcessor {
long batchStartTime = System.currentTimeMillis();
String ttl = String.valueOf(EventMeshConstants.DEFAULT_MSG_TTL_MILLS);
+ // todo: use hashmap to avoid copy
if (StringUtils.isBlank(event.getExtension(SendMessageRequestBody.TTL).toString())
- && !StringUtils.isNumeric(event.getExtension(SendMessageRequestBody.TTL).toString())) {
- event = CloudEventBuilder.from(event).withExtension(SendMessageRequestBody.TTL, ttl).build();
+ && !StringUtils.isNumeric(event.getExtension(SendMessageRequestBody.TTL).toString())) {
+ event = CloudEventBuilder.from(event).withExtension(SendMessageRequestBody.TTL, ttl)
+ .build();
}
try {
event = CloudEventBuilder.from(event)
- .withExtension("msgType", "persistent")
- .withExtension(EventMeshConstants.REQ_C2EVENTMESH_TIMESTAMP, String.valueOf(System.currentTimeMillis()))
- .withExtension(EventMeshConstants.REQ_EVENTMESH2MQ_TIMESTAMP, String.valueOf(System.currentTimeMillis()))
- .build();
+ .withExtension("msgType", "persistent")
+ .withExtension(EventMeshConstants.REQ_C2EVENTMESH_TIMESTAMP,
+ String.valueOf(System.currentTimeMillis()))
+ .withExtension(EventMeshConstants.REQ_EVENTMESH2MQ_TIMESTAMP,
+ String.valueOf(System.currentTimeMillis()))
+ .build();
if (batchMessageLogger.isDebugEnabled()) {
batchMessageLogger.debug("msg2MQMsg suc, topic:{}, msg:{}", topic, event.getData());
}
@@ -194,54 +234,67 @@ public class BatchSendMessageV2Processor implements HttpRequestProcessor {
} catch (Exception e) {
batchMessageLogger.error("msg2MQMsg err, topic:{}, msg:{}", topic, event.getData(), e);
responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getRetCode(), EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2)));
+ sendMessageBatchV2ResponseHeader,
+ SendMessageBatchV2ResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_PACKAGE_MSG_ERR.getErrMsg() +
+ EventMeshUtil.stackTrace(e, 2)));
asyncContext.onComplete(responseEventMeshCommand);
return;
}
eventMeshHTTPServer.metrics.summaryMetrics.recordSendBatchMsg(1);
- final SendMessageContext sendMessageContext = new SendMessageContext(bizNo, event, batchEventMeshProducer, eventMeshHTTPServer);
+ final SendMessageContext sendMessageContext =
+ new SendMessageContext(bizNo, event, batchEventMeshProducer, eventMeshHTTPServer);
try {
batchEventMeshProducer.send(sendMessageContext, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
long batchEndTime = System.currentTimeMillis();
- eventMeshHTTPServer.metrics.summaryMetrics.recordBatchSendMsgCost(batchEndTime - batchStartTime);
- batchMessageLogger.debug("batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
- bizNo, batchEndTime - batchStartTime, topic);
+ eventMeshHTTPServer.metrics.summaryMetrics
+ .recordBatchSendMsgCost(batchEndTime - batchStartTime);
+ batchMessageLogger.debug(
+ "batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
+ bizNo, batchEndTime - batchStartTime, topic);
}
@Override
public void onException(OnExceptionContext context) {
long batchEndTime = System.currentTimeMillis();
eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10000));
- eventMeshHTTPServer.metrics.summaryMetrics.recordBatchSendMsgCost(batchEndTime - batchStartTime);
- batchMessageLogger.error("batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
- bizNo, batchEndTime - batchStartTime, topic, context.getException());
+ eventMeshHTTPServer.metrics.summaryMetrics
+ .recordBatchSendMsgCost(batchEndTime - batchStartTime);
+ batchMessageLogger.error(
+ "batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
+ bizNo, batchEndTime - batchStartTime, topic, context.getException());
}
});
} catch (Exception e) {
responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody.buildBody(EventMeshRetCode.EVENTMESH_SEND_BATCHLOG_MSG_ERR.getRetCode(), EventMeshRetCode.EVENTMESH_SEND_BATCHLOG_MSG_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2)));
+ sendMessageBatchV2ResponseHeader,
+ SendMessageBatchV2ResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_SEND_BATCHLOG_MSG_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_SEND_BATCHLOG_MSG_ERR.getErrMsg() +
+ EventMeshUtil.stackTrace(e, 2)));
asyncContext.onComplete(responseEventMeshCommand);
long batchEndTime = System.currentTimeMillis();
eventMeshHTTPServer.getHttpRetryer().pushRetry(sendMessageContext.delay(10000));
- eventMeshHTTPServer.metrics.summaryMetrics.recordBatchSendMsgCost(batchEndTime - batchStartTime);
- batchMessageLogger.error("batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
- bizNo, batchEndTime - batchStartTime, topic, e);
+ eventMeshHTTPServer.metrics.summaryMetrics
+ .recordBatchSendMsgCost(batchEndTime - batchStartTime);
+ batchMessageLogger.error(
+ "batchMessageV2|eventMesh2mq|REQ|ASYNC|bizSeqNo={}|send2MQCost={}ms|topic={}",
+ bizNo, batchEndTime - batchStartTime, topic, e);
}
responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- sendMessageBatchV2ResponseHeader,
- SendMessageBatchV2ResponseBody.buildBody(EventMeshRetCode.SUCCESS.getRetCode(), EventMeshRetCode.SUCCESS.getErrMsg()));
+ sendMessageBatchV2ResponseHeader,
+ SendMessageBatchV2ResponseBody.buildBody(EventMeshRetCode.SUCCESS.getRetCode(),
+ EventMeshRetCode.SUCCESS.getErrMsg()));
asyncContext.onComplete(responseEventMeshCommand);
- return;
}
@Override
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java
index e0341cf..3fefd51 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HeartBeatProcessor.java
@@ -17,16 +17,6 @@
package org.apache.eventmesh.runtime.core.protocol.http.processor;
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-import io.netty.channel.ChannelHandlerContext;
-
-import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang3.StringUtils;
import org.apache.eventmesh.common.IPUtil;
import org.apache.eventmesh.common.command.HttpCommand;
import org.apache.eventmesh.common.protocol.http.body.client.HeartbeatRequestBody;
@@ -45,9 +35,21 @@ import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.Client;
import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.HttpRequestProcessor;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.eventmesh.runtime.util.RemotingHelper;
+
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import io.netty.channel.ChannelHandlerContext;
+
public class HeartBeatProcessor implements HttpRequestProcessor {
public Logger httpLogger = LoggerFactory.getLogger("http");
@@ -63,38 +65,42 @@ public class HeartBeatProcessor implements HttpRequestProcessor {
@Override
public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpCommand> asyncContext) throws Exception {
HttpCommand responseEventMeshCommand;
- httpLogger.info("cmd={}|{}|client2eventMesh|from={}|to={}", RequestCode.get(Integer.valueOf(asyncContext.getRequest().getRequestCode())),
- EventMeshConstants.PROTOCOL_HTTP,
- RemotingHelper.parseChannelRemoteAddr(ctx.channel()), IPUtil.getLocalAddress());
+ httpLogger.info("cmd={}|{}|client2eventMesh|from={}|to={}",
+ RequestCode.get(Integer.valueOf(asyncContext.getRequest().getRequestCode())),
+ EventMeshConstants.PROTOCOL_HTTP,
+ RemotingHelper.parseChannelRemoteAddr(ctx.channel()), IPUtil.getLocalAddress());
HeartbeatRequestHeader heartbeatRequestHeader = (HeartbeatRequestHeader) asyncContext.getRequest().getHeader();
HeartbeatRequestBody heartbeatRequestBody = (HeartbeatRequestBody) asyncContext.getRequest().getBody();
HeartbeatResponseHeader heartbeatResponseHeader =
- HeartbeatResponseHeader.buildHeader(Integer.valueOf(asyncContext.getRequest().getRequestCode()), eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshCluster,
- IPUtil.getLocalAddress(), eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEnv,
- eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshIDC);
+ HeartbeatResponseHeader.buildHeader(Integer.valueOf(asyncContext.getRequest().getRequestCode()),
+ eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshCluster,
+ IPUtil.getLocalAddress(), eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEnv,
+ eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshIDC);
//validate header
if (StringUtils.isBlank(heartbeatRequestHeader.getIdc())
- || StringUtils.isBlank(heartbeatRequestHeader.getPid())
- || !StringUtils.isNumeric(heartbeatRequestHeader.getPid())
- || StringUtils.isBlank(heartbeatRequestHeader.getSys())) {
+ || StringUtils.isBlank(heartbeatRequestHeader.getPid())
+ || !StringUtils.isNumeric(heartbeatRequestHeader.getPid())
+ || StringUtils.isBlank(heartbeatRequestHeader.getSys())) {
responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- heartbeatResponseHeader,
- HeartbeatResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(), EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
+ heartbeatResponseHeader,
+ HeartbeatResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
asyncContext.onComplete(responseEventMeshCommand);
return;
}
//validate body
if (StringUtils.isBlank(heartbeatRequestBody.getClientType())
- || StringUtils.isBlank(heartbeatRequestBody.getConsumerGroup())
- || CollectionUtils.isEmpty(heartbeatRequestBody.getHeartbeatEntities())) {
+ || StringUtils.isBlank(heartbeatRequestBody.getConsumerGroup())
+ || CollectionUtils.isEmpty(heartbeatRequestBody.getHeartbeatEntities())) {
responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- heartbeatResponseHeader,
- HeartbeatResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(), EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
+ heartbeatResponseHeader,
+ HeartbeatResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
asyncContext.onComplete(responseEventMeshCommand);
return;
}
@@ -126,7 +132,7 @@ public class HeartBeatProcessor implements HttpRequestProcessor {
}
//do acl check
- if(eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshServerSecurityEnable) {
+ if (eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshServerSecurityEnable) {
String remoteAddr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
String user = heartbeatRequestHeader.getUsername();
String pass = heartbeatRequestHeader.getPasswd();
@@ -137,8 +143,9 @@ public class HeartBeatProcessor implements HttpRequestProcessor {
//String errorMsg = String.format("CLIENT HAS NO PERMISSION,send failed, topic:%s, subsys:%s, realIp:%s", topic, subsys, realIp);
responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- heartbeatResponseHeader,
- SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_ACL_ERR.getRetCode(), e.getMessage()));
+ heartbeatResponseHeader,
+ SendMessageResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_ACL_ERR.getRetCode(), e.getMessage()));
asyncContext.onComplete(responseEventMeshCommand);
aclLogger.warn("CLIENT HAS NO PERMISSION,HeartBeatProcessor subscribe failed", e);
return;
@@ -161,9 +168,11 @@ public class HeartBeatProcessor implements HttpRequestProcessor {
}
synchronized (eventMeshHTTPServer.localClientInfoMapping) {
for (Map.Entry<String, List<Client>> groupTopicClientMapping : tmp.entrySet()) {
- List<Client> localClientList = eventMeshHTTPServer.localClientInfoMapping.get(groupTopicClientMapping.getKey());
+ List<Client> localClientList =
+ eventMeshHTTPServer.localClientInfoMapping.get(groupTopicClientMapping.getKey());
if (CollectionUtils.isEmpty(localClientList)) {
- eventMeshHTTPServer.localClientInfoMapping.put(groupTopicClientMapping.getKey(), groupTopicClientMapping.getValue());
+ eventMeshHTTPServer.localClientInfoMapping
+ .put(groupTopicClientMapping.getKey(), groupTopicClientMapping.getValue());
} else {
List<Client> tmpClientList = groupTopicClientMapping.getValue();
supplyClientInfoList(tmpClientList, localClientList);
@@ -184,24 +193,24 @@ public class HeartBeatProcessor implements HttpRequestProcessor {
httpLogger.debug("{}", httpCommand);
}
eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
- eventMeshHTTPServer.metrics.summaryMetrics.recordHTTPReqResTimeCost(System.currentTimeMillis() - asyncContext.getRequest().getReqTime());
+ eventMeshHTTPServer.metrics.summaryMetrics.recordHTTPReqResTimeCost(
+ System.currentTimeMillis() - asyncContext.getRequest().getReqTime());
} catch (Exception ex) {
}
}
};
- responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- EventMeshRetCode.SUCCESS.getRetCode(), EventMeshRetCode.SUCCESS.getErrMsg());
+ responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(EventMeshRetCode.SUCCESS);
asyncContext.onComplete(responseEventMeshCommand, handler);
} catch (Exception e) {
HttpCommand err = asyncContext.getRequest().createHttpCommandResponse(
- heartbeatResponseHeader,
- HeartbeatResponseBody.buildBody(EventMeshRetCode.EVENTMESH_HEARTBEAT_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_HEARTBEAT_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2)));
+ heartbeatResponseHeader,
+ HeartbeatResponseBody.buildBody(EventMeshRetCode.EVENTMESH_HEARTBEAT_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_HEARTBEAT_ERR.getErrMsg() + EventMeshUtil.stackTrace(e, 2)));
asyncContext.onComplete(err);
long endTime = System.currentTimeMillis();
httpLogger.error("message|eventMesh2mq|REQ|ASYNC|heartBeatMessageCost={}ms",
- endTime - startTime, e);
+ endTime - startTime, e);
eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgFailed();
eventMeshHTTPServer.metrics.summaryMetrics.recordSendMsgCost(endTime - startTime);
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
index 7630315..5b2914d 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
@@ -27,6 +27,7 @@ import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.exception.OnExceptionContext;
import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.IPUtil;
+import org.apache.eventmesh.common.ProtocolTransportObject;
import org.apache.eventmesh.common.command.HttpCommand;
import org.apache.eventmesh.common.protocol.http.body.message.ReplyMessageRequestBody;
import org.apache.eventmesh.common.protocol.http.body.message.ReplyMessageResponseBody;
@@ -81,7 +82,7 @@ public class ReplyMessageProcessor implements HttpRequestProcessor {
// ReplyMessageRequestBody replyMessageRequestBody = (ReplyMessageRequestBody) asyncContext.getRequest().getBody();
String protocolType = replyMessageRequestHeader.getProtocolType();
- ProtocolAdaptor httpCommandProtocolAdaptor = ProtocolPluginFactory.getProtocolAdaptor(protocolType);
+ ProtocolAdaptor<ProtocolTransportObject> httpCommandProtocolAdaptor = ProtocolPluginFactory.getProtocolAdaptor(protocolType);
CloudEvent event = httpCommandProtocolAdaptor.toCloudEvent(asyncContext.getRequest());
ReplyMessageResponseHeader replyMessageResponseHeader =
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java
index 768b0cc..d3e05d1 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java
@@ -26,6 +26,7 @@ import org.apache.eventmesh.api.exception.OnExceptionContext;
import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.IPUtil;
import org.apache.eventmesh.common.LiteMessage;
+import org.apache.eventmesh.common.ProtocolTransportObject;
import org.apache.eventmesh.common.command.HttpCommand;
import org.apache.eventmesh.common.protocol.http.body.message.SendMessageRequestBody;
import org.apache.eventmesh.common.protocol.http.body.message.SendMessageResponseBody;
@@ -90,7 +91,7 @@ public class SendAsyncMessageProcessor implements HttpRequestProcessor {
eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshIDC);
String protocolType = sendMessageRequestHeader.getProtocolType();
- ProtocolAdaptor httpCommandProtocolAdaptor = ProtocolPluginFactory.getProtocolAdaptor(protocolType);
+ ProtocolAdaptor<ProtocolTransportObject> httpCommandProtocolAdaptor = ProtocolPluginFactory.getProtocolAdaptor(protocolType);
CloudEvent event = httpCommandProtocolAdaptor.toCloudEvent(asyncContext.getRequest());
//validate event
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java
index 699325f..bdae55e 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java
@@ -21,16 +21,14 @@ import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import org.apache.eventmesh.api.RRCallback;
import org.apache.eventmesh.api.RequestReplyCallback;
-import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.IPUtil;
-import org.apache.eventmesh.common.LiteMessage;
+import org.apache.eventmesh.common.ProtocolTransportObject;
import org.apache.eventmesh.common.command.HttpCommand;
import org.apache.eventmesh.common.protocol.http.body.message.SendMessageRequestBody;
import org.apache.eventmesh.common.protocol.http.body.message.SendMessageResponseBody;
import org.apache.eventmesh.common.protocol.http.common.EventMeshRetCode;
import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
-import org.apache.eventmesh.common.protocol.http.header.message.SendMessageRequestHeader;
import org.apache.eventmesh.common.protocol.http.header.message.SendMessageResponseHeader;
import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
@@ -44,7 +42,6 @@ import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.HttpRequest
import org.apache.eventmesh.runtime.core.protocol.http.producer.EventMeshProducer;
import org.apache.eventmesh.runtime.core.protocol.http.producer.SendMessageContext;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
-import org.apache.eventmesh.runtime.util.OMSUtil;
import org.apache.eventmesh.runtime.util.RemotingHelper;
import org.apache.commons.lang3.StringUtils;
@@ -84,10 +81,7 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
EventMeshConstants.PROTOCOL_HTTP,
RemotingHelper.parseChannelRemoteAddr(ctx.channel()), IPUtil.getLocalAddress());
- SendMessageRequestHeader sendMessageRequestHeader = (SendMessageRequestHeader) asyncContext.getRequest().getHeader();
-
- String protocolType = sendMessageRequestHeader.getProtocolType();
- ProtocolAdaptor httpCommandProtocolAdaptor = ProtocolPluginFactory.getProtocolAdaptor(protocolType);
+ ProtocolAdaptor<ProtocolTransportObject> httpCommandProtocolAdaptor = ProtocolPluginFactory.getProtocolAdaptor("cloudevents");
CloudEvent event = httpCommandProtocolAdaptor.toCloudEvent(asyncContext.getRequest());
SendMessageResponseHeader sendMessageResponseHeader =
@@ -106,45 +100,58 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
|| StringUtils.isBlank(event.getType())
|| StringUtils.isBlank(event.getSubject())) {
responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- sendMessageResponseHeader,
- SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(), EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
+ sendMessageResponseHeader,
+ SendMessageResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
asyncContext.onComplete(responseEventMeshCommand);
return;
}
- String idc = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.IDC)).toString();
- String pid = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.PID)).toString();
- String sys = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.SYS)).toString();
+ String idc = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.IDC))
+ .toString();
+ String pid = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.PID))
+ .toString();
+ String sys = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.SYS))
+ .toString();
//validate event-extension
if (StringUtils.isBlank(idc)
- || StringUtils.isBlank(pid)
- || !StringUtils.isNumeric(pid)
- || StringUtils.isBlank(sys)) {
+ || StringUtils.isBlank(pid)
+ || !StringUtils.isNumeric(pid)
+ || StringUtils.isBlank(sys)) {
responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- sendMessageResponseHeader,
- SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
- EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
+ sendMessageResponseHeader,
+ SendMessageResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getErrMsg()));
asyncContext.onComplete(responseEventMeshCommand);
return;
}
- String bizNo = Objects.requireNonNull(event.getExtension(SendMessageRequestBody.BIZSEQNO)).toString();
- String uniqueId = Objects.requireNonNull(event.getExtension(SendMessageRequestBody.UNIQUEID)).toString();
- String producerGroup = Objects.requireNonNull(event.getExtension(SendMessageRequestBody.PRODUCERGROUP)).toString();
+ String bizNo =
+ Objects.requireNonNull(event.getExtension(SendMessageRequestBody.BIZSEQNO)).toString();
+ String uniqueId =
+ Objects.requireNonNull(event.getExtension(SendMessageRequestBody.UNIQUEID)).toString();
+ String producerGroup =
+ Objects.requireNonNull(event.getExtension(SendMessageRequestBody.PRODUCERGROUP))
+ .toString();
String topic = event.getSubject();
- String ttl = Objects.requireNonNull(event.getExtension(SendMessageRequestBody.TTL)).toString();
+ String ttl =
+ Objects.requireNonNull(event.getExtension(SendMessageRequestBody.TTL)).toString();
//validate body
if (StringUtils.isBlank(bizNo)
- || StringUtils.isBlank(uniqueId)
- || StringUtils.isBlank(producerGroup)
- || StringUtils.isBlank(topic)
- || event.getData() != null
- || StringUtils.isBlank(ttl)) {
+ || StringUtils.isBlank(uniqueId)
+ || StringUtils.isBlank(producerGroup)
+ || StringUtils.isBlank(topic)
+ || event.getData() != null
+ || StringUtils.isBlank(ttl)) {
responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- sendMessageResponseHeader,
- SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(), EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
+ sendMessageResponseHeader,
+ SendMessageResponseBody
+ .buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
+ EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getErrMsg()));
asyncContext.onComplete(responseEventMeshCommand);
return;
}
@@ -257,10 +264,11 @@ public class SendSyncMessageProcessor implements HttpRequestProcessor {
event = CloudEventBuilder.from(event)
.withExtension(EventMeshConstants.RSP_EVENTMESH2C_TIMESTAMP,
String.valueOf(System.currentTimeMillis()))
- .withExtension(EventMeshConstants.RSP_MQ2EVENTMESH_TIMESTAMP,
- String.valueOf(System.currentTimeMillis()))
- .build();
- final String rtnMsg = new String(event.getData().toBytes(), EventMeshConstants.DEFAULT_CHARSET);
+ .withExtension(EventMeshConstants.RSP_MQ2EVENTMESH_TIMESTAMP,
+ String.valueOf(System.currentTimeMillis()))
+ .build();
+ final String rtnMsg = new String(event.getData().toBytes(),
+ EventMeshConstants.DEFAULT_CHARSET);
HttpCommand succ = asyncContext.getRequest().createHttpCommandResponse(
sendMessageResponseHeader,
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
index d91b462..32ddee1 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SubscribeProcessor.java
@@ -72,18 +72,20 @@ public class SubscribeProcessor implements HttpRequestProcessor {
public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpCommand> asyncContext)
throws Exception {
HttpCommand responseEventMeshCommand;
+ final HttpCommand request = asyncContext.getRequest();
+ final Integer requestCode = Integer.valueOf(asyncContext.getRequest().getRequestCode());
+
httpLogger.info("cmd={}|{}|client2eventMesh|from={}|to={}",
- RequestCode.get(Integer.valueOf(asyncContext.getRequest().getRequestCode())),
+ RequestCode.get(requestCode),
EventMeshConstants.PROTOCOL_HTTP,
- RemotingHelper.parseChannelRemoteAddr(ctx.channel()), IPUtil.getLocalAddress());
- SubscribeRequestHeader subscribeRequestHeader =
- (SubscribeRequestHeader) asyncContext.getRequest().getHeader();
- SubscribeRequestBody subscribeRequestBody =
- (SubscribeRequestBody) asyncContext.getRequest().getBody();
+ RemotingHelper.parseChannelRemoteAddr(ctx.channel()), IPUtil.getLocalAddress()
+ );
+ SubscribeRequestHeader subscribeRequestHeader = (SubscribeRequestHeader) request.getHeader();
+ SubscribeRequestBody subscribeRequestBody = (SubscribeRequestBody) request.getBody();
SubscribeResponseHeader subscribeResponseHeader =
SubscribeResponseHeader
- .buildHeader(Integer.valueOf(asyncContext.getRequest().getRequestCode()),
+ .buildHeader(requestCode,
eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshCluster,
IPUtil.getLocalAddress(),
eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEnv,
@@ -94,7 +96,7 @@ public class SubscribeProcessor implements HttpRequestProcessor {
|| StringUtils.isBlank(subscribeRequestHeader.getPid())
|| !StringUtils.isNumeric(subscribeRequestHeader.getPid())
|| StringUtils.isBlank(subscribeRequestHeader.getSys())) {
- responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
+ responseEventMeshCommand = request.createHttpCommandResponse(
subscribeResponseHeader,
SubscribeResponseBody
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
@@ -108,7 +110,7 @@ public class SubscribeProcessor implements HttpRequestProcessor {
|| CollectionUtils.isEmpty(subscribeRequestBody.getTopics())
|| StringUtils.isBlank(subscribeRequestBody.getConsumerGroup())) {
- responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
+ responseEventMeshCommand = request.createHttpCommandResponse(
subscribeResponseHeader,
SubscribeResponseBody
.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
@@ -124,7 +126,6 @@ public class SubscribeProcessor implements HttpRequestProcessor {
String user = subscribeRequestHeader.getUsername();
String pass = subscribeRequestHeader.getPasswd();
String subsystem = subscribeRequestHeader.getSys();
- int requestCode = Integer.valueOf(subscribeRequestHeader.getCode());
for (SubscriptionItem item : subTopicList) {
try {
Acl.doAclCheckInHttpReceive(remoteAddr, user, pass, subsystem, item.getTopic(),
@@ -232,16 +233,14 @@ public class SubscribeProcessor implements HttpRequestProcessor {
}
eventMeshHTTPServer.sendResponse(ctx, httpCommand.httpResponse());
eventMeshHTTPServer.metrics.summaryMetrics.recordHTTPReqResTimeCost(
- System.currentTimeMillis()
- - asyncContext.getRequest().getReqTime());
+ System.currentTimeMillis() - request.getReqTime());
} catch (Exception ex) {
// ignore
}
}
};
- responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- EventMeshRetCode.SUCCESS.getRetCode(), EventMeshRetCode.SUCCESS.getErrMsg());
+ responseEventMeshCommand = request.createHttpCommandResponse(EventMeshRetCode.SUCCESS);
asyncContext.onComplete(responseEventMeshCommand, handler);
} catch (Exception e) {
HttpCommand err = asyncContext.getRequest().createHttpCommandResponse(
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
index df3921d..45b567d 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/UnSubscribeProcessor.java
@@ -208,9 +208,8 @@ public class UnSubscribeProcessor implements HttpRequestProcessor {
eventMeshHTTPServer.getConsumerManager().notifyConsumerManager(consumerGroup,
eventMeshHTTPServer.localConsumerGroupMapping.get(consumerGroup));
- responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- EventMeshRetCode.SUCCESS.getRetCode(),
- EventMeshRetCode.SUCCESS.getErrMsg());
+ responseEventMeshCommand =
+ asyncContext.getRequest().createHttpCommandResponse(EventMeshRetCode.SUCCESS);
asyncContext.onComplete(responseEventMeshCommand, handler);
} catch (Exception e) {
@@ -236,9 +235,8 @@ public class UnSubscribeProcessor implements HttpRequestProcessor {
try {
eventMeshHTTPServer.getConsumerManager()
.notifyConsumerManager(consumerGroup, null);
- responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
- EventMeshRetCode.SUCCESS.getRetCode(),
- EventMeshRetCode.SUCCESS.getErrMsg());
+ responseEventMeshCommand =
+ asyncContext.getRequest().createHttpCommandResponse(EventMeshRetCode.SUCCESS);
asyncContext.onComplete(responseEventMeshCommand, handler);
// clean ClientInfo
eventMeshHTTPServer.localClientInfoMapping.keySet()
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
index 1559e69..5f31355 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java
@@ -21,6 +21,7 @@ import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.IPUtil;
+import org.apache.eventmesh.common.ProtocolTransportObject;
import org.apache.eventmesh.common.RandomStringUtil;
import org.apache.eventmesh.common.exception.JsonException;
import org.apache.eventmesh.common.protocol.SubscriptionType;
@@ -117,11 +118,13 @@ public class AsyncHTTPPushRequest extends AbstractHTTPPushRequest {
try {
String protocolType = Objects.requireNonNull(event.getExtension(Constants.PROTOCOL_TYPE)).toString();
- ProtocolAdaptor protocolAdaptor = ProtocolPluginFactory.getProtocolAdaptor(protocolType);
+ ProtocolAdaptor<ProtocolTransportObject> protocolAdaptor = ProtocolPluginFactory.getProtocolAdaptor(protocolType);
- content = (String) protocolAdaptor.fromCloudEvent(handleMsgContext.getEvent());
+ // todo
+ ProtocolTransportObject protocolTransportObject =
+ protocolAdaptor.fromCloudEvent(handleMsgContext.getEvent());
-// content =
+ // content =
// new String(handleMsgContext.getEvent().getData().toBytes(), EventMeshConstants.DEFAULT_CHARSET);
} catch (Exception ex) {
return;
@@ -300,13 +303,11 @@ public class AsyncHTTPPushRequest extends AbstractHTTPPushRequest {
waitingRequests
.put(request.handleMsgContext.getConsumerGroup(), Sets.newConcurrentHashSet());
waitingRequests.get(request.handleMsgContext.getConsumerGroup()).add(request);
- return;
}
private void removeWaitingMap(AsyncHTTPPushRequest request) {
if (waitingRequests.containsKey(request.handleMsgContext.getConsumerGroup())) {
waitingRequests.get(request.handleMsgContext.getConsumerGroup()).remove(request);
- return;
}
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/MessageTransferTask.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/MessageTransferTask.java
index 176fe5d..4c595e1 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/MessageTransferTask.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/MessageTransferTask.java
@@ -27,6 +27,7 @@ import org.apache.eventmesh.api.SendCallback;
import org.apache.eventmesh.api.SendResult;
import org.apache.eventmesh.api.exception.OnExceptionContext;
import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.ProtocolTransportObject;
import org.apache.eventmesh.common.protocol.tcp.Package;
import org.apache.eventmesh.common.protocol.tcp.*;
import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
@@ -67,7 +68,7 @@ public class MessageTransferTask extends AbstractTask {
if (pkg.getHeader().getProperties() != null && pkg.getHeader().getProperty(Constants.PROTOCOL_TYPE) != null) {
protocolType = (String) pkg.getHeader().getProperty(Constants.PROTOCOL_TYPE);
}
- ProtocolAdaptor protocolAdaptor = ProtocolPluginFactory.getProtocolAdaptor(protocolType);
+ ProtocolAdaptor<ProtocolTransportObject> protocolAdaptor = ProtocolPluginFactory.getProtocolAdaptor(protocolType);
Package msg = new Package();
// EventMeshMessage eventMeshMessage = (EventMeshMessage) pkg.getBody();
@@ -188,16 +189,6 @@ public class MessageTransferTask extends AbstractTask {
Utils.writeAndFlush(msg, startTime, taskExecuteTime, session.getContext(), session);
}
-// @Override
-// public void onException(Throwable e) {
-// session.getSender().getUpstreamBuff().release();
-// session.getSender().failMsgCount.incrementAndGet();
-// messageLogger.error("upstreamMsg mq message error|user={}|callback cost={}, errMsg={}", session.getClient(), String.valueOf
-// (System.currentTimeMillis() - createTime), new Exception(e));
-// msg.setHeader(new Header(replyCmd, OPStatus.FAIL.getCode(), e.toString(), pkg.getHeader().getSeq()));
-// msg.setBody(accessMessage);
-// Utils.writeAndFlush(msg, startTime, taskExecuteTime, session.getContext(), session);
-// }
};
}
}
diff --git a/eventmesh-sdk-java/build.gradle b/eventmesh-sdk-java/build.gradle
index 0db0df4..f62840d 100644
--- a/eventmesh-sdk-java/build.gradle
+++ b/eventmesh-sdk-java/build.gradle
@@ -38,4 +38,10 @@ dependencies {
testImplementation "io.netty:netty-all"
testImplementation "org.apache.httpcomponents:httpclient"
+ compileOnly 'org.projectlombok:lombok:1.18.22'
+ annotationProcessor 'org.projectlombok:lombok:1.18.22'
+
+ testCompileOnly 'org.projectlombok:lombok:1.18.22'
+ testAnnotationProcessor 'org.projectlombok:lombok:1.18.22'
+
}
diff --git a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/AbstractLiteClient.java b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/AbstractLiteClient.java
index 20302af..df72fac 100644
--- a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/AbstractLiteClient.java
+++ b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/AbstractLiteClient.java
@@ -17,6 +17,8 @@
package org.apache.eventmesh.client.http;
+import lombok.extern.slf4j.Slf4j;
+
import org.apache.eventmesh.client.http.conf.LiteClientConfig;
import org.apache.eventmesh.client.http.ssl.MyX509TrustManager;
import org.apache.eventmesh.client.http.util.HttpLoadBalanceUtils;
@@ -30,10 +32,9 @@ import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManager;
import java.security.SecureRandom;
+@Slf4j
public abstract class AbstractLiteClient {
- public Logger logger = LoggerFactory.getLogger(AbstractLiteClient.class);
-
protected LiteClientConfig liteClientConfig;
protected LoadBalanceSelector<String> eventMeshServerSelector;
@@ -51,7 +52,7 @@ public abstract class AbstractLiteClient {
}
public void shutdown() throws Exception {
- logger.info("AbstractLiteClient shutdown");
+ log.info("AbstractLiteClient shutdown");
}
public CloseableHttpClient setHttpClient() throws Exception {
@@ -67,7 +68,7 @@ public abstract class AbstractLiteClient {
return HttpClients.custom().setSSLContext(sslContext)
.setSSLHostnameVerifier(new DefaultHostnameVerifier()).build();
} catch (Exception e) {
- logger.error("Error in creating HttpClient.", e);
+ log.error("Error in creating HttpClient.", e);
throw e;
}
}
diff --git a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/RemotingServer.java b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/RemotingServer.java
deleted file mode 100644
index 3c98d81..0000000
--- a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/RemotingServer.java
+++ /dev/null
@@ -1,424 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.eventmesh.client.http;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import io.netty.bootstrap.ServerBootstrap;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelFutureListener;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelOption;
-import io.netty.channel.EventLoopGroup;
-import io.netty.channel.SimpleChannelInboundHandler;
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.channel.socket.SocketChannel;
-import io.netty.channel.socket.nio.NioServerSocketChannel;
-import io.netty.handler.codec.http.DefaultFullHttpResponse;
-import io.netty.handler.codec.http.FullHttpResponse;
-import io.netty.handler.codec.http.HttpHeaderNames;
-import io.netty.handler.codec.http.HttpHeaderValues;
-import io.netty.handler.codec.http.HttpMethod;
-import io.netty.handler.codec.http.HttpObjectAggregator;
-import io.netty.handler.codec.http.HttpRequest;
-import io.netty.handler.codec.http.HttpRequestDecoder;
-import io.netty.handler.codec.http.HttpResponseEncoder;
-import io.netty.handler.codec.http.HttpResponseStatus;
-import io.netty.handler.codec.http.HttpVersion;
-import io.netty.handler.codec.http.QueryStringDecoder;
-import io.netty.handler.codec.http.multipart.Attribute;
-import io.netty.handler.codec.http.multipart.DefaultHttpDataFactory;
-import io.netty.handler.codec.http.multipart.HttpPostRequestDecoder;
-import io.netty.handler.codec.http.multipart.InterfaceHttpData;
-
-import org.apache.commons.collections4.MapUtils;
-import org.apache.commons.lang3.RandomUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.eventmesh.client.http.consumer.HandleResult;
-import org.apache.eventmesh.client.http.consumer.context.LiteConsumeContext;
-import org.apache.eventmesh.client.http.consumer.listener.LiteMessageListener;
-import org.apache.eventmesh.common.Constants;
-import org.apache.eventmesh.common.IPUtil;
-import org.apache.eventmesh.common.LiteMessage;
-import org.apache.eventmesh.common.ThreadUtil;
-import org.apache.eventmesh.common.command.HttpCommand;
-import org.apache.eventmesh.common.protocol.http.body.Body;
-import org.apache.eventmesh.common.protocol.http.body.message.PushMessageRequestBody;
-import org.apache.eventmesh.common.protocol.http.common.ClientRetCode;
-import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
-import org.apache.eventmesh.common.protocol.http.common.ProtocolVersion;
-import org.apache.eventmesh.common.protocol.http.common.RequestCode;
-import org.apache.eventmesh.common.protocol.http.header.Header;
-import org.apache.eventmesh.common.protocol.http.header.message.PushMessageRequestHeader;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class RemotingServer {
-
- public Logger logger = LoggerFactory.getLogger(this.getClass());
-
- public AtomicBoolean started = new AtomicBoolean(Boolean.FALSE);
-
- public AtomicBoolean inited = new AtomicBoolean(Boolean.FALSE);
-
- private EventLoopGroup bossGroup;
-
- private EventLoopGroup workerGroup;
-
- private int port = RandomUtils.nextInt(1000, 20000);
-
- private DefaultHttpDataFactory defaultHttpDataFactory = new DefaultHttpDataFactory(false);
-
- private ThreadPoolExecutor consumeExecutor;
-
- private LiteMessageListener messageListener;
-
- public RemotingServer() {
- }
-
- public RemotingServer(int port) {
- this.port = port;
- }
-
- public RemotingServer(ThreadPoolExecutor consumeExecutor) {
- this.consumeExecutor = consumeExecutor;
- }
-
- public RemotingServer(int port, ThreadPoolExecutor consumeExecutor) {
- this.port = port;
- this.consumeExecutor = consumeExecutor;
- }
-
- public void setPort(int port) {
- this.port = port;
- }
-
- public void setConsumeExecutor(ThreadPoolExecutor consumeExecutor) {
- this.consumeExecutor = consumeExecutor;
- }
-
- // TODO: Let different topics have different listeners
- public void registerMessageListener(LiteMessageListener eventMeshMessageListener) {
- this.messageListener = eventMeshMessageListener;
- }
-
- private EventLoopGroup initBossGroup() {
- bossGroup = new NioEventLoopGroup(1, new ThreadFactory() {
- AtomicInteger count = new AtomicInteger(0);
-
- @Override
- public Thread newThread(Runnable r) {
- Thread t = new Thread(r, "endPointBoss-" + count.incrementAndGet());
- t.setDaemon(true);
- return t;
- }
- });
-
- return bossGroup;
- }
-
- private EventLoopGroup initWokerGroup() {
- workerGroup = new NioEventLoopGroup(2, new ThreadFactory() {
- AtomicInteger count = new AtomicInteger(0);
-
- @Override
- public Thread newThread(Runnable r) {
- Thread t = new Thread(r, "endpointWorker-" + count.incrementAndGet());
- return t;
- }
- });
- return workerGroup;
- }
-
- public String getEndpointURL() {
- return String.format("http://%s:%s", IPUtil.getLocalAddress(), port);
- }
-
-
- class HTTPHandler extends SimpleChannelInboundHandler<HttpRequest> {
-
- /**
- * Parse request HEADER
- *
- * @param fullReq
- * @return
- */
- private Map<String, Object> parseHTTPHeader(HttpRequest fullReq) {
- Map<String, Object> headerParam = new HashMap<>();
- for (String key : fullReq.headers().names()) {
- if (StringUtils.equalsIgnoreCase(HttpHeaderNames.CONTENT_TYPE.toString(), key)
- || StringUtils.equalsIgnoreCase(HttpHeaderNames.ACCEPT_ENCODING.toString(), key)
- || StringUtils.equalsIgnoreCase(HttpHeaderNames.CONTENT_LENGTH.toString(), key)) {
- continue;
- }
- headerParam.put(key, fullReq.headers().get(key));
- }
- return headerParam;
- }
-
- @Override
- protected void channelRead0(final ChannelHandlerContext ctx, HttpRequest httpRequest) throws Exception {
- HttpPostRequestDecoder decoder = null;
- try {
- if (!httpRequest.decoderResult().isSuccess()) {
- sendError(ctx, HttpResponseStatus.BAD_REQUEST);
- return;
- }
-
- // Protocol version verification
- String protocolVersion = StringUtils.deleteWhitespace(httpRequest.headers().get(ProtocolKey.VERSION));
- if (StringUtils.isBlank(protocolVersion) || !ProtocolVersion.contains(protocolVersion)) {
- httpRequest.headers().set(ProtocolKey.VERSION, ProtocolVersion.V1.getVersion());
- }
-
- Map<String, Object> bodyMap = new HashMap<>();
-
- if (httpRequest.method() == HttpMethod.GET) {
- QueryStringDecoder getDecoder = new QueryStringDecoder(httpRequest.uri());
- for (Map.Entry<String, List<String>> entry : getDecoder.parameters().entrySet()) {
- bodyMap.put(entry.getKey(), entry.getValue().get(0));
- }
- } else if (httpRequest.method() == HttpMethod.POST) {
- decoder = new HttpPostRequestDecoder(defaultHttpDataFactory, httpRequest);
- List<InterfaceHttpData> parmList = decoder.getBodyHttpDatas();
- for (InterfaceHttpData parm : parmList) {
- if (parm.getHttpDataType() == InterfaceHttpData.HttpDataType.Attribute) {
- Attribute data = (Attribute) parm;
- bodyMap.put(data.getName(), data.getValue());
- }
- }
- } else {
- sendError(ctx, HttpResponseStatus.METHOD_NOT_ALLOWED);
- return;
- }
-
- /////////////////////////////////////////////////////////////////Basic inspection////////////////////////////////////////////////////
- String requestCode =
- (httpRequest.method() == HttpMethod.POST) ? StringUtils.deleteWhitespace(httpRequest.headers().get(ProtocolKey.REQUEST_CODE))
- : MapUtils.getString(bodyMap, StringUtils.lowerCase(ProtocolKey.REQUEST_CODE), "");
-
- final HttpCommand requestCommand = new HttpCommand(
- httpRequest.method().name(),
- httpRequest.protocolVersion().protocolName(), requestCode);
-
- HttpCommand responseCommand;
-
- // Verify requestCode
- if (StringUtils.isBlank(requestCode)
- || !StringUtils.isNumeric(requestCode)
- || (!String.valueOf(RequestCode.HTTP_PUSH_CLIENT_ASYNC.getRequestCode()).equals(requestCode)
- && !String.valueOf(RequestCode.HTTP_PUSH_CLIENT_SYNC.getRequestCode()).equals(requestCode))) {
- logger.error("receive invalid requestCode, {}", requestCode);
- responseCommand = requestCommand.createHttpCommandResponse(ClientRetCode.OK.getRetCode(), ClientRetCode.OK.getErrMsg());
- sendResponse(ctx, responseCommand.httpResponse());
- return;
- }
-
- requestCommand.setHeader(Header.buildHeader(requestCode, parseHTTPHeader(httpRequest)));
- requestCommand.setBody(Body.buildBody(requestCode, bodyMap));
-
- if (logger.isDebugEnabled()) {
- logger.debug("{}", requestCommand);
- }
-
- PushMessageRequestHeader pushMessageRequestHeader = (PushMessageRequestHeader) requestCommand.header;
- PushMessageRequestBody pushMessageRequestBody = (PushMessageRequestBody) requestCommand.body;
-
- String topic = pushMessageRequestBody.getTopic();
-
- // Check if there is a listener for the TOPIC
-// if (!listenerTable.containsKey(topic)) {
-// logger.error("no listenning for this topic, {}", topic);
-// responseCommand = requestCommand.createHttpCommandResponse(ClientRetCode.NOLISTEN.getRetCode(), ClientRetCode.NOLISTEN.getErrMsg());
-// sendResponse(ctx, responseCommand.httpResponse());
-// return;
-// }
-
- final LiteConsumeContext eventMeshConsumeContext = new LiteConsumeContext(pushMessageRequestHeader.getEventMeshIp(),
- pushMessageRequestHeader.getEventMeshEnv(), pushMessageRequestHeader.getEventMeshIdc(),
- pushMessageRequestHeader.getEventMeshCluster());
-
- final LiteMessage liteMessage = new LiteMessage(pushMessageRequestBody.getBizSeqNo(), pushMessageRequestBody.getUniqueId(),
- topic, pushMessageRequestBody.getContent());
-
- for (Map.Entry<String, String> entry : pushMessageRequestBody.getExtFields().entrySet()) {
- liteMessage.addProp(entry.getKey(), entry.getValue());
- }
-
- // Transfer to the consumer thread pool
- consumeExecutor.execute(new Runnable() {
- @Override
- public void run() {
- try {
- if (messageListener.reject()) {
- HttpCommand responseCommand = requestCommand.createHttpCommandResponse(handleResult2ClientRetCode(HandleResult.RETRY).getRetCode(), handleResult2ClientRetCode(HandleResult.RETRY).getErrMsg());
- sendResponse(ctx, responseCommand.httpResponse());
- return;
- }
-
- HandleResult handleResult = messageListener.handle(liteMessage, eventMeshConsumeContext);
-
- if (logger.isDebugEnabled()) {
- logger.info("bizSeqNo:{}, topic:{}, handleResult:{}", liteMessage.getBizSeqNo(), liteMessage.getTopic(), handleResult);
- }
-
- HttpCommand responseCommand = requestCommand.createHttpCommandResponse(handleResult2ClientRetCode(handleResult).getRetCode(), handleResult2ClientRetCode(handleResult).getErrMsg());
- sendResponse(ctx, responseCommand.httpResponse());
- } catch (Exception e) {
- logger.error("process error", e);
- }
- }
- });
- } catch (Exception ex) {
- logger.error("HTTPHandler.channelRead0 err", ex);
- } finally {
- try {
- decoder.destroy();
- } catch (Exception e) {
- }
- }
- }
-
- public ClientRetCode handleResult2ClientRetCode(HandleResult handleResult) {
- if (handleResult == HandleResult.OK) {
- return ClientRetCode.OK;
- } else if (handleResult == HandleResult.FAIL) {
- return ClientRetCode.FAIL;
- } else if (handleResult == HandleResult.NOLISTEN) {
- return ClientRetCode.NOLISTEN;
- } else if (handleResult == HandleResult.RETRY) {
- return ClientRetCode.RETRY;
- } else {
- return ClientRetCode.OK;
- }
- }
-
- @Override
- public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
- super.channelReadComplete(ctx);
- ctx.flush(); // 4
- }
-
- @Override
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
- if (null != cause) cause.printStackTrace();
- if (null != ctx) ctx.close();
- }
-
- /**
- * Send default error page
- *
- * @param ctx
- * @param status
- */
- private void sendError(ChannelHandlerContext ctx,
- HttpResponseStatus status) {
- FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
- status);
- response.headers().add(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN +
- "; charset=" + Constants.DEFAULT_CHARSET);
- response.headers().add(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
- response.headers().add(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
- ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
- }
-
-
- /**
- * Send response
- *
- * @param ctx
- * @param response
- */
- private void sendResponse(ChannelHandlerContext ctx,
- DefaultFullHttpResponse response) {
- ctx.writeAndFlush(response).addListener(new ChannelFutureListener() {
- @Override
- public void operationComplete(ChannelFuture f) throws Exception {
- if (!f.isSuccess()) {
- logger.warn("send response to [{}] fail, will close this channel", IPUtil.parseChannelRemoteAddr(f.channel()));
- f.channel().close();
- return;
- }
- }
- });
- }
-
- public void shutdown() throws Exception {
- if (bossGroup != null) {
- bossGroup.shutdownGracefully();
- }
-
- ThreadUtil.randomSleep(30);
-
- if (workerGroup != null) {
- workerGroup.shutdownGracefully();
- }
-
- started.compareAndSet(true, false);
- inited.compareAndSet(true, false);
- }
- }
-
- public void init() throws Exception {
- initBossGroup();
- initWokerGroup();
- inited.compareAndSet(false, true);
- }
-
- public void start() throws Exception {
- Runnable r = new Runnable() {
- @Override
- public void run() {
- ServerBootstrap b = new ServerBootstrap();
- b.group(bossGroup, workerGroup)
- .channel(NioServerSocketChannel.class)
- .childHandler(new ChannelInitializer<SocketChannel>() {
- @Override
- public void initChannel(SocketChannel ch)
- throws Exception {
- ch.pipeline()
- .addLast(new HttpRequestDecoder(),
- new HttpResponseEncoder(),
- new HttpObjectAggregator(Integer.MAX_VALUE),
- new HTTPHandler()); // 4
- }
- }).childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE);
- try {
- logger.info("EventMesh Client[{}] Started......", port);
- ChannelFuture future = b.bind(port).sync();
- future.channel().closeFuture().sync();
- started.compareAndSet(false, true);
- } catch (Exception e) {
- bossGroup.shutdownGracefully();
- workerGroup.shutdownGracefully();
- }
- }
- };
-
-
- Thread t = new Thread(r, "eventMesh-client-remoting-server");
- t.start();
- }
-}
diff --git a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/consumer/LiteConsumer.java b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/consumer/LiteConsumer.java
index 7915d40..2dad1c6 100644
--- a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/consumer/LiteConsumer.java
+++ b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/consumer/LiteConsumer.java
@@ -19,13 +19,10 @@ package org.apache.eventmesh.client.http.consumer;
import org.apache.eventmesh.client.http.AbstractLiteClient;
import org.apache.eventmesh.client.http.EventMeshRetObj;
-import org.apache.eventmesh.client.http.RemotingServer;
import org.apache.eventmesh.client.http.conf.LiteClientConfig;
-import org.apache.eventmesh.client.http.consumer.listener.LiteMessageListener;
import org.apache.eventmesh.client.http.http.HttpUtil;
import org.apache.eventmesh.client.http.http.RequestParam;
import org.apache.eventmesh.client.tcp.common.EventMeshCommon;
-import org.apache.eventmesh.client.tcp.common.EventMeshThreadFactoryImpl;
import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.EventMeshException;
import org.apache.eventmesh.common.ThreadPoolFactory;
@@ -44,7 +41,6 @@ import org.apache.http.impl.client.CloseableHttpClient;
import java.util.ArrayList;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -52,39 +48,34 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.netty.handler.codec.http.HttpMethod;
+import lombok.extern.slf4j.Slf4j;
+@Slf4j
public class LiteConsumer extends AbstractLiteClient {
- public Logger logger = LoggerFactory.getLogger(LiteConsumer.class);
-
- private RemotingServer remotingServer;
-
private ThreadPoolExecutor consumeExecutor;
protected LiteClientConfig eventMeshClientConfig;
- private List<SubscriptionItem> subscription = Lists.newArrayList();
+ private static final List<SubscriptionItem> subscription = Lists.newArrayList();
- private LiteMessageListener messageListener;
+ private static final AtomicBoolean started = new AtomicBoolean(Boolean.FALSE);
- protected final ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(4,
- new EventMeshThreadFactoryImpl("TCPClientScheduler", true));
+ protected final ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(
+ Runtime.getRuntime().availableProcessors(),
+ new ThreadFactoryBuilder().setNameFormat("TCPClientScheduler").setDaemon(true).build()
+ );
- public LiteConsumer(LiteClientConfig liteClientConfig) throws Exception {
- super(liteClientConfig);
- this.consumeExecutor =
+ public LiteConsumer(LiteClientConfig liteClientConfig) {
+ this(liteClientConfig,
ThreadPoolFactory.createThreadPoolExecutor(liteClientConfig.getConsumeThreadCore(),
- liteClientConfig.getConsumeThreadMax(), "eventMesh-client-consume-");
- this.eventMeshClientConfig = liteClientConfig;
-// this.remotingServer = new RemotingServer(10106, consumeExecutor);
-// this.remotingServer.init();
+ liteClientConfig.getConsumeThreadMax(), "EventMesh-client-consume-")
+ );
}
public LiteConsumer(LiteClientConfig liteClientConfig,
@@ -92,36 +83,39 @@ public class LiteConsumer extends AbstractLiteClient {
super(liteClientConfig);
this.consumeExecutor = customExecutor;
this.eventMeshClientConfig = liteClientConfig;
-// this.remotingServer = new RemotingServer(this.consumeExecutor);
}
- private final AtomicBoolean started = new AtomicBoolean(Boolean.FALSE);
-
@Override
public void start() throws Exception {
- Preconditions
- .checkState(eventMeshClientConfig != null, "eventMeshClientConfig can't be null");
- Preconditions.checkState(consumeExecutor != null, "consumeExecutor can't be null");
-// Preconditions.checkState(messageListener != null, "messageListener can't be null");
- logger.info("LiteConsumer starting");
+ Preconditions.checkNotNull(eventMeshClientConfig,
+ "EventMeshClientConfig can't be null");
+ Preconditions.checkNotNull(consumeExecutor, "consumeExecutor can't be null");
+ log.info("LiteConsumer starting");
super.start();
started.compareAndSet(false, true);
- logger.info("LiteConsumer started");
-// this.remotingServer.start();
+ log.info("LiteConsumer started");
}
@Override
public void shutdown() throws Exception {
- logger.info("LiteConsumer shutting down");
+ log.info("LiteConsumer shutting down");
super.shutdown();
if (consumeExecutor != null) {
consumeExecutor.shutdown();
}
scheduler.shutdown();
started.compareAndSet(true, false);
- logger.info("LiteConsumer shutdown");
+ log.info("LiteConsumer shutdown");
}
+ /**
+ * When receive message will callback the url.
+ *
+ * @param topicList topic that be subscribed
+ * @param url url will be trigger
+ * @return true if subscribe success
+ * @throws Exception
+ */
public boolean subscribe(List<SubscriptionItem> topicList, String url) throws Exception {
subscription.addAll(topicList);
if (!started.get()) {
@@ -130,7 +124,7 @@ public class LiteConsumer extends AbstractLiteClient {
RequestParam subscribeParam = generateSubscribeRequestParam(topicList, url);
- long startTime = System.currentTimeMillis();
+ long startTime = System.nanoTime();
String target = selectEventMesh();
String subRes = "";
@@ -138,17 +132,18 @@ public class LiteConsumer extends AbstractLiteClient {
subRes = HttpUtil.post(httpClient, target, subscribeParam);
}
- if (logger.isDebugEnabled()) {
- logger.debug(
+ if (log.isDebugEnabled()) {
+ log.debug(
"subscribe message by await, targetEventMesh:{}, cost:{}ms, subscribeParam:{}, "
- + "rtn:{}", target, System.currentTimeMillis() - startTime,
+ + "rtn:{}", target, (System.nanoTime() - startTime) / 1000000,
JsonUtils.serialize(subscribeParam), subRes);
}
EventMeshRetObj ret = JsonUtils.deserialize(subRes, EventMeshRetObj.class);
if (ret.getRetCode() == EventMeshRetCode.SUCCESS.getRetCode()) {
- return Boolean.TRUE;
+ // todo: remove return result
+ return true;
} else {
throw new EventMeshException(ret.getRetCode(), ret.getRetMsg());
}
@@ -157,13 +152,9 @@ public class LiteConsumer extends AbstractLiteClient {
private RequestParam generateSubscribeRequestParam(List<SubscriptionItem> topicList,
String url) {
-// final LiteMessage liteMessage = new LiteMessage();
-// liteMessage.setBizSeqNo(RandomStringUtils.randomNumeric(30))
-// .setContent("subscribe message")
-// .setUniqueId(RandomStringUtils.randomNumeric(30));
RequestParam requestParam = new RequestParam(HttpMethod.POST);
- requestParam.addHeader(ProtocolKey.REQUEST_CODE,
- String.valueOf(RequestCode.SUBSCRIBE.getRequestCode()))
+ requestParam
+ .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.SUBSCRIBE.getRequestCode())
.addHeader(ProtocolKey.ClientInstanceKey.ENV, eventMeshClientConfig.getEnv())
.addHeader(ProtocolKey.ClientInstanceKey.IDC, eventMeshClientConfig.getIdc())
.addHeader(ProtocolKey.ClientInstanceKey.IP, eventMeshClientConfig.getIp())
@@ -191,8 +182,8 @@ public class LiteConsumer extends AbstractLiteClient {
}
RequestParam requestParam = new RequestParam(HttpMethod.POST);
- requestParam.addHeader(ProtocolKey.REQUEST_CODE,
- String.valueOf(RequestCode.HEARTBEAT.getRequestCode()))
+ requestParam
+ .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.HEARTBEAT.getRequestCode())
.addHeader(ProtocolKey.ClientInstanceKey.ENV, eventMeshClientConfig.getEnv())
.addHeader(ProtocolKey.ClientInstanceKey.IDC, eventMeshClientConfig.getIdc())
.addHeader(ProtocolKey.ClientInstanceKey.IP, eventMeshClientConfig.getIp())
@@ -220,7 +211,7 @@ public class LiteConsumer extends AbstractLiteClient {
}
RequestParam requestParam = generateHeartBeatRequestParam(topicList, url);
- long startTime = System.currentTimeMillis();
+ long startTime = System.nanoTime();
String target = selectEventMesh();
String res = "";
@@ -228,10 +219,10 @@ public class LiteConsumer extends AbstractLiteClient {
res = HttpUtil.post(httpClient, target, requestParam);
}
- if (logger.isDebugEnabled()) {
- logger.debug(
+ if (log.isDebugEnabled()) {
+ log.debug(
"heartBeat message by await, targetEventMesh:{}, cost:{}ms, rtn:{}",
- target, System.currentTimeMillis() - startTime, res);
+ target, (System.nanoTime() - startTime) / 1000000, res);
}
EventMeshRetObj ret = JsonUtils.deserialize(res, EventMeshRetObj.class);
@@ -240,7 +231,7 @@ public class LiteConsumer extends AbstractLiteClient {
throw new EventMeshException(ret.getRetCode(), ret.getRetMsg());
}
} catch (Exception e) {
- logger.error("send heartBeat error", e);
+ log.error("send heartBeat error", e);
}
}
}, EventMeshCommon.HEARTBEAT, EventMeshCommon.HEARTBEAT, TimeUnit.MILLISECONDS);
@@ -248,17 +239,11 @@ public class LiteConsumer extends AbstractLiteClient {
public boolean unsubscribe(List<String> topicList, String url) throws Exception {
Set<String> unSub = new HashSet<>(topicList);
- Iterator<SubscriptionItem> itr = subscription.iterator();
- while (itr.hasNext()) {
- SubscriptionItem item = itr.next();
- if (unSub.contains(item.getTopic())) {
- itr.remove();
- }
- }
+ subscription.removeIf(item -> unSub.contains(item.getTopic()));
RequestParam unSubscribeParam = generateUnSubscribeRequestParam(topicList, url);
- long startTime = System.currentTimeMillis();
+ long startTime = System.nanoTime();
String target = selectEventMesh();
String unSubRes = "";
@@ -266,17 +251,17 @@ public class LiteConsumer extends AbstractLiteClient {
unSubRes = HttpUtil.post(httpClient, target, unSubscribeParam);
}
- if (logger.isDebugEnabled()) {
- logger.debug(
+ if (log.isDebugEnabled()) {
+ log.debug(
"unSubscribe message by await, targetEventMesh:{}, cost:{}ms, unSubscribeParam:{}, "
- + "rtn:{}", target, System.currentTimeMillis() - startTime,
+ + "rtn:{}", target, (System.nanoTime() - startTime) / 1000000,
JsonUtils.serialize(unSubscribeParam), unSubRes);
}
EventMeshRetObj ret = JsonUtils.deserialize(unSubRes, EventMeshRetObj.class);
if (ret.getRetCode() == EventMeshRetCode.SUCCESS.getRetCode()) {
- return Boolean.TRUE;
+ return true;
} else {
throw new EventMeshException(ret.getRetCode(), ret.getRetMsg());
}
@@ -284,8 +269,8 @@ public class LiteConsumer extends AbstractLiteClient {
private RequestParam generateUnSubscribeRequestParam(List<String> topicList, String url) {
RequestParam requestParam = new RequestParam(HttpMethod.POST);
- requestParam.addHeader(ProtocolKey.REQUEST_CODE,
- String.valueOf(RequestCode.UNSUBSCRIBE.getRequestCode()))
+ requestParam
+ .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.UNSUBSCRIBE.getRequestCode())
.addHeader(ProtocolKey.ClientInstanceKey.ENV, eventMeshClientConfig.getEnv())
.addHeader(ProtocolKey.ClientInstanceKey.IDC, eventMeshClientConfig.getIdc())
.addHeader(ProtocolKey.ClientInstanceKey.IP, eventMeshClientConfig.getIp())
@@ -302,12 +287,6 @@ public class LiteConsumer extends AbstractLiteClient {
return requestParam;
}
- public void registerMessageListener(LiteMessageListener messageListener)
- throws EventMeshException {
- this.messageListener = messageListener;
- remotingServer.registerMessageListener(this.messageListener);
- }
-
public String selectEventMesh() {
if (liteClientConfig.isUseTls()) {
return Constants.HTTPS_PROTOCOL_PREFIX + eventMeshServerSelector.select();
diff --git a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/http/RequestParam.java b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/http/RequestParam.java
index 4d23222..de8c82f 100644
--- a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/http/RequestParam.java
+++ b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/http/RequestParam.java
@@ -24,14 +24,14 @@ import java.util.HashMap;
import java.util.Map;
import io.netty.handler.codec.http.HttpMethod;
+import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@Slf4j
public class RequestParam {
- public Logger logger = LoggerFactory.getLogger(this.getClass());
-
private Map<String, String[]> queryParams;
private HttpMethod httpMethod;
@@ -90,7 +90,7 @@ public class RequestParam {
}
}
} catch (UnsupportedEncodingException e) {
- logger.error("get query params failed.", e);
+ log.error("get query params failed.", e);
return "";
}
return stringBuilder.substring(1);
@@ -113,11 +113,11 @@ public class RequestParam {
return this;
}
- public RequestParam addHeader(String key, String value) {
+ public RequestParam addHeader(String key, Object value) {
if (headers == null) {
headers = new HashMap<>();
}
- headers.put(key, value);
+ headers.put(key, value.toString());
return this;
}
diff --git a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/LiteProducer.java b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/LiteProducer.java
index 376e6e6..0fadad6 100644
--- a/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/LiteProducer.java
+++ b/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/http/producer/LiteProducer.java
@@ -45,29 +45,30 @@ import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
import io.netty.handler.codec.http.HttpMethod;
+import lombok.extern.slf4j.Slf4j;
+@Slf4j
public class LiteProducer extends AbstractLiteClient {
- public Logger logger = LoggerFactory.getLogger(LiteProducer.class);
-
public LiteProducer(LiteClientConfig liteClientConfig) {
super(liteClientConfig);
}
- private AtomicBoolean started = new AtomicBoolean(Boolean.FALSE);
+ private static final AtomicBoolean started = new AtomicBoolean(Boolean.FALSE);
@Override
public void start() throws Exception {
- Preconditions.checkState(liteClientConfig != null, "liteClientConfig can't be null");
- Preconditions.checkState(liteClientConfig.getLiteEventMeshAddr() != null,
- "liteClientConfig.liteServerAddr can't be null");
+ Preconditions.checkNotNull(liteClientConfig, "liteClientConfig can't be null");
+ Preconditions.checkNotNull(
+ liteClientConfig.getLiteEventMeshAddr(), "liteClientConfig.liteServerAddr can't be null"
+ );
if (started.get()) {
return;
}
- logger.info("LiteProducer starting");
+ log.info("LiteProducer starting");
super.start();
started.compareAndSet(false, true);
- logger.info("LiteProducer started");
+ log.info("LiteProducer started");
}
@Override
@@ -75,10 +76,10 @@ public class LiteProducer extends AbstractLiteClient {
if (!started.get()) {
return;
}
- logger.info("LiteProducer shutting down");
+ log.info("LiteProducer shutting down");
super.shutdown();
started.compareAndSet(true, false);
- logger.info("LiteProducer shutdown");
+ log.info("LiteProducer shutdown");
}
public AtomicBoolean getStarted() {
@@ -94,8 +95,8 @@ public class LiteProducer extends AbstractLiteClient {
Preconditions.checkState(StringUtils.isNotBlank(message.getContent()),
"eventMeshMessage[content] invalid");
RequestParam requestParam = new RequestParam(HttpMethod.POST);
- requestParam.addHeader(ProtocolKey.REQUEST_CODE,
- String.valueOf(RequestCode.MSG_SEND_ASYNC.getRequestCode()))
+ requestParam
+ .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_ASYNC.getRequestCode())
.addHeader(ProtocolKey.ClientInstanceKey.ENV, liteClientConfig.getEnv())
.addHeader(ProtocolKey.ClientInstanceKey.IDC, liteClientConfig.getIdc())
.addHeader(ProtocolKey.ClientInstanceKey.IP, liteClientConfig.getIp())
@@ -109,12 +110,11 @@ public class LiteProducer extends AbstractLiteClient {
.addBody(SendMessageRequestBody.PRODUCERGROUP, liteClientConfig.getProducerGroup())
.addBody(SendMessageRequestBody.TOPIC, message.getTopic())
.addBody(SendMessageRequestBody.CONTENT, message.getContent())
- .addBody(SendMessageRequestBody.TTL,
- message.getPropKey(Constants.EVENTMESH_MESSAGE_CONST_TTL))
+ .addBody(SendMessageRequestBody.TTL, message.getPropKey(Constants.EVENTMESH_MESSAGE_CONST_TTL))
.addBody(SendMessageRequestBody.BIZSEQNO, message.getBizSeqNo())
.addBody(SendMessageRequestBody.UNIQUEID, message.getUniqueId());
- long startTime = System.currentTimeMillis();
+ long startTime = System.nanoTime();
String target = selectEventMesh();
String res = "";
@@ -122,15 +122,15 @@ public class LiteProducer extends AbstractLiteClient {
res = HttpUtil.post(httpClient, target, requestParam);
}
- if (logger.isDebugEnabled()) {
- logger.debug("publish async message, targetEventMesh:{}, cost:{}ms, message:{}, rtn:{}",
- target, System.currentTimeMillis() - startTime, message, res);
+ if (log.isDebugEnabled()) {
+ log.debug("publish async message, targetEventMesh:{}, cost:{}ms, message:{}, rtn:{}",
+ target, (System.nanoTime() - startTime) / 1000000, message, res);
}
EventMeshRetObj ret = JsonUtils.deserialize(res, EventMeshRetObj.class);
if (ret.getRetCode() == EventMeshRetCode.SUCCESS.getRetCode()) {
- return Boolean.TRUE;
+ return true;
} else {
throw new EventMeshException(ret.getRetCode(), ret.getRetMsg());
}
@@ -153,8 +153,8 @@ public class LiteProducer extends AbstractLiteClient {
Preconditions.checkState(StringUtils.isNotBlank(message.getContent()),
"eventMeshMessage[content] invalid");
RequestParam requestParam = new RequestParam(HttpMethod.POST);
- requestParam.addHeader(ProtocolKey.REQUEST_CODE,
- String.valueOf(RequestCode.MSG_SEND_SYNC.getRequestCode()))
+ requestParam
+ .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_SYNC.getRequestCode())
.addHeader(ProtocolKey.ClientInstanceKey.ENV, liteClientConfig.getEnv())
.addHeader(ProtocolKey.ClientInstanceKey.IDC, liteClientConfig.getIdc())
.addHeader(ProtocolKey.ClientInstanceKey.IP, liteClientConfig.getIp())
@@ -172,7 +172,7 @@ public class LiteProducer extends AbstractLiteClient {
.addBody(SendMessageRequestBody.BIZSEQNO, message.getBizSeqNo())
.addBody(SendMessageRequestBody.UNIQUEID, message.getUniqueId());
- long startTime = System.currentTimeMillis();
+ long startTime = System.nanoTime();
String target = selectEventMesh();
String res = "";
@@ -180,10 +180,10 @@ public class LiteProducer extends AbstractLiteClient {
res = HttpUtil.post(httpClient, target, requestParam);
}
- if (logger.isDebugEnabled()) {
- logger.debug(
+ if (log.isDebugEnabled()) {
+ log.debug(
"publish sync message by await, targetEventMesh:{}, cost:{}ms, message:{}, rtn:{}",
- target, System.currentTimeMillis() - startTime, message, res);
+ target, (System.nanoTime() - startTime) / 1000000, message, res);
}
EventMeshRetObj ret = JsonUtils.deserialize(res, EventMeshRetObj.class);
@@ -195,7 +195,7 @@ public class LiteProducer extends AbstractLiteClient {
.setTopic(replyMessage.topic);
return eventMeshMessage;
}
-
+ // todo: return msg
return null;
}
@@ -210,8 +210,8 @@ public class LiteProducer extends AbstractLiteClient {
Preconditions.checkState(ObjectUtils.allNotNull(rrCallback),
"rrCallback invalid");
RequestParam requestParam = new RequestParam(HttpMethod.POST);
- requestParam.addHeader(ProtocolKey.REQUEST_CODE,
- String.valueOf(RequestCode.MSG_SEND_SYNC.getRequestCode()))
+ requestParam
+ .addHeader(ProtocolKey.REQUEST_CODE, RequestCode.MSG_SEND_SYNC.getRequestCode())
.addHeader(ProtocolKey.ClientInstanceKey.ENV, liteClientConfig.getEnv())
.addHeader(ProtocolKey.ClientInstanceKey.IDC, liteClientConfig.getIdc())
.addHeader(ProtocolKey.ClientInstanceKey.IP, liteClientConfig.getIp())
@@ -229,7 +229,7 @@ public class LiteProducer extends AbstractLiteClient {
.addBody(SendMessageRequestBody.BIZSEQNO, message.getBizSeqNo())
.addBody(SendMessageRequestBody.UNIQUEID, message.getUniqueId());
- long startTime = System.currentTimeMillis();
+ long startTime = System.nanoTime();
String target = selectEventMesh();
try (CloseableHttpClient httpClient = setHttpClient()) {
@@ -237,9 +237,9 @@ public class LiteProducer extends AbstractLiteClient {
new RRCallbackResponseHandlerAdapter(message, rrCallback, timeout));
}
- if (logger.isDebugEnabled()) {
- logger.debug("publish sync message by async, target:{}, cost:{}, message:{}", target,
- System.currentTimeMillis() - startTime, message);
+ if (log.isDebugEnabled()) {
+ log.debug("publish sync message by async, target:{}, cost:{}, message:{}", target,
+ (System.nanoTime() - startTime) / 1000000, message);
}
}
}
diff --git a/style/checkStyle.xml b/style/checkStyle.xml
index 9e86d55..acea7a7 100644
--- a/style/checkStyle.xml
+++ b/style/checkStyle.xml
@@ -54,7 +54,7 @@
<module name="LineLength">
<property name="fileExtensions" value="java"/>
- <property name="max" value="100"/>
+ <property name="max" value="120"/>
<property name="ignorePattern" value="^package.*|^import.*|a href|href|http://|https://|ftp://"/>
</module>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org