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 24873dd02 [ISSUE #4631]Optimize the message body of the Java SDK's
returned reply message (#4632)
24873dd02 is described below
commit 24873dd0227dbcddc9610dcea1aa4a62fc33be0e
Author: mxsm <[email protected]>
AuthorDate: Sun Dec 10 12:57:41 2023 +0800
[ISSUE #4631]Optimize the message body of the Java SDK's returned reply
message (#4632)
* [ISSUE #4631]Optimize the message body of the SDK's returned reply message
* optimize code, remove useless code
---
.../common/protocol/grpc/common/ProtocolKey.java | 1 +
.../protocol/grpc/common/SubscriptionReply.java | 59 ----------------------
.../protocol/grpc/service/ConsumerService.java | 3 +-
.../client/grpc/consumer/SubStreamHandler.java | 27 ++--------
4 files changed, 7 insertions(+), 83 deletions(-)
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/ProtocolKey.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/ProtocolKey.java
index 0aede5644..ec47fda44 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/ProtocolKey.java
+++
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/ProtocolKey.java
@@ -48,6 +48,7 @@ public class ProtocolKey {
public static final String GRPC_RESPONSE_TIME = "time";
public static final String SUB_MESSAGE_TYPE = "submessagetype";
+ public static final String SUB_REPLY_MESSAGE = "subscription_reply";
/**
* CloudEvents spec
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/SubscriptionReply.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/SubscriptionReply.java
deleted file mode 100644
index da40a0363..000000000
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/SubscriptionReply.java
+++ /dev/null
@@ -1,59 +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.common.protocol.grpc.common;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-@Data
-@Builder
-@AllArgsConstructor
-@NoArgsConstructor
-public class SubscriptionReply {
-
- public static final String TYPE = "subscription_reply";
-
- private String producerGroup;
-
- private String topic;
-
- private String content;
-
- private String ttl;
-
- private String uniqueId;
-
- private String seqNum;
-
- private String tag;
-
- @Builder.Default
- private Map<String, String> properties = new HashMap<>();
-
- public void putAllProperties(Map<String, String> properties) {
- if (null == properties || properties.isEmpty()) {
- return;
- }
- properties.putAll(properties);
- }
-}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/ConsumerService.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/ConsumerService.java
index 8852c1b40..8963cb95b 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/ConsumerService.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/ConsumerService.java
@@ -23,7 +23,6 @@ import
org.apache.eventmesh.common.protocol.grpc.cloudevents.ConsumerServiceGrpc
import
org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils;
import org.apache.eventmesh.common.protocol.grpc.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.grpc.common.StatusCode;
-import org.apache.eventmesh.common.protocol.grpc.common.SubscriptionReply;
import org.apache.eventmesh.runtime.boot.EventMeshGrpcServer;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
import
org.apache.eventmesh.runtime.core.protocol.grpc.processor.ReplyMessageProcessor;
@@ -87,7 +86,7 @@ public class ConsumerService extends
ConsumerServiceGrpc.ConsumerServiceImplBase
public void onNext(CloudEvent subscription) {
final String subMessageType =
Optional.ofNullable(subscription.getAttributesMap().get(ProtocolKey.SUB_MESSAGE_TYPE))
.orElse(CloudEventAttributeValue.newBuilder().build()).getCeString();
- if (StringUtils.equals(subMessageType,
SubscriptionReply.TYPE)) {
+ if (StringUtils.equals(subMessageType,
ProtocolKey.SUB_REPLY_MESSAGE)) {
log.info("cmd={}|{}|client2eventMesh|from={}|to={}",
"reply-to-server", EventMeshConstants.PROTOCOL_GRPC,
EventMeshCloudEventUtils.getIp(subscription),
eventMeshGrpcServer.getEventMeshGrpcConfiguration().getEventMeshIp());
handleSubscribeReply(subscription, emitter);
diff --git
a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java
b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java
index 5f1ee67ca..0d4e7fd94 100644
---
a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java
+++
b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/SubStreamHandler.java
@@ -25,13 +25,9 @@ import
org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent.CloudEve
import
org.apache.eventmesh.common.protocol.grpc.cloudevents.ConsumerServiceGrpc.ConsumerServiceStub;
import
org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils;
import org.apache.eventmesh.common.protocol.grpc.common.ProtocolKey;
-import org.apache.eventmesh.common.protocol.grpc.common.SubscriptionReply;
-import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.common.utils.LogUtils;
import java.io.Serializable;
-import java.util.HashMap;
-import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
@@ -118,26 +114,13 @@ public class SubStreamHandler<T> extends Thread
implements Serializable {
private CloudEvent buildReplyMessage(final CloudEvent reqMessage, final T
replyMessage) {
final CloudEvent cloudEvent =
EventMeshCloudEventBuilder.buildEventMeshCloudEvent(replyMessage,
clientConfig, listener.getProtocolType());
- SubscriptionReply subscriptionReply =
SubscriptionReply.builder().producerGroup(clientConfig.getConsumerGroup())
- .topic(EventMeshCloudEventUtils.getSubject(cloudEvent))
- .content(EventMeshCloudEventUtils.getDataContent(cloudEvent))
- .seqNum(EventMeshCloudEventUtils.getSeqNum(cloudEvent))
- .uniqueId(EventMeshCloudEventUtils.getUniqueId(cloudEvent))
- .ttl(EventMeshCloudEventUtils.getTtl(cloudEvent)).build();
-
- Map<String, String> prop = new HashMap<>();
- Map<String, CloudEventAttributeValue> reqMessageMap =
reqMessage.getAttributesMap();
- reqMessageMap.entrySet().forEach(entry -> prop.put(entry.getKey(),
entry.getValue().getCeString()));
- Map<String, CloudEventAttributeValue> cloudEventMap =
cloudEvent.getAttributesMap();
- cloudEventMap.entrySet().forEach(entry -> prop.put(entry.getKey(),
entry.getValue().getCeString()));
- subscriptionReply.putAllProperties(prop);
-
- return
CloudEvent.newBuilder(cloudEvent).putAllAttributes(reqMessageMap)
+
+ return
CloudEvent.newBuilder(cloudEvent).putAllAttributes(reqMessage.getAttributesMap()).putAllAttributes(cloudEvent.getAttributesMap())
.putAttributes(ProtocolKey.DATA_CONTENT_TYPE,
CloudEventAttributeValue.newBuilder().setCeString(EventMeshDataContentType.JSON.getCode()).build())
- //Indicate that it is a subscription response
- .putAttributes(ProtocolKey.SUB_MESSAGE_TYPE,
CloudEventAttributeValue.newBuilder().setCeString(SubscriptionReply.TYPE).build())
- .setTextData(JsonUtils.toJSONString(subscriptionReply)).build();
+ // Indicate that it is a subscription response
+ .putAttributes(ProtocolKey.SUB_MESSAGE_TYPE,
CloudEventAttributeValue.newBuilder().setCeString(ProtocolKey.SUB_REPLY_MESSAGE).build())
+ .build();
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
