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 bae130723 [ISSUE #4166] Fix grpc AsyncPublishInstance has no push
messages. (#4167)
bae130723 is described below
commit bae1307239d4db5623a691cdb510bf49cb8fe5c3
Author: Alonexc <[email protected]>
AuthorDate: Fri Jun 30 16:35:58 2023 +0800
[ISSUE #4166] Fix grpc AsyncPublishInstance has no push messages. (#4167)
* Create .lift.toml
* Update .lift.toml
* Update .lift.toml
* Update .lift.toml
* Update .lift.toml
* Update .lift.toml
* Update .lift.toml
* Update .lift.toml
* Update .lift.toml
* delete lift
* push message optimization methods and correction of topic.
* push message optimization methods and correction of topic.
---
.../src/main/java/org/apache/eventmesh/grpc/GrpcAbstractDemo.java | 8 ++++----
.../grpc/pub/cloudevents/CloudEventsBatchPublishInstance.java | 3 ++-
.../grpc/pub/cloudevents/CloudEventsPublishInstance.java | 3 ++-
.../grpc/pub/cloudevents/CloudEventsRequestInstance.java | 3 ++-
.../grpc/pub/eventmeshmessage/AsyncPublishBroadcast.java | 3 ++-
.../eventmesh/grpc/pub/eventmeshmessage/AsyncPublishInstance.java | 3 ++-
.../eventmesh/grpc/pub/eventmeshmessage/BatchPublishInstance.java | 3 ++-
.../eventmesh/grpc/pub/eventmeshmessage/RequestReplyInstance.java | 3 ++-
8 files changed, 18 insertions(+), 11 deletions(-)
diff --git
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/GrpcAbstractDemo.java
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/GrpcAbstractDemo.java
index 65ff999af..28e260b8d 100644
---
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/GrpcAbstractDemo.java
+++
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/GrpcAbstractDemo.java
@@ -53,10 +53,10 @@ public class GrpcAbstractDemo {
.build();
}
- protected static CloudEvent buildCloudEvent(final Map<String, String>
content) {
+ protected static CloudEvent buildCloudEvent(final Map<String, String>
content, String topic) {
return CloudEventBuilder.v1()
.withId(UUID.randomUUID().toString())
- .withSubject(ExampleConstants.EVENTMESH_GRPC_ASYNC_TEST_TOPIC)
+ .withSubject(topic)
.withSource(URI.create("/"))
.withDataContentType(ExampleConstants.CLOUDEVENT_CONTENT_TYPE)
.withType(EventMeshCommon.CLOUD_EVENTS_PROTOCOL_NAME)
@@ -66,10 +66,10 @@ public class GrpcAbstractDemo {
}
- protected static EventMeshMessage buildEventMeshMessage(final Map<String,
String> content) {
+ protected static EventMeshMessage buildEventMeshMessage(final Map<String,
String> content, String topic) {
return EventMeshMessage.builder()
.content(JsonUtils.toJSONString(content))
- .topic(ExampleConstants.EVENTMESH_GRPC_BROADCAT_TEST_TOPIC)
+ .topic(topic)
.uniqueId(RandomStringUtils.generateNum(30))
.bizSeqNo(RandomStringUtils.generateNum(30))
.build()
diff --git
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/cloudevents/CloudEventsBatchPublishInstance.java
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/cloudevents/CloudEventsBatchPublishInstance.java
index 03705c85c..264dcc64a 100644
---
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/cloudevents/CloudEventsBatchPublishInstance.java
+++
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/cloudevents/CloudEventsBatchPublishInstance.java
@@ -45,7 +45,8 @@ public class CloudEventsBatchPublishInstance extends
GrpcAbstractDemo {
final List<CloudEvent> cloudEventList = new ArrayList<>();
for (int i = 0; i < 5; i++) {
- cloudEventList.add(buildCloudEvent(content));
+ cloudEventList.add(buildCloudEvent(content,
+ ExampleConstants.EVENTMESH_GRPC_BROADCAT_TEST_TOPIC));
}
eventMeshGrpcProducer.publish(cloudEventList);
ThreadUtils.sleep(10, TimeUnit.SECONDS);
diff --git
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/cloudevents/CloudEventsPublishInstance.java
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/cloudevents/CloudEventsPublishInstance.java
index 309af8604..dc9cfca7a 100644
---
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/cloudevents/CloudEventsPublishInstance.java
+++
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/cloudevents/CloudEventsPublishInstance.java
@@ -44,7 +44,8 @@ public class CloudEventsPublishInstance extends
GrpcAbstractDemo {
content.put("content", "testAsyncMessage");
for (int i = 0; i < MESSAGE_SIZE; i++) {
- eventMeshGrpcProducer.publish(buildCloudEvent(content));
+ eventMeshGrpcProducer.publish(buildCloudEvent(content,
+ ExampleConstants.EVENTMESH_GRPC_ASYNC_TEST_TOPIC));
ThreadUtils.sleep(1, TimeUnit.SECONDS);
}
diff --git
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/cloudevents/CloudEventsRequestInstance.java
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/cloudevents/CloudEventsRequestInstance.java
index 589432f26..f733d1ce1 100644
---
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/cloudevents/CloudEventsRequestInstance.java
+++
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/cloudevents/CloudEventsRequestInstance.java
@@ -43,7 +43,8 @@ public class CloudEventsRequestInstance extends
GrpcAbstractDemo {
content.put("content", "testRequestReplyMessage");
for (int i = 0; i < MESSAGE_SIZE; i++) {
- eventMeshGrpcProducer.requestReply(buildCloudEvent(content),
EventMeshCommon.DEFAULT_TIME_OUT_MILLS);
+ eventMeshGrpcProducer.requestReply(buildCloudEvent(content,
+ ExampleConstants.EVENTMESH_GRPC_RR_TEST_TOPIC),
EventMeshCommon.DEFAULT_TIME_OUT_MILLS);
ThreadUtils.sleep(1, TimeUnit.SECONDS);
}
diff --git
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/AsyncPublishBroadcast.java
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/AsyncPublishBroadcast.java
index b4d7c489d..cee6b4da9 100644
---
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/AsyncPublishBroadcast.java
+++
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/AsyncPublishBroadcast.java
@@ -43,7 +43,8 @@ public class AsyncPublishBroadcast extends GrpcAbstractDemo {
content.put("content", "testAsyncMessage");
for (int i = 0; i < MESSAGE_SIZE; i++) {
- eventMeshGrpcProducer.publish(buildEventMeshMessage(content));
+ eventMeshGrpcProducer.publish(buildEventMeshMessage(content,
+ ExampleConstants.EVENTMESH_GRPC_BROADCAT_TEST_TOPIC));
ThreadUtils.sleep(1, TimeUnit.SECONDS);
}
ThreadUtils.sleep(30, TimeUnit.SECONDS);
diff --git
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/AsyncPublishInstance.java
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/AsyncPublishInstance.java
index 162d360ed..57d29aca6 100644
---
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/AsyncPublishInstance.java
+++
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/AsyncPublishInstance.java
@@ -43,7 +43,8 @@ public class AsyncPublishInstance extends GrpcAbstractDemo {
content.put("content", "testAsyncMessage");
for (int i = 0; i < MESSAGE_SIZE; i++) {
- buildEventMeshMessage(content);
+ eventMeshGrpcProducer.publish(buildEventMeshMessage(content,
+ ExampleConstants.EVENTMESH_GRPC_ASYNC_TEST_TOPIC));
ThreadUtils.sleep(1, TimeUnit.SECONDS);
}
ThreadUtils.sleep(30, TimeUnit.SECONDS);
diff --git
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/BatchPublishInstance.java
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/BatchPublishInstance.java
index 941f72ce8..3deed5166 100644
---
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/BatchPublishInstance.java
+++
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/BatchPublishInstance.java
@@ -45,7 +45,8 @@ public class BatchPublishInstance extends GrpcAbstractDemo {
List<EventMeshMessage> messageList = new ArrayList<>();
for (int i = 0; i < 5; i++) {
- messageList.add(buildEventMeshMessage(content));
+ messageList.add(buildEventMeshMessage(content,
+ ExampleConstants.EVENTMESH_GRPC_BROADCAT_TEST_TOPIC));
}
eventMeshGrpcProducer.publish(messageList);
diff --git
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/RequestReplyInstance.java
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/RequestReplyInstance.java
index 723c0637b..4d71f93b7 100644
---
a/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/RequestReplyInstance.java
+++
b/eventmesh-examples/src/main/java/org/apache/eventmesh/grpc/pub/eventmeshmessage/RequestReplyInstance.java
@@ -45,7 +45,8 @@ public class RequestReplyInstance extends GrpcAbstractDemo {
content.put("content", "testRequestReplyMessage");
for (int i = 0; i < MESSAGE_SIZE; i++) {
-
eventMeshGrpcProducer.requestReply(buildEventMeshMessage(content),
+
eventMeshGrpcProducer.requestReply(buildEventMeshMessage(content,
+ ExampleConstants.EVENTMESH_GRPC_RR_TEST_TOPIC),
EventMeshCommon.DEFAULT_TIME_OUT_MILLS);
ThreadUtils.sleep(1, TimeUnit.SECONDS);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
