Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Issue #780] Modify the define level of EventListener from Topic to Consumer #781

Merged
merged 8 commits into from
Feb 23, 2022
Merged

[Issue #780] Modify the define level of EventListener from Topic to Consumer #781

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
e1a1807
modify: add group field in UserAgent, delete ProducerGroup and Consum…
lrhkobe Feb 17, 2022
b6e03a5
Merge branch 'master' of github.com:apache/incubator-eventmesh
lrhkobe Feb 17, 2022
3e09bfb
modify: fix checksyle error
lrhkobe Feb 18, 2022
991f3fb
modify: fix checksyle error in ClientGroupWrapper.java
lrhkobe Feb 18, 2022
c090f5b
modify: move EventListner in the level of Consumer instead of binding…
lrhkobe Feb 22, 2022
6f787aa
Merge branch 'master' of github.com:apache/incubator-eventmesh
lrhkobe Feb 22, 2022
5bbab55
modify: fix the eventListener level problem in grpc protocal
lrhkobe Feb 22, 2022
b99d44d
modify: fix the eventListener problem in test case
lrhkobe Feb 22, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion ...gin/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/consumer/Consumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,11 @@ public interface Consumer extends LifeCycle {

void updateOffset(List<CloudEvent> cloudEvents, AbstractContext context);

void subscribe(String topic, final EventListener listener) throws Exception;
//void subscribe(String topic, final EventListener listener) throws Exception;

void subscribe(String topic) throws Exception;

void unsubscribe(String topic);

void registerEventListener(EventListener listener);
}
23 changes: 9 additions & 14 deletions ...etmq/src/main/java/org/apache/eventmesh/connector/rocketmq/consumer/PushConsumerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ public class PushConsumerImpl {
private final DefaultMQPushConsumer rocketmqPushConsumer;
private final Properties properties;
private AtomicBoolean started = new AtomicBoolean(false);
private final Map<String, EventListener> subscribeTable = new ConcurrentHashMap<>();
private EventListener eventListener;
private final ClientConfig clientConfig;

public PushConsumerImpl(final Properties properties) {
Expand Down Expand Up @@ -134,9 +134,7 @@ public DefaultMQPushConsumer getRocketmqPushConsumer() {
return rocketmqPushConsumer;
}


public void subscribe(String topic, String subExpression, EventListener listener) {
this.subscribeTable.put(topic, listener);
public void subscribe(String topic, String subExpression) {
try {
this.rocketmqPushConsumer.subscribe(topic, subExpression);
} catch (MQClientException e) {
Expand All @@ -146,7 +144,6 @@ public void subscribe(String topic, String subExpression, EventListener listener


public void unsubscribe(String topic) {
this.subscribeTable.remove(topic);
try {
this.rocketmqPushConsumer.unsubscribe(topic);
} catch (Exception e) {
Expand Down Expand Up @@ -197,9 +194,7 @@ public EventMeshConsumeConcurrentlyStatus handleMessage(MessageExt msg,
cloudEvent = cloudEventBuilder.build();
}

EventListener listener = PushConsumerImpl.this.subscribeTable.get(msg.getTopic());

if (listener == null) {
if (eventListener == null) {
throw new ConnectorRuntimeException(String.format("The topic/queue %s isn't attached to this consumer",
msg.getTopic()));
}
Expand Down Expand Up @@ -231,7 +226,7 @@ public void commit(EventMeshAction action) {

eventMeshAsyncConsumeContext.setAbstractContext(context);

listener.consume(cloudEvent, eventMeshAsyncConsumeContext);
eventListener.consume(cloudEvent, eventMeshAsyncConsumeContext);

return EventMeshConsumeConcurrentlyStatus.valueOf(
contextProperties.getProperty(NonStandardKeys.MESSAGE_CONSUME_STATUS));
Expand Down Expand Up @@ -270,9 +265,7 @@ public EventMeshConsumeConcurrentlyStatus handleMessage(MessageExt msg,
cloudEvent = cloudEventBuilder.build();
}

EventListener listener = PushConsumerImpl.this.subscribeTable.get(msg.getTopic());

if (listener == null) {
if (eventListener == null) {
throw new ConnectorRuntimeException(String.format("The topic/queue %s isn't attached to this consumer",
msg.getTopic()));
}
Expand Down Expand Up @@ -306,12 +299,14 @@ public void commit(EventMeshAction action) {

eventMeshAsyncConsumeContext.setAbstractContext(context);

listener.consume(cloudEvent, eventMeshAsyncConsumeContext);
eventListener.consume(cloudEvent, eventMeshAsyncConsumeContext);

return EventMeshConsumeConcurrentlyStatus.valueOf(
contextProperties.getProperty(NonStandardKeys.MESSAGE_CONSUME_STATUS));
}
}


public void registerEventListener(EventListener listener) {
this.eventListener = listener;
}
}
9 changes: 7 additions & 2 deletions .../src/main/java/org/apache/eventmesh/connector/rocketmq/consumer/RocketMQConsumerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,8 +70,8 @@ public synchronized void init(Properties keyValue) throws Exception {
}

@Override
public void subscribe(String topic, EventListener listener) throws Exception {
pushConsumer.subscribe(topic, "*", listener);
public void subscribe(String topic) throws Exception {
pushConsumer.subscribe(topic, "*");
}

@Override
Expand Down Expand Up @@ -99,6 +99,11 @@ public void unsubscribe(String topic) {
pushConsumer.unsubscribe(topic);
}

@Override
public void registerEventListener(EventListener listener) {
pushConsumer.registerEventListener(listener);
}

@Override
public synchronized void shutdown() {
pushConsumer.shutdown();
Expand Down
10 changes: 1 addition & 9 deletions ...h-connector-rocketmq/src/test/java/org/apache/rocketmq/consumer/PushConsumerImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,15 +89,7 @@ public void testConsumeMessage() {
consumedMsg.setBody(testBody);
consumedMsg.putUserProperty(NonStandardKeys.MESSAGE_DESTINATION, "TOPIC");
consumedMsg.setTopic("HELLO_QUEUE");
consumer.subscribe("HELLO_QUEUE", "*", new EventListener() {

@Override
public void consume(CloudEvent cloudEvent, org.apache.eventmesh.api.AsyncConsumeContext context) {
assertThat(cloudEvent.getExtension("MESSAGE_ID")).isEqualTo("NewMsgId");
assertThat(cloudEvent.getData()).isEqualTo(testBody);
context.commit(EventMeshAction.CommitMessage);
}
});
consumer.subscribe("HELLO_QUEUE", "*");
((MessageListenerConcurrently) rocketmqPushConsumer
.getMessageListener()).consumeMessage(Collections.singletonList(consumedMsg), null);

Expand Down
13 changes: 9 additions & 4 deletions .../src/main/java/org/apache/eventmesh/connector/standalone/consumer/StandaloneConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,8 @@ public class StandaloneConsumer implements Consumer {

private StandaloneBroker standaloneBroker;

private EventListener listener;

private AtomicBoolean isStarted;

private final ConcurrentHashMap<String, SubScribeTask> subscribeTaskTable;
Expand Down Expand Up @@ -90,10 +92,8 @@ public void updateOffset(List<CloudEvent> cloudEvents, AbstractContext context)
}

@Override
public void subscribe(String topic, EventListener listener) throws Exception {
if (listener == null) {
throw new IllegalArgumentException("listener cannot be null");
}
public void subscribe(String topic) throws Exception {

if (subscribeTaskTable.containsKey(topic)) {
return;
}
Expand All @@ -116,4 +116,9 @@ public void unsubscribe(String topic) {
subscribeTaskTable.remove(topic);
}
}

@Override
public void registerEventListener(EventListener listener) {
this.listener = listener;
}
}
99 changes: 6 additions & 93 deletions ...in/java/org/apache/eventmesh/connector/standalone/consumer/StandaloneConsumerAdaptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,104 +69,17 @@ public void updateOffset(List<CloudEvent> cloudEvents, AbstractContext context)
}

@Override
public void subscribe(String topic, EventListener listener) throws Exception {
consumer.subscribe(topic, listener);
public void subscribe(String topic) throws Exception {
consumer.subscribe(topic);
}

@Override
public void unsubscribe(String topic) {
consumer.unsubscribe(topic);
}

//@Override
//public void init(Properties keyValue) throws Exception {
// String producerGroup = keyValue.getProperty("producerGroup");
//
// MessagingAccessPointImpl messagingAccessPoint = new MessagingAccessPointImpl(keyValue);
// consumer = (StandaloneConsumer) messagingAccessPoint.createConsumer(keyValue);
//
//}
//
//@Override
//public void updateOffset(List<Message> msgs, AbstractContext context) {
// for(Message message : msgs) {
// consumer.updateOffset(message);
// }
//}
//
//@Override
//public void subscribe(String topic, AsyncMessageListener listener) throws Exception {
// // todo: support subExpression
// consumer.subscribe(topic, "*", listener);
//}
//
//@Override
//public void unsubscribe(String topic) {
// consumer.unsubscribe(topic);
//}
//
//@Override
//public void subscribe(String topic, String subExpression, MessageListener listener) {
// throw new UnsupportedOperationException("not supported yet");
//}
//
//@Override
//public void subscribe(String topic, MessageSelector selector, MessageListener listener) {
// throw new UnsupportedOperationException("not supported yet");
//}
//
//@Override
//public <T> void subscribe(String topic, String subExpression, GenericMessageListener<T> listener) {
// throw new UnsupportedOperationException("not supported yet");
//}
//
//@Override
//public <T> void subscribe(String topic, MessageSelector selector, GenericMessageListener<T> listener) {
// throw new UnsupportedOperationException("not supported yet");
//}
//
//@Override
//public void subscribe(String topic, String subExpression, AsyncMessageListener listener) {
// throw new UnsupportedOperationException("not supported yet");
//}
//
//@Override
//public void subscribe(String topic, MessageSelector selector, AsyncMessageListener listener) {
// throw new UnsupportedOperationException("not supported yet");
//}
//
//@Override
//public <T> void subscribe(String topic, String subExpression, AsyncGenericMessageListener<T> listener) {
// throw new UnsupportedOperationException("not supported yet");
//}
//
//@Override
//public <T> void subscribe(String topic, MessageSelector selector, AsyncGenericMessageListener<T> listener) {
// throw new UnsupportedOperationException("not supported yet");
//}
//
//@Override
//public void updateCredential(Properties credentialProperties) {
//
//}
//
//@Override
//public boolean isStarted() {
// return consumer.isStarted();
//}
//
//@Override
//public boolean isClosed() {
// return consumer.isClosed();
//}
//
//@Override
//public void start() {
// consumer.start();
//}
//
//@Override
//public void shutdown() {
// consumer.shutdown();
//}
@Override
public void registerEventListener(EventListener listener) {
consumer.registerEventListener(listener);
}
}
10 changes: 5 additions & 5 deletions ...esh-runtime/src/main/java/org/apache/eventmesh/runtime/core/plugin/MQConsumerWrapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,8 +44,8 @@ public MQConsumerWrapper(String connectorPluginType) {
}
}

public void subscribe(String topic, EventListener listener) throws Exception {
meshMQPushConsumer.subscribe(topic, listener);
public void subscribe(String topic) throws Exception {
meshMQPushConsumer.subscribe(topic);
}

public void unsubscribe(String topic) throws Exception {
Expand All @@ -69,9 +69,9 @@ public synchronized void shutdown() throws Exception {
started.compareAndSet(false, true);
}

//public void registerMessageListener(MessageListenerConcurrently messageListenerConcurrently) {
// meshMQPushConsumer.registerMessageListener(messageListenerConcurrently);
//}
public void registerEventListener(EventListener listener) {
meshMQPushConsumer.registerEventListener(listener);
}

public void updateOffset(List<CloudEvent> events, AbstractContext eventMeshConsumeConcurrentlyContext) {
meshMQPushConsumer.updateOffset(events, eventMeshConsumeConcurrentlyContext);
Expand Down
8 changes: 6 additions & 2 deletions ...main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/EventMeshConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,8 @@ public synchronized void init() throws Exception {
keyValue.put("instanceName", EventMeshUtil.buildMeshClientID(consumerGroup,
eventMeshGrpcConfiguration.eventMeshCluster));
persistentMqConsumer.init(keyValue);
EventListener clusterEventListner = createEventListener(SubscriptionMode.CLUSTERING);
persistentMqConsumer.registerEventListener(clusterEventListner);

Properties broadcastKeyValue = new Properties();
broadcastKeyValue.put("isBroadcast", "true");
Expand All @@ -148,6 +150,8 @@ public synchronized void init() throws Exception {
broadcastKeyValue.put("instanceName", EventMeshUtil.buildMeshClientID(consumerGroup,
eventMeshGrpcConfiguration.eventMeshCluster));
broadcastMqConsumer.init(broadcastKeyValue);
EventListener broadcastEventListner = createEventListener(SubscriptionMode.BROADCASTING);
broadcastMqConsumer.registerEventListener(broadcastEventListner);

serviceState = ServiceState.INITED;
logger.info("EventMeshConsumer [{}] initialized.............", consumerGroup);
Expand Down Expand Up @@ -184,9 +188,9 @@ public ServiceState getStatus() {

public void subscribe(String topic, SubscriptionMode subscriptionMode) throws Exception {
if (SubscriptionMode.CLUSTERING.equals(subscriptionMode)) {
persistentMqConsumer.subscribe(topic, createEventListener(subscriptionMode));
persistentMqConsumer.subscribe(topic);
} else if (SubscriptionMode.BROADCASTING.equals(subscriptionMode)) {
broadcastMqConsumer.subscribe(topic, createEventListener(subscriptionMode));
broadcastMqConsumer.subscribe(topic);
} else {
logger.error("Subscribe Failed. Incorrect Subscription Mode");
throw new Exception("Subscribe Failed. Incorrect Subscription Mode");
Expand Down
Loading