This is an automated email from the ASF dual-hosted git repository.
pandaapo 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 f17b82ee4 [ISSUE #4652]Fix EventMeshGrpcConsumer subscribe webhook
send heartBeat throw NPE (#4653)
f17b82ee4 is described below
commit f17b82ee4a50dea72eec3574d6a2b654879fb44a
Author: mxsm <[email protected]>
AuthorDate: Fri Dec 15 17:32:14 2023 +0800
[ISSUE #4652]Fix EventMeshGrpcConsumer subscribe webhook send heartBeat
throw NPE (#4653)
* [ISSUE #4652]Fix EventMeshGrpcConsumer subscribe webhook send heartBeat
throw NPE
* fix code style
* fix runtime code style
---
.../common/protocol/grpc/common}/GrpcType.java | 2 +-
.../protocol/grpc/consumer/ConsumerManager.java | 2 +-
.../protocol/grpc/consumer/EventMeshConsumer.java | 2 +-
.../consumergroup/ConsumerGroupClient.java | 1 +
.../consumergroup/ConsumerGroupTopicConfig.java | 1 +
.../consumer/consumergroup/StreamTopicConfig.java | 1 +
.../consumer/consumergroup/WebhookTopicConfig.java | 1 +
.../grpc/processor/SubscribeProcessor.java | 2 +-
.../grpc/processor/SubscribeStreamProcessor.java | 2 +-
.../core/protocol/grpc/push/HandleMsgContext.java | 2 +-
.../core/protocol/grpc/push/MessageHandler.java | 2 +-
.../core/protocol/grpc/service/ServiceUtils.java | 2 +-
.../grpc/consumer/EventMeshGrpcConsumer.java | 88 ++++++++++++++++------
13 files changed, 75 insertions(+), 33 deletions(-)
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/GrpcType.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/GrpcType.java
similarity index 91%
rename from
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/GrpcType.java
rename to
eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/GrpcType.java
index 507942a5f..81317310c 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/GrpcType.java
+++
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/GrpcType.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup;
+package org.apache.eventmesh.common.protocol.grpc.common;
public enum GrpcType {
WEBHOOK, STREAM
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/ConsumerManager.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/ConsumerManager.java
index e4ec6a7a6..5552cbbf5 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/ConsumerManager.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/ConsumerManager.java
@@ -18,12 +18,12 @@
package org.apache.eventmesh.runtime.core.protocol.grpc.consumer;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
+import org.apache.eventmesh.common.protocol.grpc.common.GrpcType;
import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.boot.EventMeshGrpcServer;
import org.apache.eventmesh.runtime.common.ServiceState;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.ConsumerGroupClient;
-import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.GrpcType;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/EventMeshConsumer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/EventMeshConsumer.java
index 3a4fde027..5567dd9e8 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/EventMeshConsumer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/EventMeshConsumer.java
@@ -33,6 +33,7 @@ import
org.apache.eventmesh.api.meta.config.EventMeshMetaConfig;
import org.apache.eventmesh.common.Constants;
import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
+import org.apache.eventmesh.common.protocol.grpc.common.GrpcType;
import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.common.utils.ThreadUtils;
@@ -45,7 +46,6 @@ import
org.apache.eventmesh.runtime.core.consumergroup.ConsumerGroupTopicMetadat
import org.apache.eventmesh.runtime.core.plugin.MQConsumerWrapper;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.ConsumerGroupClient;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.ConsumerGroupTopicConfig;
-import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.GrpcType;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.StreamTopicConfig;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.WebhookTopicConfig;
import org.apache.eventmesh.runtime.core.protocol.grpc.push.HandleMsgContext;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/ConsumerGroupClient.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/ConsumerGroupClient.java
index 1d3b3c1cd..6d185803e 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/ConsumerGroupClient.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/ConsumerGroupClient.java
@@ -19,6 +19,7 @@ package
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent;
+import org.apache.eventmesh.common.protocol.grpc.common.GrpcType;
import org.apache.eventmesh.runtime.core.protocol.grpc.service.EventEmitter;
import java.util.Date;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/ConsumerGroupTopicConfig.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/ConsumerGroupTopicConfig.java
index c7044e4ed..a166dab29 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/ConsumerGroupTopicConfig.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/ConsumerGroupTopicConfig.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
+import org.apache.eventmesh.common.protocol.grpc.common.GrpcType;
public abstract class ConsumerGroupTopicConfig {
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java
index 6094e2210..1c461db46 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java
@@ -19,6 +19,7 @@ package
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent;
+import org.apache.eventmesh.common.protocol.grpc.common.GrpcType;
import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.eventmesh.runtime.core.protocol.grpc.service.EventEmitter;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/WebhookTopicConfig.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/WebhookTopicConfig.java
index 358cfef58..818efd8dc 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/WebhookTopicConfig.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/WebhookTopicConfig.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
+import org.apache.eventmesh.common.protocol.grpc.common.GrpcType;
import java.util.ArrayList;
import java.util.HashSet;
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 bd0b0f6c4..25386138a 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
@@ -21,6 +21,7 @@ import org.apache.eventmesh.api.exception.AclException;
import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent;
import
org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils;
+import org.apache.eventmesh.common.protocol.grpc.common.GrpcType;
import org.apache.eventmesh.common.protocol.grpc.common.StatusCode;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.utils.JsonUtils;
@@ -30,7 +31,6 @@ import org.apache.eventmesh.runtime.boot.EventMeshGrpcServer;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.ConsumerManager;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.EventMeshConsumer;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.ConsumerGroupClient;
-import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.GrpcType;
import org.apache.eventmesh.runtime.core.protocol.grpc.service.EventEmitter;
import org.apache.eventmesh.runtime.core.protocol.grpc.service.ServiceUtils;
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 aa21f5623..b0c029837 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
@@ -21,6 +21,7 @@ import org.apache.eventmesh.api.exception.AclException;
import org.apache.eventmesh.common.protocol.SubscriptionItem;
import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent;
import
org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils;
+import org.apache.eventmesh.common.protocol.grpc.common.GrpcType;
import org.apache.eventmesh.common.protocol.grpc.common.StatusCode;
import org.apache.eventmesh.common.protocol.http.common.RequestCode;
import org.apache.eventmesh.common.utils.JsonUtils;
@@ -29,7 +30,6 @@ import org.apache.eventmesh.runtime.boot.EventMeshGrpcServer;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.ConsumerManager;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.EventMeshConsumer;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.ConsumerGroupClient;
-import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.GrpcType;
import org.apache.eventmesh.runtime.core.protocol.grpc.service.EventEmitter;
import org.apache.eventmesh.runtime.core.protocol.grpc.service.ServiceUtils;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/HandleMsgContext.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/HandleMsgContext.java
index 63d40ff85..da666d942 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/HandleMsgContext.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/HandleMsgContext.java
@@ -19,10 +19,10 @@ package
org.apache.eventmesh.runtime.core.protocol.grpc.push;
import org.apache.eventmesh.api.AbstractContext;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
+import org.apache.eventmesh.common.protocol.grpc.common.GrpcType;
import org.apache.eventmesh.runtime.boot.EventMeshGrpcServer;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.EventMeshConsumer;
import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.ConsumerGroupTopicConfig;
-import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.GrpcType;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
import io.cloudevents.CloudEvent;
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/MessageHandler.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/MessageHandler.java
index bb3737c35..15e3ae93d 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/MessageHandler.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/MessageHandler.java
@@ -18,7 +18,7 @@
package org.apache.eventmesh.runtime.core.protocol.grpc.push;
import org.apache.eventmesh.common.ThreadPoolFactory;
-import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.GrpcType;
+import org.apache.eventmesh.common.protocol.grpc.common.GrpcType;
import org.apache.commons.collections4.MapUtils;
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 bb4a56193..c59c7e357 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
@@ -25,13 +25,13 @@ import
org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent;
import
org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent.CloudEventAttributeValue;
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.GrpcType;
import org.apache.eventmesh.common.protocol.grpc.common.ProtoSupport;
import org.apache.eventmesh.common.protocol.grpc.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.grpc.common.StatusCode;
import org.apache.eventmesh.common.utils.JsonUtils;
import org.apache.eventmesh.common.utils.RandomStringUtils;
import org.apache.eventmesh.runtime.constants.EventMeshConstants;
-import
org.apache.eventmesh.runtime.core.protocol.grpc.consumer.consumergroup.GrpcType;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
diff --git
a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/EventMeshGrpcConsumer.java
b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/EventMeshGrpcConsumer.java
index 868191d99..2ccc335f2 100644
---
a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/EventMeshGrpcConsumer.java
+++
b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/consumer/EventMeshGrpcConsumer.java
@@ -37,6 +37,7 @@ import
org.apache.eventmesh.common.protocol.grpc.cloudevents.HeartbeatServiceGrp
import
org.apache.eventmesh.common.protocol.grpc.cloudevents.HeartbeatServiceGrpc.HeartbeatServiceBlockingStub;
import org.apache.eventmesh.common.protocol.grpc.common.ClientType;
import
org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils;
+import org.apache.eventmesh.common.protocol.grpc.common.GrpcType;
import org.apache.eventmesh.common.protocol.grpc.common.ProtocolKey;
import org.apache.eventmesh.common.protocol.grpc.common.Response;
import org.apache.eventmesh.common.protocol.grpc.common.StatusCode;
@@ -45,6 +46,7 @@ import org.apache.eventmesh.common.utils.LogUtils;
import org.apache.commons.collections4.MapUtils;
+import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -52,6 +54,7 @@ import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import io.grpc.ManagedChannel;
@@ -68,7 +71,7 @@ public class EventMeshGrpcConsumer implements AutoCloseable {
private ManagedChannel channel;
private final EventMeshGrpcClientConfig clientConfig;
- private final Map<String, SubscriptionInfo> subscriptionMap = new
ConcurrentHashMap<>();
+ private final Map<String /* topic */, SubscriptionInfo> subscriptionMap =
new ConcurrentHashMap<>();
private final ScheduledThreadPoolExecutor scheduler = new
ScheduledThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),
new EventMeshThreadFactory("GRPCClientScheduler", true));
@@ -93,27 +96,26 @@ public class EventMeshGrpcConsumer implements AutoCloseable
{
heartBeat();
}
+ /**
+ * Subscribes to an event at a specified URL(Webhook).
+ *
+ * @param subscriptionItems The list of subscription items.
+ * @param url The URL to subscribe to.
+ * @return A response containing information about the subscription result.
+ */
public Response subscribe(final List<SubscriptionItem> subscriptionItems,
final String url) {
LogUtils.info(log, "Create subscription: {} , url: {}",
subscriptionItems, url);
- addSubscription(subscriptionItems, url);
+ addSubscription(subscriptionItems, url, GrpcType.WEBHOOK);
- final CloudEvent subscription =
EventMeshCloudEventBuilder.buildEventSubscription(clientConfig,
EventMeshProtocolType.EVENT_MESH_MESSAGE,
- url, subscriptionItems);
- try {
- CloudEvent response = consumerClient.subscribe(subscription);
- LogUtils.info(log, "Received response:{}", response);
- return Response.builder()
- .respCode(EventMeshCloudEventUtils.getResponseCode(response))
- .respMsg(EventMeshCloudEventUtils.getResponseMessage(response))
- .respTime(EventMeshCloudEventUtils.getResponseTime(response))
- .build();
- } catch (Exception e) {
- log.error("Error in subscribe.", e);
- }
- return null;
+ return subscribeWebhook(subscriptionItems, url);
}
+ /**
+ * Subscribes to a streaming.
+ *
+ * @param subscriptionItems The list of subscription items for streaming.
+ */
public void subscribe(final List<SubscriptionItem> subscriptionItems) {
LogUtils.info(log, "Create streaming subscription: {}",
subscriptionItems);
@@ -122,7 +124,7 @@ public class EventMeshGrpcConsumer implements AutoCloseable
{
return;
}
- addSubscription(subscriptionItems, SDK_STREAM_URL);
+ addSubscription(subscriptionItems, SDK_STREAM_URL, GrpcType.STREAM);
CloudEvent subscription =
EventMeshCloudEventBuilder.buildEventSubscription(clientConfig,
EventMeshProtocolType.EVENT_MESH_MESSAGE, null,
subscriptionItems);
@@ -135,9 +137,26 @@ public class EventMeshGrpcConsumer implements
AutoCloseable {
subStreamHandler.sendSubscription(subscription);
}
- private void addSubscription(final List<SubscriptionItem>
subscriptionItems, final String url) {
+ private Response subscribeWebhook(List<SubscriptionItem>
subscriptionItems, String url) {
+ final CloudEvent subscription =
EventMeshCloudEventBuilder.buildEventSubscription(clientConfig,
EventMeshProtocolType.EVENT_MESH_MESSAGE,
+ url, subscriptionItems);
+ try {
+ CloudEvent response = consumerClient.subscribe(subscription);
+ LogUtils.info(log, "Received response:{}", response);
+ return Response.builder()
+ .respCode(EventMeshCloudEventUtils.getResponseCode(response))
+ .respMsg(EventMeshCloudEventUtils.getResponseMessage(response))
+ .respTime(EventMeshCloudEventUtils.getResponseTime(response))
+ .build();
+ } catch (Exception e) {
+ log.error("Error in subscribe.", e);
+ }
+ return null;
+ }
+
+ private void addSubscription(final List<SubscriptionItem>
subscriptionItems, final String url, final GrpcType grpcType) {
for (SubscriptionItem item : subscriptionItems) {
- subscriptionMap.putIfAbsent(item.getTopic(), new
SubscriptionInfo(item, url));
+ subscriptionMap.putIfAbsent(item.getTopic(), new
SubscriptionInfo(item, url, grpcType));
}
}
@@ -246,14 +265,27 @@ public class EventMeshGrpcConsumer implements
AutoCloseable {
if (subscriptionMap.isEmpty()) {
return;
}
-
- final Map<String, List<SubscriptionItem>> subscriptionGroup =
subscriptionMap.values().stream()
+ final Collection<SubscriptionInfo> values = subscriptionMap.values();
+ final AtomicBoolean isStreamSub = new AtomicBoolean(false);
+ for (SubscriptionInfo info : values) {
+ if (info.grpcType == GrpcType.STREAM) {
+ isStreamSub.compareAndSet(false, true);
+ break;
+ }
+ }
+ final Map<String, List<SubscriptionItem>> subscriptionGroup =
values.stream()
.collect(Collectors.groupingBy(SubscriptionInfo::getUrl,
mapping(SubscriptionInfo::getSubscriptionItem, toList())));
subscriptionGroup.forEach((url, items) -> {
- CloudEvent subscription =
EventMeshCloudEventBuilder.buildEventSubscription(clientConfig,
EventMeshProtocolType.EVENT_MESH_MESSAGE, url,
- items);
- subStreamHandler.sendSubscription(subscription);
+ if (isStreamSub.get()) {
+ CloudEvent subscription =
EventMeshCloudEventBuilder.buildEventSubscription(clientConfig,
EventMeshProtocolType.EVENT_MESH_MESSAGE,
+ url,
+ items);
+ subStreamHandler.sendSubscription(subscription);
+ } else {
+ subscribeWebhook(items, url);
+ }
+
});
}
@@ -276,10 +308,16 @@ public class EventMeshGrpcConsumer implements
AutoCloseable {
private transient SubscriptionItem subscriptionItem;
private transient String url;
+ private GrpcType grpcType;
- SubscriptionInfo(final SubscriptionItem subscriptionItem, final String
url) {
+ SubscriptionInfo(final SubscriptionItem subscriptionItem, final String
url, final GrpcType grpcType) {
this.subscriptionItem = subscriptionItem;
this.url = url;
+ this.grpcType = grpcType;
+ }
+
+ public GrpcType getGrpcType() {
+ return grpcType;
}
public SubscriptionItem getSubscriptionItem() {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
