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 857ad6b7d [ISSUE #4190] Code optimization for the EventMeshConsumer
class.
857ad6b7d is described below
commit 857ad6b7d22f4b458ecaeb2b13668cdce250c9bc
Author: yanrongzhen <[email protected]>
AuthorDate: Mon Aug 14 15:40:00 2023 +0800
[ISSUE #4190] Code optimization for the EventMeshConsumer class.
---
.../protocol/grpc/consumer/EventMeshConsumer.java | 58 ++++++++++++----------
1 file changed, 33 insertions(+), 25 deletions(-)
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 ba913fa57..52f5c87f5 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
@@ -65,17 +65,17 @@ public class EventMeshConsumer {
private final transient String consumerGroup;
- private final transient EventMeshGrpcServer eventMeshGrpcServer;
+ private final EventMeshGrpcServer eventMeshGrpcServer;
- private final transient EventMeshGrpcConfiguration
eventMeshGrpcConfiguration;
+ private final EventMeshGrpcConfiguration eventMeshGrpcConfiguration;
- private final transient MQConsumerWrapper persistentMqConsumer;
+ private final MQConsumerWrapper persistentMqConsumer;
- private final transient MQConsumerWrapper broadcastMqConsumer;
+ private final MQConsumerWrapper broadcastMqConsumer;
- private final transient MessageHandler messageHandler;
+ private final MessageHandler messageHandler;
- private transient ServiceState serviceState;
+ private ServiceState serviceState;
/**
* Key: topic Value: ConsumerGroupTopicConfig
@@ -201,36 +201,44 @@ public class EventMeshConsumer {
}
public void subscribe(final String topic, final SubscriptionMode
subscriptionMode) throws Exception {
- if (SubscriptionMode.CLUSTERING == subscriptionMode) {
- persistentMqConsumer.subscribe(topic);
- } else if (SubscriptionMode.BROADCASTING == subscriptionMode) {
- broadcastMqConsumer.subscribe(topic);
- } else {
- //log.error("Subscribe Failed. Incorrect Subscription Mode");
- throw new Exception("Subscribe Failed. Incorrect Subscription
Mode");
+ switch (subscriptionMode) {
+ case CLUSTERING:
+ persistentMqConsumer.subscribe(topic);
+ break;
+ case BROADCASTING:
+ broadcastMqConsumer.subscribe(topic);
+ break;
+ default:
+ throw new Exception("Subscribe Failed. Incorrect Subscription
Mode");
}
}
public void unsubscribe(final SubscriptionItem subscriptionItem) throws
Exception {
final SubscriptionMode mode = subscriptionItem.getMode();
final String topic = subscriptionItem.getTopic();
- if (SubscriptionMode.CLUSTERING == mode) {
- persistentMqConsumer.unsubscribe(topic);
- } else if (SubscriptionMode.BROADCASTING == mode) {
- broadcastMqConsumer.unsubscribe(topic);
- } else {
- throw new Exception("Unsubscribe Failed. Incorrect Subscription
Mode");
+ switch (mode) {
+ case CLUSTERING:
+ persistentMqConsumer.unsubscribe(topic);
+ break;
+ case BROADCASTING:
+ broadcastMqConsumer.unsubscribe(topic);
+ break;
+ default:
+ throw new Exception("Unsubscribe Failed. Incorrect
Subscription Mode");
}
}
public void updateOffset(final SubscriptionMode subscriptionMode, final
List<CloudEvent> events, final AbstractContext context)
throws Exception {
- if (SubscriptionMode.CLUSTERING == subscriptionMode) {
- persistentMqConsumer.updateOffset(events, context);
- } else if (SubscriptionMode.BROADCASTING == subscriptionMode) {
- broadcastMqConsumer.updateOffset(events, context);
- } else {
- throw new Exception("Subscribe Failed. Incorrect Subscription
Mode");
+ switch (subscriptionMode) {
+ case CLUSTERING:
+ persistentMqConsumer.updateOffset(events, context);
+ break;
+ case BROADCASTING:
+ broadcastMqConsumer.updateOffset(events, context);
+ break;
+ default:
+ throw new Exception("Subscribe Failed. Incorrect Subscription
Mode");
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
