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 525a8dd74 [ISSUE #4164]Fix EventMeshGrpcConsumer client subscribe
topic error (#4165)
525a8dd74 is described below
commit 525a8dd74b9a76908afb7ceaa86f78c59d471d08
Author: mxsm <[email protected]>
AuthorDate: Fri Jun 30 16:35:44 2023 +0800
[ISSUE #4164]Fix EventMeshGrpcConsumer client subscribe topic error (#4165)
* [ISSUE #4164]Fix EventMeshGrpcConsumer client subscribe topic error
* fix code style
* fix code style
* fix code style
---
.../grpc/sub/app/controller/SubController.java | 25 --------
.../AbstractPublishBatchCloudEventProcessor.java | 6 +-
.../AbstractPublishCloudEventProcessor.java | 6 +-
.../processor/BatchPublishCloudEventProcessor.java | 2 +-
.../grpc/processor/HeartbeatProcessor.java | 12 ++--
.../processor/PublishCloudEventsProcessor.java | 4 +-
.../grpc/processor/ReplyMessageProcessor.java | 10 +--
.../grpc/processor/RequestCloudEventProcessor.java | 6 +-
.../grpc/processor/SubscribeProcessor.java | 8 +--
.../grpc/processor/SubscribeStreamProcessor.java | 8 +--
.../grpc/processor/UnsubscribeProcessor.java | 6 +-
.../protocol/grpc/service/ConsumerService.java | 14 ++--
.../protocol/grpc/service/HeartbeatService.java | 2 +-
.../protocol/grpc/service/PublisherService.java | 6 +-
.../core/protocol/grpc/service/ServiceUtils.java | 74 +++++++++++++++++++---
.../client/grpc/consumer/SubStreamHandler.java | 3 +-
.../grpc/util/EventMeshCloudEventBuilder.java | 3 +-
.../grpc/util/EventMeshCloudEventBuilderTest.java | 15 +++++
18 files changed, 129 insertions(+), 81 deletions(-)
diff --git
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/app/controller/SubController.java
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/app/controller/SubController.java
index aa95ac448..af14e09af 100644
---
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/app/controller/SubController.java
+++
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/sub/app/controller/SubController.java
@@ -17,14 +17,10 @@
package org.apache.eventmesh.grpc.sub.app.controller;
-import org.apache.eventmesh.client.tcp.common.EventMeshCommon;
import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.grpc.sub.app.service.SubService;
-import org.apache.commons.lang3.StringUtils;
-
-import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
@@ -35,11 +31,6 @@ import
org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
-import io.cloudevents.CloudEvent;
-import io.cloudevents.CloudEventData;
-import io.cloudevents.core.format.EventFormat;
-import io.cloudevents.core.provider.EventFormatProvider;
-
import lombok.extern.slf4j.Slf4j;
@Slf4j
@@ -59,22 +50,6 @@ public class SubController {
log.info("=======receive message======= {}", content);
}
- if (StringUtils.equals(EventMeshCommon.CLOUD_EVENTS_PROTOCOL_NAME,
protocolType)) {
- final String contentType =
request.getHeader(ProtocolKey.CONTENT_TYPE);
-
- final EventFormat eventFormat =
EventFormatProvider.getInstance().resolveFormat(contentType);
- if (eventFormat != null) {
- final CloudEvent event =
eventFormat.deserialize(content.getBytes(StandardCharsets.UTF_8));
- final CloudEventData cloudEventData = event.getData();
- if (cloudEventData != null) {
- final String data = new String(cloudEventData.toBytes(),
StandardCharsets.UTF_8);
- if (log.isInfoEnabled()) {
- log.info("=======receive data======= {}", data);
- }
- }
- }
- }
-
subService.consumeMessage(content);
final Map<String, Object> map = new HashMap<>();
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/AbstractPublishBatchCloudEventProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/AbstractPublishBatchCloudEventProcessor.java
index d98f9d806..43cdb9a3c 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/AbstractPublishBatchCloudEventProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/AbstractPublishBatchCloudEventProcessor.java
@@ -56,18 +56,18 @@ public abstract class
AbstractPublishBatchCloudEventProcessor implements Publish
// control flow rate limit
if
(!eventMeshGrpcServer.getMsgRateLimiter().tryAcquire(EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS,
TimeUnit.MILLISECONDS)) {
log.error("Send message speed over limit.");
-
ServiceUtils.completed(StatusCode.EVENTMESH_SEND_MESSAGE_SPEED_OVER_LIMIT_ERR,
emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_SEND_MESSAGE_SPEED_OVER_LIMIT_ERR,
emitter);
return;
}
StatusCode cloudEventCheck = cloudEventBatchCheck(cloudEvent);
if (cloudEventCheck != StatusCode.SUCCESS) {
- ServiceUtils.completed(cloudEventCheck, emitter);
+ ServiceUtils.sendResponseCompleted(cloudEventCheck, emitter);
return;
}
StatusCode aclCheck = this.aclCheck(cloudEvent.getEvents(0));
if (aclCheck != StatusCode.SUCCESS) {
- ServiceUtils.completed(aclCheck, emitter);
+ ServiceUtils.sendResponseCompleted(aclCheck, emitter);
return;
}
handleCloudEvent(cloudEvent, emitter);
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/AbstractPublishCloudEventProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/AbstractPublishCloudEventProcessor.java
index 484ef1e93..2f46f2f06 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/AbstractPublishCloudEventProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/AbstractPublishCloudEventProcessor.java
@@ -55,18 +55,18 @@ public abstract class AbstractPublishCloudEventProcessor
implements PublishProce
// control flow rate limit
if
(!eventMeshGrpcServer.getMsgRateLimiter().tryAcquire(EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS,
TimeUnit.MILLISECONDS)) {
log.error("Send message speed over limit.");
- ServiceUtils.streamCompleted(cloudEvent,
StatusCode.EVENTMESH_SEND_MESSAGE_SPEED_OVER_LIMIT_ERR, emitter);
+ ServiceUtils.sendStreamResponseCompleted(cloudEvent,
StatusCode.EVENTMESH_SEND_MESSAGE_SPEED_OVER_LIMIT_ERR, emitter);
return;
}
StatusCode cloudEventCheck = cloudEventCheck(cloudEvent);
if (cloudEventCheck != StatusCode.SUCCESS) {
- ServiceUtils.completed(cloudEventCheck, emitter);
+ ServiceUtils.sendResponseCompleted(cloudEventCheck, emitter);
return;
}
StatusCode aclCheck = this.aclCheck(cloudEvent);
if (aclCheck != StatusCode.SUCCESS) {
- ServiceUtils.completed(aclCheck, emitter);
+ ServiceUtils.sendResponseCompleted(aclCheck, emitter);
return;
}
handleCloudEvent(cloudEvent, emitter);
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/BatchPublishCloudEventProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/BatchPublishCloudEventProcessor.java
index 60e794637..06f946aed 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/BatchPublishCloudEventProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/BatchPublishCloudEventProcessor.java
@@ -85,7 +85,7 @@ public class BatchPublishCloudEventProcessor extends
AbstractPublishBatchCloudEv
}
});
}
- ServiceUtils.completed(StatusCode.SUCCESS, "batch publish success",
emitter);
+ ServiceUtils.sendResponseCompleted(StatusCode.SUCCESS, "batch publish
success", emitter);
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/HeartbeatProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/HeartbeatProcessor.java
index 256b56389..805f16f35 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/HeartbeatProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/HeartbeatProcessor.java
@@ -56,12 +56,12 @@ public class HeartbeatProcessor {
public void process(CloudEvent heartbeat, EventEmitter<CloudEvent>
emitter) throws Exception {
if (!ServiceUtils.validateCloudEventAttributes(heartbeat)) {
- ServiceUtils.completed(StatusCode.EVENTMESH_PROTOCOL_HEADER_ERR,
emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_PROTOCOL_HEADER_ERR,
emitter);
return;
}
if (!ServiceUtils.validateHeartBeat(heartbeat)) {
- ServiceUtils.completed(StatusCode.EVENTMESH_PROTOCOL_BODY_ERR,
emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_PROTOCOL_BODY_ERR,
emitter);
return;
}
@@ -69,14 +69,14 @@ public class HeartbeatProcessor {
doAclCheck(heartbeat);
} catch (AclException e) {
aclLogger.warn("CLIENT HAS NO PERMISSION, HeartbeatProcessor
failed", e);
- ServiceUtils.completed(StatusCode.EVENTMESH_ACL_ERR,
e.getMessage(), emitter);
+ ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_ACL_ERR,
e.getMessage(), emitter);
return;
}
// only handle heartbeat for consumers
org.apache.eventmesh.common.protocol.grpc.common.ClientType clientType
= EventMeshCloudEventUtils.getClientType(heartbeat);
if (org.apache.eventmesh.common.protocol.grpc.common.ClientType.SUB !=
clientType) {
- ServiceUtils.completed(StatusCode.EVENTMESH_PROTOCOL_BODY_ERR,
emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_PROTOCOL_BODY_ERR,
emitter);
return;
}
@@ -106,11 +106,11 @@ public class HeartbeatProcessor {
// consumer group client is lost, and the client needs to
resubscribe.
if (!consumerManager.updateClientTime(hbClient)) {
- ServiceUtils.completed(StatusCode.CLIENT_RESUBSCRIBE, emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.CLIENT_RESUBSCRIBE, emitter);
return;
}
}
- ServiceUtils.completed(StatusCode.SUCCESS, "heartbeat success",
emitter);
+ ServiceUtils.sendResponseCompleted(StatusCode.SUCCESS, "heartbeat
success", emitter);
}
private void doAclCheck(CloudEvent heartbeat) throws AclException {
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/PublishCloudEventsProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/PublishCloudEventsProcessor.java
index bce303b26..048531587 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/PublishCloudEventsProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/PublishCloudEventsProcessor.java
@@ -66,7 +66,7 @@ public class PublishCloudEventsProcessor extends
AbstractPublishCloudEventProces
eventMeshProducer.send(sendMessageContext, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
- ServiceUtils.completed(StatusCode.SUCCESS,
sendResult.toString(), emitter);
+ ServiceUtils.sendResponseCompleted(StatusCode.SUCCESS,
sendResult.toString(), emitter);
long endTime = System.currentTimeMillis();
log.info("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
endTime - startTime, topic, seqNum, uniqueId);
@@ -75,7 +75,7 @@ public class PublishCloudEventsProcessor extends
AbstractPublishCloudEventProces
@Override
public void onException(OnExceptionContext context) {
- ServiceUtils.completed(StatusCode.EVENTMESH_SEND_ASYNC_MSG_ERR,
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_SEND_ASYNC_MSG_ERR,
EventMeshUtil.stackTrace(context.getException(), 2),
emitter);
long endTime = System.currentTimeMillis();
log.error("message|eventMesh2mq|REQ|ASYNC|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/ReplyMessageProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/ReplyMessageProcessor.java
index 4b5e20f90..d82ae77a6 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/ReplyMessageProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/ReplyMessageProcessor.java
@@ -66,12 +66,12 @@ public class ReplyMessageProcessor {
public void process(CloudEvent message, EventEmitter<CloudEvent> emitter)
throws Exception {
if (!ServiceUtils.validateCloudEventAttributes(message)) {
- ServiceUtils.streamCompleted(message,
StatusCode.EVENTMESH_PROTOCOL_HEADER_ERR, emitter);
+ ServiceUtils.sendStreamResponseCompleted(message,
StatusCode.EVENTMESH_PROTOCOL_HEADER_ERR, emitter);
return;
}
if (!ServiceUtils.validateCloudEventData(message)) {
- ServiceUtils.streamCompleted(message,
StatusCode.EVENTMESH_PROTOCOL_BODY_ERR, emitter);
+ ServiceUtils.sendStreamResponseCompleted(message,
StatusCode.EVENTMESH_PROTOCOL_BODY_ERR, emitter);
return;
}
@@ -79,7 +79,7 @@ public class ReplyMessageProcessor {
doAclCheck(message);
} catch (Exception e) {
aclLogger.warn("CLIENT HAS NO
PERMISSION,RequestReplyMessageProcessor reply failed", e);
- ServiceUtils.streamCompleted(message,
StatusCode.EVENTMESH_ACL_ERR, e.getMessage(), emitter);
+ ServiceUtils.sendStreamResponseCompleted(message,
StatusCode.EVENTMESH_ACL_ERR, e.getMessage(), emitter);
return;
}
@@ -87,7 +87,7 @@ public class ReplyMessageProcessor {
if (!eventMeshGrpcServer.getMsgRateLimiter()
.tryAcquire(EventMeshConstants.DEFAULT_FASTFAIL_TIMEOUT_IN_MILLISECONDS,
TimeUnit.MILLISECONDS)) {
log.error("Send message speed over limit.");
- ServiceUtils.streamCompleted(message,
StatusCode.EVENTMESH_SEND_MESSAGE_SPEED_OVER_LIMIT_ERR, emitter);
+ ServiceUtils.sendStreamResponseCompleted(message,
StatusCode.EVENTMESH_SEND_MESSAGE_SPEED_OVER_LIMIT_ERR, emitter);
return;
}
@@ -122,7 +122,7 @@ public class ReplyMessageProcessor {
@Override
public void onException(OnExceptionContext onExceptionContext) {
- ServiceUtils.streamCompleted(messageReply,
StatusCode.EVENTMESH_REPLY_MSG_ERR,
+ ServiceUtils.sendStreamResponseCompleted(messageReply,
StatusCode.EVENTMESH_REPLY_MSG_ERR,
EventMeshUtil.stackTrace(onExceptionContext.getException(), 2), emitter);
long endTime = System.currentTimeMillis();
log.error("message|mq2eventmesh|REPLY|ReplyToServer|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/RequestCloudEventProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/RequestCloudEventProcessor.java
index 7cbd18c58..a256f5bb7 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/RequestCloudEventProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/RequestCloudEventProcessor.java
@@ -79,7 +79,8 @@ public class RequestCloudEventProcessor extends
AbstractPublishCloudEventProcess
endTime - startTime, topic, seqNum, uniqueId);
eventMeshGrpcServer.getMetricsMonitor().recordSendMsgToClient();
} catch (Exception e) {
- ServiceUtils.streamCompleted(message,
StatusCode.EVENTMESH_REQUEST_REPLY_MSG_ERR, EventMeshUtil.stackTrace(e, 2),
emitter);
+ ServiceUtils.sendStreamResponseCompleted(message,
StatusCode.EVENTMESH_REQUEST_REPLY_MSG_ERR, EventMeshUtil.stackTrace(e, 2),
+ emitter);
long endTime = System.currentTimeMillis();
log.error("message|mq2eventmesh|REPLY|RequestReply|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
endTime - startTime, topic, seqNum, uniqueId, e);
@@ -88,7 +89,8 @@ public class RequestCloudEventProcessor extends
AbstractPublishCloudEventProcess
@Override
public void onException(Throwable e) {
- ServiceUtils.streamCompleted(message,
StatusCode.EVENTMESH_REQUEST_REPLY_MSG_ERR, EventMeshUtil.stackTrace(e, 2),
emitter);
+ ServiceUtils.sendStreamResponseCompleted(message,
StatusCode.EVENTMESH_REQUEST_REPLY_MSG_ERR, EventMeshUtil.stackTrace(e, 2),
+ emitter);
long endTime = System.currentTimeMillis();
log.error("message|eventMesh2mq|REPLY|RequestReply|send2MQCost={}ms|topic={}|bizSeqNo={}|uniqueId={}",
endTime - startTime, topic, seqNum, uniqueId, e);
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/SubscribeProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/SubscribeProcessor.java
index f8a276fce..5a74e01ba 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/SubscribeProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/SubscribeProcessor.java
@@ -62,12 +62,12 @@ public class SubscribeProcessor {
Objects.requireNonNull(emitter, "emitter can not be null");
if (!ServiceUtils.validateCloudEventAttributes(subscription)) {
- ServiceUtils.completed(StatusCode.EVENTMESH_PROTOCOL_HEADER_ERR,
emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_PROTOCOL_HEADER_ERR,
emitter);
return;
}
if (!ServiceUtils.validateSubscription(grpcType, subscription)) {
- ServiceUtils.completed(StatusCode.EVENTMESH_PROTOCOL_BODY_ERR,
emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_PROTOCOL_BODY_ERR,
emitter);
return;
}
@@ -77,7 +77,7 @@ public class SubscribeProcessor {
if (log.isWarnEnabled()) {
log.warn("CLIENT HAS NO PERMISSION to Subscribe. failed", e);
}
- ServiceUtils.completed(StatusCode.EVENTMESH_ACL_ERR,
e.getMessage(), emitter);
+ ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_ACL_ERR,
e.getMessage(), emitter);
return;
}
@@ -136,7 +136,7 @@ public class SubscribeProcessor {
log.warn("EventMesh consumer [{}] didn't restart.",
consumerGroup);
}
}
- ServiceUtils.completed(StatusCode.SUCCESS, "subscribe success",
emitter);
+ ServiceUtils.sendResponseCompleted(StatusCode.SUCCESS, "subscribe
success", emitter);
}
private void doAclCheck(final CloudEvent subscription) throws AclException
{
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/SubscribeStreamProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/SubscribeStreamProcessor.java
index 3c3d6bf21..3616b31f4 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/SubscribeStreamProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/SubscribeStreamProcessor.java
@@ -64,12 +64,12 @@ public class SubscribeStreamProcessor {
public void process(CloudEvent subscription, EventEmitter<CloudEvent>
emitter) throws Exception {
if (!ServiceUtils.validateCloudEventAttributes(subscription)) {
- ServiceUtils.completed(StatusCode.EVENTMESH_PROTOCOL_HEADER_ERR,
emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_PROTOCOL_HEADER_ERR,
emitter);
return;
}
if (!ServiceUtils.validateSubscription(grpcType, subscription)) {
- ServiceUtils.completed(StatusCode.EVENTMESH_PROTOCOL_BODY_ERR,
emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_PROTOCOL_BODY_ERR,
emitter);
return;
}
@@ -77,7 +77,7 @@ public class SubscribeStreamProcessor {
doAclCheck(subscription);
} catch (AclException e) {
aclLogger.warn("CLIENT HAS NO PERMISSION to Subscribe. failed", e);
- ServiceUtils.streamCompleted(subscription,
StatusCode.EVENTMESH_ACL_ERR, e.getMessage(), emitter);
+ ServiceUtils.sendStreamResponseCompleted(subscription,
StatusCode.EVENTMESH_ACL_ERR, e.getMessage(), emitter);
return;
}
@@ -135,7 +135,7 @@ public class SubscribeStreamProcessor {
log.warn("EventMesh consumer [{}] didn't restart.", consumerGroup);
}
- ServiceUtils.streamCompleted(subscription, StatusCode.SUCCESS,
"subscribe success", emitter);
+ ServiceUtils.sendStreamResponse(subscription, StatusCode.SUCCESS,
"subscribe success", emitter);
}
private void doAclCheck(CloudEvent subscription) throws AclException {
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/UnsubscribeProcessor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/UnsubscribeProcessor.java
index 2301ad629..708e3cfee 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/UnsubscribeProcessor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/processor/UnsubscribeProcessor.java
@@ -49,12 +49,12 @@ public class UnsubscribeProcessor {
public void process(CloudEvent subscription, EventEmitter<CloudEvent>
emitter) throws Exception {
if (!ServiceUtils.validateCloudEventAttributes(subscription)) {
- ServiceUtils.completed(StatusCode.EVENTMESH_PROTOCOL_HEADER_ERR,
emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_PROTOCOL_HEADER_ERR,
emitter);
return;
}
if (!ServiceUtils.validateSubscription(null, subscription)) {
- ServiceUtils.completed(StatusCode.EVENTMESH_PROTOCOL_BODY_ERR,
emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_PROTOCOL_BODY_ERR,
emitter);
return;
}
@@ -111,6 +111,6 @@ public class UnsubscribeProcessor {
log.warn("EventMesh consumer [{}] didn't restart.", consumerGroup);
}
- ServiceUtils.completed(StatusCode.SUCCESS, "unsubscribe success",
emitter);
+ ServiceUtils.sendResponseCompleted(StatusCode.SUCCESS, "unsubscribe
success", emitter);
}
}
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 f8801dfa8..e1f940738 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
@@ -32,7 +32,7 @@ import
org.apache.eventmesh.runtime.core.protocol.grpc.processor.UnsubscribeProc
import org.apache.commons.collections4.CollectionUtils;
-import java.util.List;
+import java.util.Set;
import java.util.concurrent.ThreadPoolExecutor;
import io.grpc.stub.StreamObserver;
@@ -73,7 +73,7 @@ public class ConsumerService extends
ConsumerServiceGrpc.ConsumerServiceImplBase
} catch (Exception e) {
log.error("Error code {}, error message {}",
StatusCode.EVENTMESH_SUBSCRIBE_ERR.getRetCode(),
StatusCode.EVENTMESH_SUBSCRIBE_ERR.getErrMsg(), e);
- ServiceUtils.completed(StatusCode.EVENTMESH_SUBSCRIBE_ERR,
e.getMessage(), emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_SUBSCRIBE_ERR,
e.getMessage(), emitter);
}
});
}
@@ -85,8 +85,8 @@ public class ConsumerService extends
ConsumerServiceGrpc.ConsumerServiceImplBase
return new StreamObserver<CloudEvent>() {
@Override
public void onNext(CloudEvent subscription) {
- List<SubscriptionItem> subscriptionItems =
JsonUtils.parseTypeReferenceObject(subscription.getTextData(),
- new TypeReference<List<SubscriptionItem>>() {
+ Set<SubscriptionItem> subscriptionItems =
JsonUtils.parseTypeReferenceObject(subscription.getTextData(),
+ new TypeReference<Set<SubscriptionItem>>() {
});
if (CollectionUtils.isNotEmpty(subscriptionItems)) {
log.info("cmd={}|{}|client2eventMesh|from={}|to={}",
"subscribeStream", EventMeshConstants.PROTOCOL_GRPC,
@@ -122,7 +122,7 @@ public class ConsumerService extends
ConsumerServiceGrpc.ConsumerServiceImplBase
streamProcessor.process(request, emitter);
} catch (Exception e) {
log.error("Error code {}, error message {}",
StatusCode.EVENTMESH_SUBSCRIBE_ERR, e.getMessage(), e);
- ServiceUtils.streamCompleted(request,
StatusCode.EVENTMESH_SUBSCRIBE_ERR, e.getMessage(), emitter);
+ ServiceUtils.sendStreamResponseCompleted(request,
StatusCode.EVENTMESH_SUBSCRIBE_ERR, e.getMessage(), emitter);
}
});
}
@@ -134,7 +134,7 @@ public class ConsumerService extends
ConsumerServiceGrpc.ConsumerServiceImplBase
replyMessageProcessor.process(subscription, emitter);
} catch (Exception e) {
log.error("Error code {}, error message {}",
StatusCode.EVENTMESH_SUBSCRIBE_ERR, e.getMessage(), e);
- ServiceUtils.streamCompleted(subscription,
StatusCode.EVENTMESH_SUBSCRIBE_ERR, e.getMessage(), emitter);
+ ServiceUtils.sendStreamResponseCompleted(subscription,
StatusCode.EVENTMESH_SUBSCRIBE_ERR, e.getMessage(), emitter);
}
});
}
@@ -153,7 +153,7 @@ public class ConsumerService extends
ConsumerServiceGrpc.ConsumerServiceImplBase
} catch (Exception e) {
log.error("Error code {}, error message {}",
StatusCode.EVENTMESH_UNSUBSCRIBE_ERR.getRetCode(),
StatusCode.EVENTMESH_UNSUBSCRIBE_ERR.getErrMsg(), e);
- ServiceUtils.completed(StatusCode.EVENTMESH_UNSUBSCRIBE_ERR,
e.getMessage(), emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_UNSUBSCRIBE_ERR,
e.getMessage(), emitter);
}
});
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/HeartbeatService.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/HeartbeatService.java
index 69a87eec4..0b85ac371 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/HeartbeatService.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/HeartbeatService.java
@@ -60,7 +60,7 @@ public class HeartbeatService extends
HeartbeatServiceGrpc.HeartbeatServiceImplB
} catch (Exception e) {
log.error("Error code {}, error message {}",
StatusCode.EVENTMESH_HEARTBEAT_ERR.getRetCode(),
StatusCode.EVENTMESH_HEARTBEAT_ERR.getErrMsg(), e);
- ServiceUtils.completed(StatusCode.EVENTMESH_HEARTBEAT_ERR,
e.getMessage(), emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_HEARTBEAT_ERR,
e.getMessage(), emitter);
}
});
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/PublisherService.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/PublisherService.java
index 1f41724fa..7969ffe6c 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/PublisherService.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/PublisherService.java
@@ -69,7 +69,7 @@ public class PublisherService extends
PublisherServiceGrpc.PublisherServiceImplB
} catch (Exception e) {
log.error("Error code {}, error message {}",
StatusCode.EVENTMESH_SEND_ASYNC_MSG_ERR.getRetCode(),
StatusCode.EVENTMESH_SEND_ASYNC_MSG_ERR.getErrMsg(), e);
-
ServiceUtils.completed(StatusCode.EVENTMESH_SEND_ASYNC_MSG_ERR, e.getMessage(),
emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_SEND_ASYNC_MSG_ERR,
e.getMessage(), emitter);
}
});
}
@@ -98,7 +98,7 @@ public class PublisherService extends
PublisherServiceGrpc.PublisherServiceImplB
} catch (Exception e) {
log.error("Error code {}, error message {}",
StatusCode.EVENTMESH_REQUEST_REPLY_MSG_ERR.getRetCode(),
StatusCode.EVENTMESH_REQUEST_REPLY_MSG_ERR.getErrMsg(), e);
- ServiceUtils.streamCompleted(request,
StatusCode.EVENTMESH_REQUEST_REPLY_MSG_ERR, e.getMessage(), emitter);
+ ServiceUtils.sendStreamResponseCompleted(request,
StatusCode.EVENTMESH_REQUEST_REPLY_MSG_ERR, e.getMessage(), emitter);
}
});
}
@@ -126,7 +126,7 @@ public class PublisherService extends
PublisherServiceGrpc.PublisherServiceImplB
} catch (Exception e) {
log.error("Error code {}, error message {}",
StatusCode.EVENTMESH_BATCH_PUBLISH_ERR.getRetCode(),
StatusCode.EVENTMESH_BATCH_PUBLISH_ERR.getErrMsg(), e);
- ServiceUtils.completed(StatusCode.EVENTMESH_BATCH_PUBLISH_ERR,
e.getMessage(), emitter);
+
ServiceUtils.sendResponseCompleted(StatusCode.EVENTMESH_BATCH_PUBLISH_ERR,
e.getMessage(), emitter);
}
});
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/ServiceUtils.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/ServiceUtils.java
index b7a8db91f..128d058e9 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/ServiceUtils.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/ServiceUtils.java
@@ -152,7 +152,14 @@ public class ServiceUtils {
return true;
}
- public static void completed(StatusCode code, String message,
EventEmitter<CloudEvent> emitter) {
+ /**
+ * Sends a completed response event to the given EventEmitter.
+ *
+ * @param code The status code for the response.
+ * @param message The message for the response.
+ * @param emitter The EventEmitter to send the response event to.
+ */
+ public static void sendResponseCompleted(StatusCode code, String message,
EventEmitter<CloudEvent> emitter) {
Instant instant = now();
CloudEvent.Builder builder =
CloudEvent.newBuilder().setId(RandomStringUtils.generateUUID())
@@ -166,32 +173,75 @@ public class ServiceUtils {
emitter.onCompleted();
}
- public static void completed(StatusCode code, EventEmitter<CloudEvent>
emitter) {
+ /**
+ * Sends a completed response event to the emitter.
+ *
+ * @param code The status code of the response
+ * @param emitter The emitter to send the event to
+ */
+ public static void sendResponseCompleted(StatusCode code,
EventEmitter<CloudEvent> emitter) {
Instant instant = now();
CloudEvent.Builder builder = CloudEvent.newBuilder()
.putAttributes(ProtocolKey.GRPC_RESPONSE_CODE,
CloudEventAttributeValue.newBuilder().setCeString(code.getRetCode()).build())
- .putAttributes(ProtocolKey.GRPC_RESPONSE_MESSAGE,
-
CloudEventAttributeValue.newBuilder().setCeString(code.getErrMsg()).build())
+ .putAttributes(ProtocolKey.GRPC_RESPONSE_MESSAGE,
CloudEventAttributeValue.newBuilder().setCeString(code.getErrMsg()).build())
.putAttributes(ProtocolKey.GRPC_RESPONSE_TIME,
CloudEventAttributeValue.newBuilder()
.setCeTimestamp(Timestamp.newBuilder().setSeconds(instant.getEpochSecond()).setNanos(instant.getNano()).build()).build());
emitter.onNext(builder.build());
emitter.onCompleted();
}
- public static void streamCompleted(CloudEvent cloudEvent, StatusCode code,
EventEmitter<CloudEvent> emitter) {
+ /**
+ * Sends a completed response event to the emitter for a stream.
+ *
+ * @param cloudEvent The original CloudEvent
+ * @param code The status code of the response
+ * @param emitter The emitter to send the event to
+ */
+ public static void sendStreamResponseCompleted(CloudEvent cloudEvent,
StatusCode code, EventEmitter<CloudEvent> emitter) {
+ sendStreamResponse(cloudEvent, code, emitter);
+ emitter.onCompleted();
+ }
+
+ /**
+ * Sends a completed response event to the emitter for a stream with a
custom message.
+ *
+ * @param cloudEvent The original CloudEvent
+ * @param code The status code of the response
+ * @param message The custom message for the response
+ * @param emitter The emitter to send the event to
+ */
+ public static void sendStreamResponseCompleted(CloudEvent cloudEvent,
StatusCode code, String message, EventEmitter<CloudEvent> emitter) {
+ sendStreamResponse(cloudEvent, code, message, emitter);
+ emitter.onCompleted();
+ }
+
+ /**
+ * Sends a response event to the emitter for a stream.
+ *
+ * @param cloudEvent The original CloudEvent
+ * @param code The status code of the response
+ * @param emitter The emitter to send the event to
+ */
+ public static void sendStreamResponse(CloudEvent cloudEvent, StatusCode
code, EventEmitter<CloudEvent> emitter) {
Instant instant = now();
CloudEvent.Builder builder = CloudEvent.newBuilder(cloudEvent)
.putAttributes(ProtocolKey.GRPC_RESPONSE_CODE,
CloudEventAttributeValue.newBuilder().setCeString(code.getRetCode()).build())
- .putAttributes(ProtocolKey.GRPC_RESPONSE_MESSAGE,
-
CloudEventAttributeValue.newBuilder().setCeString(code.getErrMsg()).build())
+ .putAttributes(ProtocolKey.GRPC_RESPONSE_MESSAGE,
CloudEventAttributeValue.newBuilder().setCeString(code.getErrMsg()).build())
.putAttributes(ProtocolKey.GRPC_RESPONSE_TIME,
CloudEventAttributeValue.newBuilder()
.setCeTimestamp(Timestamp.newBuilder().setSeconds(instant.getEpochSecond()).setNanos(instant.getNano()).build()).build());
emitter.onNext(builder.build());
- emitter.onCompleted();
}
- public static void streamCompleted(CloudEvent cloudEvent, StatusCode code,
String message, EventEmitter<CloudEvent> emitter) {
+ /**
+ * Sends a response event to the emitter for a stream with a custom
message.
+ *
+ * @param cloudEvent The original CloudEvent
+ * @param code The status code of the response
+ * @param message The custom message for the response
+ * @param emitter The emitter to send the event to
+ */
+ public static void sendStreamResponse(CloudEvent cloudEvent, StatusCode
code, String message, EventEmitter<CloudEvent> emitter) {
Instant instant = OffsetDateTime.now().toInstant();
CloudEvent.Builder builder = CloudEvent.newBuilder(cloudEvent)
.putAttributes(ProtocolKey.GRPC_RESPONSE_CODE,
CloudEventAttributeValue.newBuilder().setCeString(code.getRetCode()).build())
@@ -201,9 +251,13 @@ public class ServiceUtils {
.setCeTimestamp(Timestamp.newBuilder().setSeconds(instant.getEpochSecond()).setNanos(instant.getNano()).build()).build());
emitter.onNext(builder.build());
- emitter.onCompleted();
}
+ /**
+ * Returns the current instant.
+ *
+ * @return The current instant
+ */
private static Instant now() {
return OffsetDateTime.of(LocalDateTime.now(ZoneId.systemDefault()),
ZoneOffset.UTC).toInstant();
}
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 5bd4d9208..40cd41787 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
@@ -31,6 +31,7 @@ import org.apache.eventmesh.common.utils.JsonUtils;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.CountDownLatch;
import io.grpc.stub.StreamObserver;
@@ -71,7 +72,7 @@ public class SubStreamHandler<T> extends Thread {
@Override
public void onNext(final CloudEvent message) {
T msg =
EventMeshCloudEventBuilder.buildMessageFromEventMeshCloudEvent(message,
listener.getProtocolType());
- if (msg instanceof Map) {
+ if (msg instanceof Set) {
if (log.isInfoEnabled()) {
log.info("Received message from Server:{}", message);
}
diff --git
a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/util/EventMeshCloudEventBuilder.java
b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/util/EventMeshCloudEventBuilder.java
index a98766b95..feaef016b 100644
---
a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/util/EventMeshCloudEventBuilder.java
+++
b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/util/EventMeshCloudEventBuilder.java
@@ -269,8 +269,9 @@ public class EventMeshCloudEventBuilder {
// This is GRPC response cloudEvent
if (StringUtils.isEmpty(seq) && StringUtils.isEmpty(uniqueId)) {
+ //The SubscriptionItem collection contains the content for the
subscription.
return (T) JsonUtils.parseTypeReferenceObject(content,
- new TypeReference<HashMap<String, String>>() {
+ new TypeReference<Set<HashMap<String, String>>>() {
});
}
diff --git
a/eventmesh-sdks/eventmesh-sdk-java/src/test/java/org/apache/eventmesh/client/grpc/util/EventMeshCloudEventBuilderTest.java
b/eventmesh-sdks/eventmesh-sdk-java/src/test/java/org/apache/eventmesh/client/grpc/util/EventMeshCloudEventBuilderTest.java
index 201413f10..9e98cb8bf 100644
---
a/eventmesh-sdks/eventmesh-sdk-java/src/test/java/org/apache/eventmesh/client/grpc/util/EventMeshCloudEventBuilderTest.java
+++
b/eventmesh-sdks/eventmesh-sdk-java/src/test/java/org/apache/eventmesh/client/grpc/util/EventMeshCloudEventBuilderTest.java
@@ -29,11 +29,14 @@ import
org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent.CloudEve
import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEventBatch;
import
org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils;
import org.apache.eventmesh.common.protocol.grpc.common.ProtocolKey;
+import org.apache.eventmesh.common.utils.JsonUtils;
import java.net.URI;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.UUID;
import org.junit.Assert;
@@ -166,5 +169,17 @@ public class EventMeshCloudEventBuilderTest {
Assert.assertTrue(event1 instanceof io.cloudevents.CloudEvent);
Object event2 =
EventMeshCloudEventBuilder.buildMessageFromEventMeshCloudEvent(eventmesh1,
EventMeshProtocolType.EVENT_MESH_MESSAGE);
Assert.assertTrue(event2 instanceof EventMeshMessage);
+
+ final Map<String, CloudEventAttributeValue> attributeValueMap =
EventMeshCloudEventBuilder.buildCommonCloudEventAttributes(clientConfig,
+ EventMeshProtocolType.EVENT_MESH_MESSAGE);
+ attributeValueMap.put(ProtocolKey.CONSUMERGROUP,
CloudEventAttributeValue.newBuilder().setCeString(clientConfig.getConsumerGroup()).build());
+ attributeValueMap.put(ProtocolKey.DATA_CONTENT_TYPE,
CloudEventAttributeValue.newBuilder().setCeString("application/json").build());
+ Set<SubscriptionItem> set = new HashSet<>();
+ SubscriptionItem subscriptionItem = new SubscriptionItem("111",
SubscriptionMode.CLUSTERING, SubscriptionType.SYNC);
+ set.add(subscriptionItem);
+ CloudEvent eventmesh2 =
CloudEvent.newBuilder().setSpecVersion("1.0").setType("eventmesh").setSource(URI.create("/").toString())
+
.setId(UUID.randomUUID().toString()).putAllAttributes(attributeValueMap).setTextData(JsonUtils.toJSONString(set)).build();
+ Object event4 =
EventMeshCloudEventBuilder.buildMessageFromEventMeshCloudEvent(eventmesh2,
EventMeshProtocolType.CLOUD_EVENTS);
+ Assert.assertTrue(event4 instanceof Set);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
