This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 7f3fc8ab4 [ISSUE #3473]The code processing logic is the same in both
places. (#3792)
7f3fc8ab4 is described below
commit 7f3fc8ab42c7a1d0774f50f8f9b04f3523f7b96a
Author: pandaapo <[email protected]>
AuthorDate: Thu May 4 00:19:25 2023 +0800
[ISSUE #3473]The code processing logic is the same in both places. (#3792)
* Reuse code logic and api provided.
* fix checkstyle error.
---
.../http/SendMessageBatchV2ProtocolResolver.java | 25 ++-----------
.../http/SendMessageRequestProtocolResolver.java | 27 +++-----------
.../http/SendMessageBatchV2ProtocolResolver.java | 17 +++------
.../http/SendMessageRequestProtocolResolver.java | 42 +++-------------------
4 files changed, 15 insertions(+), 96 deletions(-)
diff --git
a/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/resolver/http/SendMessageBatchV2ProtocolResolver.java
b/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/resolver/http/SendMessageBatchV2ProtocolResolver.java
index c436ac15c..fd426859f 100644
---
a/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/resolver/http/SendMessageBatchV2ProtocolResolver.java
+++
b/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/resolver/http/SendMessageBatchV2ProtocolResolver.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.protocol.cloudevents.resolver.http;
+import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.protocol.http.body.Body;
import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageBatchV2RequestBody;
import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
@@ -27,7 +28,6 @@ import
org.apache.eventmesh.protocol.api.exception.ProtocolHandleException;
import org.apache.commons.lang3.StringUtils;
-import java.nio.charset.StandardCharsets;
import java.util.Objects;
import io.cloudevents.CloudEvent;
@@ -62,28 +62,9 @@ public class SendMessageBatchV2ProtocolResolver {
String content = sendMessageBatchV2RequestBody.getMsg();
CloudEvent event = null;
- if (StringUtils.equals(SpecVersion.V1.toString(),
protocolVersion)) {
+ if (StringUtils.equalsAny(protocolVersion,
SpecVersion.V1.toString(), SpecVersion.V03.toString())) {
event =
Objects.requireNonNull(EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE))
- .deserialize(content.getBytes(StandardCharsets.UTF_8));
- event = CloudEventBuilder.from(event)
- .withExtension(ProtocolKey.REQUEST_CODE, code)
- .withExtension(ProtocolKey.ClientInstanceKey.ENV, env)
- .withExtension(ProtocolKey.ClientInstanceKey.IDC, idc)
- .withExtension(ProtocolKey.ClientInstanceKey.IP, ip)
- .withExtension(ProtocolKey.ClientInstanceKey.PID, pid)
- .withExtension(ProtocolKey.ClientInstanceKey.SYS, sys)
- .withExtension(ProtocolKey.ClientInstanceKey.USERNAME,
username)
- .withExtension(ProtocolKey.ClientInstanceKey.PASSWD,
passwd)
- .withExtension(ProtocolKey.VERSION, version.getVersion())
- .withExtension(ProtocolKey.LANGUAGE, language)
- .withExtension(ProtocolKey.PROTOCOL_TYPE, protocolType)
- .withExtension(ProtocolKey.PROTOCOL_DESC, protocolDesc)
- .withExtension(ProtocolKey.PROTOCOL_VERSION,
protocolVersion)
-
.withExtension(SendMessageBatchV2RequestBody.PRODUCERGROUP, producerGroup)
- .build();
- } else if (StringUtils.equals(SpecVersion.V03.toString(),
protocolVersion)) {
- event =
Objects.requireNonNull(EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE))
- .deserialize(content.getBytes(StandardCharsets.UTF_8));
+ .deserialize(content.getBytes(Constants.DEFAULT_CHARSET));
event = CloudEventBuilder.from(event)
.withExtension(ProtocolKey.REQUEST_CODE, code)
.withExtension(ProtocolKey.ClientInstanceKey.ENV, env)
diff --git
a/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/resolver/http/SendMessageRequestProtocolResolver.java
b/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/resolver/http/SendMessageRequestProtocolResolver.java
index 9f14a0a05..328fbf127 100644
---
a/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/resolver/http/SendMessageRequestProtocolResolver.java
+++
b/eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/src/main/java/org/apache/eventmesh/protocol/cloudevents/resolver/http/SendMessageRequestProtocolResolver.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.protocol.cloudevents.resolver.http;
+import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.protocol.http.body.Body;
import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageRequestBody;
import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
@@ -27,7 +28,6 @@ import
org.apache.eventmesh.protocol.api.exception.ProtocolHandleException;
import org.apache.commons.lang3.StringUtils;
-import java.nio.charset.StandardCharsets;
import java.util.Objects;
import io.cloudevents.CloudEvent;
@@ -62,29 +62,10 @@ public class SendMessageRequestProtocolResolver {
String content = sendMessageRequestBody.getContent();
CloudEvent event = null;
- if (StringUtils.equals(SpecVersion.V1.toString(),
protocolVersion)) {
+ if (StringUtils.equalsAny(protocolVersion,
SpecVersion.V1.toString(), SpecVersion.V03.toString())) {
event =
Objects.requireNonNull(EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE))
- .deserialize(content.getBytes(StandardCharsets.UTF_8));
- event = CloudEventBuilder.v1(event)
- .withExtension(ProtocolKey.REQUEST_CODE, code)
- .withExtension(ProtocolKey.ClientInstanceKey.ENV, env)
- .withExtension(ProtocolKey.ClientInstanceKey.IDC, idc)
- .withExtension(ProtocolKey.ClientInstanceKey.IP, ip)
- .withExtension(ProtocolKey.ClientInstanceKey.PID, pid)
- .withExtension(ProtocolKey.ClientInstanceKey.SYS, sys)
- .withExtension(ProtocolKey.ClientInstanceKey.USERNAME,
username)
- .withExtension(ProtocolKey.ClientInstanceKey.PASSWD,
passwd)
- .withExtension(ProtocolKey.VERSION, version.getVersion())
- .withExtension(ProtocolKey.LANGUAGE, language)
- .withExtension(ProtocolKey.PROTOCOL_TYPE, protocolType)
- .withExtension(ProtocolKey.PROTOCOL_DESC, protocolDesc)
- .withExtension(ProtocolKey.PROTOCOL_VERSION,
protocolVersion)
- .withExtension(SendMessageRequestBody.PRODUCERGROUP,
producerGroup)
- .build();
- } else if (StringUtils.equals(SpecVersion.V03.toString(),
protocolVersion)) {
- event =
Objects.requireNonNull(EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE))
- .deserialize(content.getBytes(StandardCharsets.UTF_8));
- event = CloudEventBuilder.v03(event)
+ .deserialize(content.getBytes(Constants.DEFAULT_CHARSET));
+ event = CloudEventBuilder.from(event)
.withExtension(ProtocolKey.REQUEST_CODE, code)
.withExtension(ProtocolKey.ClientInstanceKey.ENV, env)
.withExtension(ProtocolKey.ClientInstanceKey.IDC, idc)
diff --git
a/eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/java/org/apache/eventmesh/protocol/meshmessage/resolver/http/SendMessageBatchV2ProtocolResolver.java
b/eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/java/org/apache/eventmesh/protocol/meshmessage/resolver/http/SendMessageBatchV2ProtocolResolver.java
index 656b7f7c3..43259805b 100644
---
a/eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/java/org/apache/eventmesh/protocol/meshmessage/resolver/http/SendMessageBatchV2ProtocolResolver.java
+++
b/eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/java/org/apache/eventmesh/protocol/meshmessage/resolver/http/SendMessageBatchV2ProtocolResolver.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.protocol.meshmessage.resolver.http;
+import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.protocol.http.body.Body;
import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageBatchV2RequestBody;
import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageRequestBody;
@@ -28,7 +29,6 @@ import
org.apache.eventmesh.protocol.api.exception.ProtocolHandleException;
import org.apache.commons.lang3.StringUtils;
import java.net.URI;
-import java.nio.charset.StandardCharsets;
import io.cloudevents.CloudEvent;
import io.cloudevents.SpecVersion;
@@ -41,17 +41,8 @@ public class SendMessageBatchV2ProtocolResolver {
SendMessageBatchV2RequestHeader sendMessageBatchV2RequestHeader =
(SendMessageBatchV2RequestHeader) header;
SendMessageBatchV2RequestBody sendMessageBatchV2RequestBody =
(SendMessageBatchV2RequestBody) body;
- CloudEventBuilder cloudEventBuilder;
- switch
(SpecVersion.parse(sendMessageBatchV2RequestHeader.getProtocolVersion())) {
- case V1:
- cloudEventBuilder = CloudEventBuilder.v1();
- break;
- case V03:
- cloudEventBuilder = CloudEventBuilder.v03();
- break;
- default:
- return null; // unsupported protocol version
- }
+ CloudEventBuilder cloudEventBuilder =
CloudEventBuilder.fromSpecVersion(
+
SpecVersion.parse(sendMessageBatchV2RequestHeader.getProtocolVersion()));
return getBuildCloudEvent(sendMessageBatchV2RequestHeader,
sendMessageBatchV2RequestBody, cloudEventBuilder);
} catch (Exception e) {
@@ -65,7 +56,7 @@ public class SendMessageBatchV2ProtocolResolver {
.withSubject(sendMessageBatchV2RequestBody.getTopic())
.withType("eventmeshmessage")
.withSource(URI.create("/"))
-
.withData(sendMessageBatchV2RequestBody.getMsg().getBytes(StandardCharsets.UTF_8))
+
.withData(sendMessageBatchV2RequestBody.getMsg().getBytes(Constants.DEFAULT_CHARSET))
.withExtension(ProtocolKey.REQUEST_CODE,
sendMessageBatchV2RequestHeader.getCode())
.withExtension(ProtocolKey.ClientInstanceKey.ENV,
sendMessageBatchV2RequestHeader.getEnv())
.withExtension(ProtocolKey.ClientInstanceKey.IDC,
sendMessageBatchV2RequestHeader.getIdc())
diff --git
a/eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/java/org/apache/eventmesh/protocol/meshmessage/resolver/http/SendMessageRequestProtocolResolver.java
b/eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/java/org/apache/eventmesh/protocol/meshmessage/resolver/http/SendMessageRequestProtocolResolver.java
index 215f2e38c..e41da4ad6 100644
---
a/eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/java/org/apache/eventmesh/protocol/meshmessage/resolver/http/SendMessageRequestProtocolResolver.java
+++
b/eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/java/org/apache/eventmesh/protocol/meshmessage/resolver/http/SendMessageRequestProtocolResolver.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.protocol.meshmessage.resolver.http;
+import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.protocol.http.body.Body;
import
org.apache.eventmesh.common.protocol.http.body.message.SendMessageRequestBody;
import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
@@ -28,7 +29,6 @@ import
org.apache.eventmesh.protocol.api.exception.ProtocolHandleException;
import org.apache.commons.lang3.StringUtils;
import java.net.URI;
-import java.nio.charset.StandardCharsets;
import java.util.Map;
import io.cloudevents.CloudEvent;
@@ -61,48 +61,14 @@ public class SendMessageRequestProtocolResolver {
CloudEvent event = null;
CloudEventBuilder cloudEventBuilder;
- if (StringUtils.equals(SpecVersion.V1.toString(),
protocolVersion)) {
- cloudEventBuilder = CloudEventBuilder.v1();
+ if (StringUtils.equalsAny(protocolVersion,
SpecVersion.V1.toString(), SpecVersion.V03.toString())) {
+ cloudEventBuilder =
CloudEventBuilder.fromSpecVersion(SpecVersion.parse(protocolVersion));
cloudEventBuilder =
cloudEventBuilder.withId(sendMessageRequestBody.getBizSeqNo())
.withSubject(sendMessageRequestBody.getTopic())
.withType("eventmeshmessage")
.withSource(URI.create("/"))
- .withData(content.getBytes(StandardCharsets.UTF_8))
- .withExtension(ProtocolKey.REQUEST_CODE, code)
- .withExtension(ProtocolKey.ClientInstanceKey.ENV, env)
- .withExtension(ProtocolKey.ClientInstanceKey.IDC, idc)
- .withExtension(ProtocolKey.ClientInstanceKey.IP, ip)
- .withExtension(ProtocolKey.ClientInstanceKey.PID, pid)
- .withExtension(ProtocolKey.ClientInstanceKey.SYS, sys)
- .withExtension(ProtocolKey.ClientInstanceKey.USERNAME,
username)
- .withExtension(ProtocolKey.ClientInstanceKey.PASSWD,
passwd)
- .withExtension(ProtocolKey.VERSION, version.getVersion())
- .withExtension(ProtocolKey.LANGUAGE, language)
- .withExtension(ProtocolKey.PROTOCOL_TYPE, protocolType)
- .withExtension(ProtocolKey.PROTOCOL_DESC, protocolDesc)
- .withExtension(ProtocolKey.PROTOCOL_VERSION,
protocolVersion)
- .withExtension(SendMessageRequestBody.BIZSEQNO,
sendMessageRequestBody.getBizSeqNo())
- .withExtension(SendMessageRequestBody.UNIQUEID,
sendMessageRequestBody.getUniqueId())
- .withExtension(SendMessageRequestBody.PRODUCERGROUP,
- sendMessageRequestBody.getProducerGroup())
- .withExtension(SendMessageRequestBody.TTL,
sendMessageRequestBody.getTtl());
- if (StringUtils.isNotEmpty(sendMessageRequestBody.getTag())) {
- cloudEventBuilder =
cloudEventBuilder.withExtension(SendMessageRequestBody.TAG,
sendMessageRequestBody.getTag());
- }
- if (sendMessageRequestBody.getExtFields() != null &&
sendMessageRequestBody.getExtFields().size() > 0) {
- for (Map.Entry<String, String> entry :
sendMessageRequestBody.getExtFields().entrySet()) {
- cloudEventBuilder =
cloudEventBuilder.withExtension(entry.getKey(), entry.getValue());
- }
- }
- event = cloudEventBuilder.build();
- } else if (StringUtils.equals(SpecVersion.V03.toString(),
protocolVersion)) {
- cloudEventBuilder = CloudEventBuilder.v03();
- cloudEventBuilder =
cloudEventBuilder.withId(sendMessageRequestBody.getBizSeqNo())
- .withSubject(sendMessageRequestBody.getTopic())
- .withType("eventmeshmessage")
- .withSource(URI.create("/"))
- .withData(content.getBytes(StandardCharsets.UTF_8))
+ .withData(content.getBytes(Constants.DEFAULT_CHARSET))
.withExtension(ProtocolKey.REQUEST_CODE, code)
.withExtension(ProtocolKey.ClientInstanceKey.ENV, env)
.withExtension(ProtocolKey.ClientInstanceKey.IDC, idc)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
