You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by mi...@apache.org on 2022/11/14 07:34:19 UTC
[incubator-eventmesh] branch protocol-amqp updated: implementation of amqp producer
This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch protocol-amqp
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/protocol-amqp by this push:
new f79d303f implementation of amqp producer
new 3056d758 Merge pull request #2208 from jackyluo-learning/protocol-amqp-shadow4
f79d303f is described below
commit f79d303fbf89a62d0a3a17590a32514cca39939f
Author: jackyluo <15...@qq.com>
AuthorDate: Mon Nov 14 15:10:46 2022 +0800
implementation of amqp producer
---
eventmesh-runtime/build.gradle | 1 +
.../core/protocol/amqp/producer/AmqpProducer.java | 89 ++++++++++++++++++++++
.../amqp/producer/AmqpProducerManager.java | 67 ++++++++++++++++
.../protocol/amqp/producer/PutMessageContext.java | 66 ++++++++++++++++
4 files changed, 223 insertions(+)
diff --git a/eventmesh-runtime/build.gradle b/eventmesh-runtime/build.gradle
index 60376611..318a052d 100644
--- a/eventmesh-runtime/build.gradle
+++ b/eventmesh-runtime/build.gradle
@@ -61,6 +61,7 @@ dependencies {
implementation project(":eventmesh-protocol-plugin:eventmesh-protocol-cloudevents")
implementation project(":eventmesh-protocol-plugin:eventmesh-protocol-meshmessage")
implementation project(":eventmesh-protocol-plugin:eventmesh-protocol-openmessage")
+ implementation project(":eventmesh-protocol-plugin:eventmesh-protocol-amqp")
implementation project(":eventmesh-protocol-plugin:eventmesh-protocol-http")
implementation project(":eventmesh-protocol-plugin:eventmesh-protocol-webhook")
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/amqp/producer/AmqpProducer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/amqp/producer/AmqpProducer.java
new file mode 100644
index 00000000..3e9c0a64
--- /dev/null
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/amqp/producer/AmqpProducer.java
@@ -0,0 +1,89 @@
+package org.apache.eventmesh.runtime.core.protocol.amqp.producer;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.protocol.amqp.resolver.AmqpProtocolResolver;
+import org.apache.eventmesh.runtime.common.ServiceState;
+import org.apache.eventmesh.runtime.configuration.EventMeshAmqpConfiguration;
+import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.consumergroup.ProducerGroupConf;
+import org.apache.eventmesh.runtime.core.plugin.MQProducerWrapper;
+import org.apache.eventmesh.runtime.util.EventMeshUtil;
+
+import java.util.Properties;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.cloudevents.CloudEvent;
+
+public class AmqpProducer {
+
+ private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+ private ProducerGroupConf producerGroupConfig;
+
+ private MQProducerWrapper mqProducerWrapper;
+
+ private ServiceState serviceState;
+
+ public void send(PutMessageContext putMessageContext, SendCallback sendCallback)
+ throws Exception {
+ CloudEvent ce= AmqpProtocolResolver.buildEvent(putMessageContext.getMessage());
+ mqProducerWrapper.send(ce, sendCallback);
+ }
+
+
+ public synchronized void init(EventMeshAmqpConfiguration eventMeshAmqpConfiguration,
+ ProducerGroupConf producerGroupConfig) throws Exception {
+ this.producerGroupConfig = producerGroupConfig;
+
+ Properties keyValue = new Properties();
+ keyValue.put(EventMeshConstants.PRODUCER_GROUP, producerGroupConfig.getGroupName());
+ keyValue.put(EventMeshConstants.INSTANCE_NAME, EventMeshUtil.buildMeshClientID(
+ producerGroupConfig.getGroupName(), eventMeshAmqpConfiguration.eventMeshCluster));
+
+ //TODO for defibus
+ keyValue.put(EventMeshConstants.EVENT_MESH_IDC, eventMeshAmqpConfiguration.eventMeshIDC);
+ mqProducerWrapper = new MQProducerWrapper(
+ eventMeshAmqpConfiguration.eventMeshConnectorPluginType);
+ mqProducerWrapper.init(keyValue);
+ serviceState = ServiceState.INITED;
+ logger.info("AmqpProducer [{}] inited...........", producerGroupConfig.getGroupName());
+ }
+
+ public synchronized void start() throws Exception {
+ if (serviceState == null || ServiceState.RUNNING.equals(serviceState)) {
+ return;
+ }
+
+ mqProducerWrapper.start();
+ serviceState = ServiceState.RUNNING;
+ logger.info("AmqpProducer [{}] started..........", producerGroupConfig.getGroupName());
+ }
+
+ public synchronized void shutdown() throws Exception {
+ if (serviceState == null || ServiceState.INITED.equals(serviceState)) {
+ return;
+ }
+
+ mqProducerWrapper.shutdown();
+ serviceState = ServiceState.STOPED;
+ logger.info("AmqpProducer [{}] shutdown.........", producerGroupConfig.getGroupName());
+ }
+
+ public ServiceState getStatus() {
+ return this.serviceState;
+ }
+
+
+
+
+
+
+
+
+
+
+
+
+}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/amqp/producer/AmqpProducerManager.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/amqp/producer/AmqpProducerManager.java
new file mode 100644
index 00000000..31e8d09a
--- /dev/null
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/amqp/producer/AmqpProducerManager.java
@@ -0,0 +1,67 @@
+package org.apache.eventmesh.runtime.core.protocol.amqp.producer;
+
+import org.apache.eventmesh.runtime.boot.EventMeshAmqpServer;
+import org.apache.eventmesh.runtime.core.consumergroup.ProducerGroupConf;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AmqpProducerManager {
+
+ public Logger logger = LoggerFactory.getLogger(this.getClass());
+
+ private ConcurrentHashMap<String, AmqpProducer> producerTable = new ConcurrentHashMap<>();
+
+ private EventMeshAmqpServer amqpServer;
+
+ public void init() throws Exception {
+ logger.info("AMQP ProducerManager inited......");
+ }
+
+ public void start() throws Exception {
+ logger.info("AMQP ProducerManager started......");
+ }
+
+ public AmqpProducer getAmqpProducer(String producerGroup) {
+
+ return producerTable.computeIfAbsent(producerGroup, pr -> {
+ ProducerGroupConf producerGroupConfig = new ProducerGroupConf(producerGroup);
+ AmqpProducer amqpProducer = null;
+ try {
+ amqpProducer = createAmqpProducer(producerGroupConfig);
+ amqpProducer.start();
+ return amqpProducer;
+ } catch (Exception e) {
+ e.printStackTrace();
+ logger.info("create amqp producer error {}", producerGroup, e);
+ }
+ return null;
+ });
+ }
+
+ private synchronized AmqpProducer createAmqpProducer(
+ ProducerGroupConf producerGroupConfig) throws Exception {
+ if (producerTable.containsKey(producerGroupConfig.getGroupName())) {
+ return producerTable.get(producerGroupConfig.getGroupName());
+ }
+ AmqpProducer amqpProducer = new AmqpProducer();
+ amqpProducer.init(amqpServer.getEventMeshAmqpConfiguration(),
+ producerGroupConfig);
+ producerTable.put(producerGroupConfig.getGroupName(), amqpProducer);
+ return amqpProducer;
+ }
+
+ public void shutdown() {
+ for (AmqpProducer amqpProducer : producerTable.values()) {
+ try {
+ amqpProducer.shutdown();
+ } catch (Exception ex) {
+ logger.error("shutdown amqpProducer [{}] err", amqpProducer, ex);
+ }
+ }
+ logger.info("amqpProducerManager shutdown......");
+ }
+
+}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/amqp/producer/PutMessageContext.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/amqp/producer/PutMessageContext.java
new file mode 100644
index 00000000..df6f6a8b
--- /dev/null
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/amqp/producer/PutMessageContext.java
@@ -0,0 +1,66 @@
+package org.apache.eventmesh.runtime.core.protocol.amqp.producer;
+
+import org.apache.eventmesh.common.protocol.amqp.AmqpMessage;
+
+import java.util.Map;
+import java.util.Set;
+
+public class PutMessageContext {
+
+ private String exchangeName;
+ private String routingKey;
+ private Map<String, Object> props;
+ private AmqpMessage message;
+ // temporary impl
+ Set<String> queues;
+
+
+ public PutMessageContext(String exchangeName, String routingKey, Map<String, Object> props,
+ AmqpMessage message, Set<String> queues) {
+ this.exchangeName = exchangeName;
+ this.routingKey = routingKey;
+ this.props = props;
+ this.message = message;
+ this.queues = queues;
+ }
+
+ public String getExchangeName() {
+ return exchangeName;
+ }
+
+ public void setExchangeName(String exchangeName) {
+ this.exchangeName = exchangeName;
+ }
+
+ public String getRoutingKey() {
+ return routingKey;
+ }
+
+ public void setRoutingKey(String routingKey) {
+ this.routingKey = routingKey;
+ }
+
+ public Map<String, Object> getProps() {
+ return props;
+ }
+
+ public void setProps(Map<String, Object> props) {
+ this.props = props;
+ }
+
+ public AmqpMessage getMessage() {
+ return message;
+ }
+
+ public void setMessage(AmqpMessage message) {
+ this.message = message;
+ }
+
+ public Set<String> getQueues() {
+ return queues;
+ }
+
+ public void setQueues(Set<String> queues) {
+ this.queues = queues;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org