[EDGENT-447][Connectors] contribute a RabbitMQ connector
diff --git a/connectors/pom.xml b/connectors/pom.xml
index 9190cfa..edc2e5b 100644
--- a/connectors/pom.xml
+++ b/connectors/pom.xml
@@ -49,6 +49,7 @@
<module>websocket-jetty</module>
<module>websocket-misc</module>
<module>websocket-server</module>
+ <module>rabbitmq</module>
</modules>
</project>
diff --git a/connectors/rabbitmq/pom.xml b/connectors/rabbitmq/pom.xml
new file mode 100644
index 0000000..9c67fad
--- /dev/null
+++ b/connectors/rabbitmq/pom.xml
@@ -0,0 +1,70 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>edgent-connectors</artifactId>
+ <groupId>org.apache.edgent</groupId>
+ <version>1.3.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>edgent-connectors-rabbitmq</artifactId>
+
+ <name>Apache Edgent (Java 8): Connectors: RabbitMQ</name>
+
+ <properties>
+ <remote-resources-maven-plugin.remote-resources.dir>../../src/main/ibm-remote-resources</remote-resources-maven-plugin.remote-resources.dir>
+ <edgent.version></edgent.version>
+ <rabbitmq.version>5.1.2</rabbitmq.version>
+ </properties>
+
+ <dependencies>
+ <!-- edgent start -->
+ <dependency>
+ <groupId>org.apache.edgent</groupId>
+ <artifactId>edgent-api-topology</artifactId>
+ <version>1.3.0-SNAPSHOT</version>
+ </dependency>
+ <!-- edgent end -->
+
+ <!-- edgent test start -->
+ <dependency>
+ <groupId>org.apache.edgent</groupId>
+ <artifactId>edgent-providers-direct</artifactId>
+ <version>1.3.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.edgent</groupId>
+ <artifactId>edgent-providers-direct</artifactId>
+ <version>1.3.0-SNAPSHOT</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.edgent</groupId>
+ <artifactId>edgent-connectors-common</artifactId>
+ <version>1.3.0-SNAPSHOT</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.edgent</groupId>
+ <artifactId>edgent-api-topology</artifactId>
+ <version>1.3.0-SNAPSHOT</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <!-- edgent test end -->
+
+ <!-- rabbitmq start -->
+ <dependency>
+ <groupId>com.rabbitmq</groupId>
+ <artifactId>amqp-client</artifactId>
+ <version>${rabbitmq.version}</version>
+ </dependency>
+ <!-- rabbitmq end -->
+ </dependencies>
+
+</project>
\ No newline at end of file
diff --git a/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/RabbitmqConfigKeyConstants.java b/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/RabbitmqConfigKeyConstants.java
new file mode 100644
index 0000000..f35a8fe
--- /dev/null
+++ b/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/RabbitmqConfigKeyConstants.java
@@ -0,0 +1,92 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package org.apache.edgent.connectors.rabbitmq;
+
+/**
+ * Defines all RabbitMQ config key constant items.
+ */
+public class RabbitmqConfigKeyConstants {
+
+ /**
+ * config key for connection URI, eg: amqp://userName:password@hostName:portNumber/virtualHost
+ */
+ public static final String RABBITMQ_CONFIG_KEY_URI = "rabbitmq.connection.uri";
+
+ /**
+ * config key for RabbitMQ server host
+ */
+ public static final String RABBITMQ_CONFIG_KEY_HOST = "rabbitmq.connection.host";
+
+ /**
+ * config key for RabbitMQ server port, default port is : 5672
+ */
+ public static final String RABBITMQ_CONFIG_KEY_PORT = "rabbitmq.connection.port";
+
+ /**
+ * config key for virtual host which used to split multi-users
+ */
+ public static final String RABBITMQ_CONFIG_KEY_VIRTUAL_HOST = "rabbitmq.connection.virtualHost";
+
+ /**
+ * config key for authorization (user name)
+ */
+ public static final String RABBITMQ_CONFIG_KEY_AUTH_NAME = "rabbitmq.connection.authUsername";
+
+ /**
+ * config key for authorization (password)
+ */
+ public static final String RABBITMQ_CONFIG_KEY_AUTH_PASSWORD = "rabbitmq.connection.authPassword";
+
+ /**
+ * config key for specifying whether enable auto recovery or not.
+ */
+ public static final String RABBITMQ_CONFIG_KEY_AUTO_RECOVERY = "rabbitmq.connection.autoRecovery";
+
+ /**
+ * config key for connection timeout
+ */
+ public static final String RABBITMQ_CONFIG_KEY_TIMEOUT = "rabbitmq.connection.timeout";
+
+ /**
+ * config key for connection network recovery interval
+ */
+ public static final String RABBITMQ_CONFIG_KEY_NETWORK_RECOVERY_INTERVAL = "rabbitmq.connection.networkRecoveryInterval";
+
+ /**
+ * config key for connection requested-heartbeat
+ */
+ public static final String RABBITMQ_CONFIG_KEY_REQUESTED_HEARTBEAT = "rabbitmq.connection.requestedHeartbeat";
+
+ /**
+ * config key for specifying whether enable topology recovery or not.
+ */
+ public static final String RABBITMQ_CONFIG_KEY_TOPOLOGY_RECOVERY_ENABLED = "rabbitmq.connection.topologyRecoveryEnabled";
+
+ /**
+ * config key for connection requested channel max num
+ */
+ public static final String RABBITMQ_CONFIG_KEY_REQUESTED_CHANNEL_MAX = "rabbitmq.connection.requestedChannelMax";
+
+ /**
+ * config key for connection requested frame max
+ */
+ public static final String RABBITMQ_CONFIG_KEY_REQUESTED_FRAME_MAX = "rabbitmq.connection.requestedFrameMax";
+
+}
diff --git a/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/RabbitmqConsumer.java b/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/RabbitmqConsumer.java
new file mode 100644
index 0000000..4c2e881
--- /dev/null
+++ b/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/RabbitmqConsumer.java
@@ -0,0 +1,87 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package org.apache.edgent.connectors.rabbitmq;
+
+import org.apache.edgent.connectors.rabbitmq.runtime.RabbitmqConnector;
+import org.apache.edgent.connectors.rabbitmq.runtime.RabbitmqSubscriber;
+import org.apache.edgent.function.Function;
+import org.apache.edgent.function.Supplier;
+import org.apache.edgent.topology.TStream;
+import org.apache.edgent.topology.Topology;
+
+import java.util.Map;
+
+/**
+ * {@code RabbitmqConsumer} is a consumer to consume messages from a RabbitMQ messaging broker
+ * <p>
+ * The connector uses and includes components from the RabbitMQ 3.7.3 release.
+ * It has been successfully tested against 3.7.3.
+ * For more information about RabbitMQ see <a href="http://www.rabbitmq.com/">http://www.rabbitmq.com/</a>
+ * </p>
+ * Smaple use:
+ * <pre>{@code
+ * Map<String, Object> config = new HashMap<>();
+ * config.put(RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_HOST, "127.0.0.1");
+ * config.put(RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_PORT, 5672);
+ * String queue = "testQueue";
+ *
+ * Topology t = ...
+ *
+ * RabbitmqConsumer consumer = new RabbitmqConsumer(t, () -> configMap);
+ * TStream<String> receivedStream = consumer.subscribe((byte[] bytes) -> new String(bytes), queue);
+ *
+ * //...
+ * }
+ * </pre>
+ */
+public class RabbitmqConsumer {
+
+ private final RabbitmqConnector connector;
+ private final Topology topology;
+
+ /**
+ * Create a consumer connector for consuming tuples from a RabbitMQ queue.
+ * <p>
+ * See the RabbitMQ java client document :
+ * <a href="http://www.rabbitmq.com/api-guide.html">http://www.rabbitmq.com/api-guide.html</a>
+ * The full config option please see RabbitMQ java client API Reference :
+ * < a href="https://rabbitmq.github.io/rabbitmq-java-client/api/current/com/rabbitmq/client/ConnectionFactory.html">ConnectionFactory</>
+ * </p>
+ * @param topology topology to add to
+ * @param config RabbitmqProducer configuration information.
+ */
+ public RabbitmqConsumer(Topology topology, Supplier<Map<String, Object>> config) {
+ this.topology = topology;
+ this.connector = new RabbitmqConnector(config);
+ }
+
+ /**
+ * Subscribe to the specified topics and yield a stream of tuples from the published RabbitMQ records.
+ *
+ * @param toTupleFn A function that yields a tuple from a byte array,
+ * @param queue the specified RabbitMQ queue
+ * @param <T> A function that yields a tuple from a
+ * @return stream of tuples
+ */
+ public <T> TStream<T> subscribe(Function<byte[], T> toTupleFn, String queue) {
+ return topology.events(new RabbitmqSubscriber<>(connector, queue, toTupleFn));
+ }
+
+}
diff --git a/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/RabbitmqProducer.java b/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/RabbitmqProducer.java
new file mode 100644
index 0000000..8437285
--- /dev/null
+++ b/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/RabbitmqProducer.java
@@ -0,0 +1,99 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package org.apache.edgent.connectors.rabbitmq;
+
+import org.apache.edgent.connectors.rabbitmq.runtime.RabbitmqConnector;
+import org.apache.edgent.connectors.rabbitmq.runtime.RabbitmqPublisher;
+import org.apache.edgent.function.Function;
+import org.apache.edgent.function.Supplier;
+import org.apache.edgent.topology.TSink;
+import org.apache.edgent.topology.TStream;
+import org.apache.edgent.topology.Topology;
+
+import java.util.Map;
+
+/**
+ * {@code RabbitmqProducer} is a producer to produce messages to a RabbitMQ messaging broker
+ * <p>
+ * The connector uses and includes components from the RabbitMQ 3.7.3 release.
+ * It has been successfully tested against 3.7.3.
+ * For more information about RabbitMQ see <a href="http://www.rabbitmq.com/">http://www.rabbitmq.com/</a>
+ * </p>
+ * Sample use:
+ * <pre>{@code
+ * Map<String, Object> config = new HashMap<>();
+ * config.put(RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_HOST, "127.0.0.1");
+ * config.put(RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_PORT, 5672);
+ * String queue = "testQueue";
+ *
+ * Topology t = newTopology("testSimple");
+ * RabbitmqProducer producer = new RabbitmqProducer(t, () -> config);
+ *
+ * //TStream<String> stream = ...
+ *
+ * TSink<String> sink = producer.publish(stream, queue, (String s) -> s.getBytes());
+ * }
+ * </pre>
+ */
+public class RabbitmqProducer {
+
+ private final RabbitmqConnector connector;
+ private final Topology topology;
+
+ /**
+ * Create a producer connector for publishing tuples to a RabbitMQ queue.
+ * <p>
+ * See the RabbitMQ java client document :
+ * <a href="http://www.rabbitmq.com/api-guide.html">http://www.rabbitmq.com/api-guide.html</a>
+ * The full config option please see RabbitMQ java client API Reference :
+ * < a href="https://rabbitmq.github.io/rabbitmq-java-client/api/current/com/rabbitmq/client/ConnectionFactory.html">ConnectionFactory</>
+ * </p>
+ * @param topology topology to add to
+ * @param config RabbitmqProducer configuration information.
+ */
+ public RabbitmqProducer(Topology topology, Supplier<Map<String, Object>> config) {
+ this.topology = topology;
+ this.connector = new RabbitmqConnector(config);
+ }
+
+ /**
+ * Publish the stream of tuples to the specified queue.
+ * @param stream The stream to publish
+ * @param queue The specified queue of RabbitMQ
+ * @param msgFn A function that yields the byte[] records from the tuple
+ * @param <T> Tuple type
+ * @return {@link TSink}
+ */
+ public <T> TSink<T> publish(TStream<T> stream, String queue, Function<T, byte[]> msgFn) {
+ return stream.sink(new RabbitmqPublisher<>(connector, queue, msgFn));
+ }
+
+ /**
+ * Publish the stream of tuples to the specified queue.
+ * @param stream The stream to publish
+ * @param queue The specified queue of RabbitMQ
+ * @param msg The string message to publish
+ * @return
+ */
+ public TSink<String> publish(TStream<String> stream, String queue, String msg) {
+ return publish(stream, queue, (String m) -> msg.getBytes());
+ }
+
+}
diff --git a/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/package-info.java b/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/package-info.java
new file mode 100644
index 0000000..49dafd2
--- /dev/null
+++ b/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/package-info.java
@@ -0,0 +1,28 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+/**
+ * RabbitMQ stream connector.
+ * <P>
+ * Stream tuples may be published to RabbitMQ queues
+ * and created by subscribing to queues.
+ * For more information about RabbitMQ see
+ * <a href="http://www.rabbitmq.com/">http://www.rabbitmq.com/</a>
+ */
+package org.apache.edgent.connectors.rabbitmq;
\ No newline at end of file
diff --git a/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/runtime/RabbitmqConnector.java b/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/runtime/RabbitmqConnector.java
new file mode 100644
index 0000000..b83cfb6
--- /dev/null
+++ b/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/runtime/RabbitmqConnector.java
@@ -0,0 +1,188 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package org.apache.edgent.connectors.rabbitmq.runtime;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import org.apache.edgent.function.Supplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Map;
+
+import static org.apache.edgent.connectors.rabbitmq.RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_AUTH_NAME;
+import static org.apache.edgent.connectors.rabbitmq.RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_AUTH_PASSWORD;
+import static org.apache.edgent.connectors.rabbitmq.RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_AUTO_RECOVERY;
+import static org.apache.edgent.connectors.rabbitmq.RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_HOST;
+import static org.apache.edgent.connectors.rabbitmq.RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_NETWORK_RECOVERY_INTERVAL;
+import static org.apache.edgent.connectors.rabbitmq.RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_PORT;
+import static org.apache.edgent.connectors.rabbitmq.RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_REQUESTED_CHANNEL_MAX;
+import static org.apache.edgent.connectors.rabbitmq.RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_REQUESTED_FRAME_MAX;
+import static org.apache.edgent.connectors.rabbitmq.RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_REQUESTED_HEARTBEAT;
+import static org.apache.edgent.connectors.rabbitmq.RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_TIMEOUT;
+import static org.apache.edgent.connectors.rabbitmq.RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_TOPOLOGY_RECOVERY_ENABLED;
+import static org.apache.edgent.connectors.rabbitmq.RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_URI;
+import static org.apache.edgent.connectors.rabbitmq.RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_VIRTUAL_HOST;
+
+/**
+ * A connector to an RabbitMQ server.
+ */
+public class RabbitmqConnector implements AutoCloseable {
+
+ private static final long serialVersionUID = 1L;
+ private static final Logger logger = LoggerFactory.getLogger(RabbitmqConnector.class);
+
+ private final Supplier<Map<String, Object>> configFn;
+ private ConnectionFactory connectionFactory;
+ private Connection connection;
+ private Channel channel;
+ private String id;
+
+ public RabbitmqConnector(Supplier<Map<String, Object>> configFn) {
+ this.configFn = configFn;
+ initConnection();
+ }
+
+ public synchronized Channel channel() {
+ if (channel == null) {
+ if (connection != null) {
+ try {
+ channel = connection.createChannel();
+ } catch (IOException e) {
+ logger.error("IOExcetion occurs when create connection channel {}", e);
+ throw new RuntimeException(e);
+ } catch (Exception e) {
+ logger.error("Unknown Exception : {}", e);
+ }
+ } else {
+ logger.error("Inner statue inconformity ： the rabbitmq connection is null.");
+ throw new RuntimeException("Inner statue inconformity ： the rabbitmq connection is null.");
+ }
+ }
+
+ return channel;
+ }
+
+ @Override
+ public synchronized void close() throws Exception {
+ if (channel != null) {
+ channel.close();
+ }
+
+ if (connection != null) {
+ connection.close();
+ }
+
+ if (connectionFactory != null) {
+ connectionFactory = null;
+ }
+ }
+
+ public String id() {
+ if (id == null) {
+ // include our short object Id
+ id = "RabbitMQ " + toString().substring(toString().indexOf('@') + 1);
+ }
+ return id;
+ }
+
+ private void initConnection() {
+ try {
+ this.connectionFactory = getConnectionFactory();
+ this.connection = connectionFactory.newConnection();
+ } catch (Exception e) {
+ logger.error("{}", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ private ConnectionFactory getConnectionFactory() throws Exception {
+ ConnectionFactory connectionFactory = new ConnectionFactory();
+ Map<String, Object> configMap = configFn.get();
+
+ if (configMap.containsKey(RABBITMQ_CONFIG_KEY_URI)) {
+ connectionFactory.setUri(configMap.get(RABBITMQ_CONFIG_KEY_URI).toString());
+ } else {
+ if (!configMap.containsKey(RABBITMQ_CONFIG_KEY_HOST)) {
+ throw new RuntimeException("Missed key : " + RABBITMQ_CONFIG_KEY_HOST);
+ }
+
+ connectionFactory.setHost(configMap.get(RABBITMQ_CONFIG_KEY_HOST).toString());
+
+ if (!configMap.containsKey(RABBITMQ_CONFIG_KEY_PORT)) {
+ throw new RuntimeException("Missed key : " + RABBITMQ_CONFIG_KEY_PORT);
+ }
+
+ connectionFactory.setPort(Integer.valueOf(configMap.get(RABBITMQ_CONFIG_KEY_PORT).toString()));
+ }
+
+ if (configMap.containsKey(RABBITMQ_CONFIG_KEY_VIRTUAL_HOST)) {
+ connectionFactory.setVirtualHost(configMap.get(RABBITMQ_CONFIG_KEY_VIRTUAL_HOST).toString());
+ }
+
+ if (configMap.containsKey(RABBITMQ_CONFIG_KEY_AUTH_NAME)) {
+ connectionFactory.setUsername(configMap.get(RABBITMQ_CONFIG_KEY_AUTH_NAME).toString());
+ }
+
+ if (configMap.containsKey(RABBITMQ_CONFIG_KEY_AUTH_PASSWORD)) {
+ connectionFactory.setPassword(configMap.get(RABBITMQ_CONFIG_KEY_AUTH_PASSWORD).toString());
+ }
+
+ if (configMap.containsKey(RABBITMQ_CONFIG_KEY_AUTO_RECOVERY)) {
+ connectionFactory.setAutomaticRecoveryEnabled(
+ Boolean.valueOf(configMap.get(RABBITMQ_CONFIG_KEY_AUTO_RECOVERY).toString()));
+ }
+
+ if (configMap.containsKey(RABBITMQ_CONFIG_KEY_TIMEOUT)) {
+ connectionFactory.setConnectionTimeout(Integer.valueOf(configMap.get(RABBITMQ_CONFIG_KEY_TIMEOUT).toString()));
+ }
+
+ if (configMap.containsKey(RABBITMQ_CONFIG_KEY_NETWORK_RECOVERY_INTERVAL)) {
+ connectionFactory.setNetworkRecoveryInterval(
+ Integer.valueOf(configMap.get(RABBITMQ_CONFIG_KEY_NETWORK_RECOVERY_INTERVAL).toString()));
+ }
+
+ if (configMap.containsKey(RABBITMQ_CONFIG_KEY_REQUESTED_HEARTBEAT)) {
+ connectionFactory.setRequestedHeartbeat(
+ Integer.valueOf(configMap.get(RABBITMQ_CONFIG_KEY_REQUESTED_HEARTBEAT).toString()));
+ }
+
+ if (configMap.containsKey(RABBITMQ_CONFIG_KEY_TOPOLOGY_RECOVERY_ENABLED)) {
+ connectionFactory.setTopologyRecoveryEnabled(
+ Boolean.valueOf(configMap.get(RABBITMQ_CONFIG_KEY_TOPOLOGY_RECOVERY_ENABLED).toString()));
+ }
+
+ if (configMap.containsKey(RABBITMQ_CONFIG_KEY_REQUESTED_CHANNEL_MAX)) {
+ connectionFactory.setRequestedChannelMax(
+ Integer.parseInt(configMap.get(RABBITMQ_CONFIG_KEY_REQUESTED_CHANNEL_MAX).toString()));
+ }
+
+ if (configMap.containsKey(RABBITMQ_CONFIG_KEY_REQUESTED_FRAME_MAX)) {
+ connectionFactory.setRequestedChannelMax(
+ Integer.valueOf(configMap.get(RABBITMQ_CONFIG_KEY_REQUESTED_FRAME_MAX).toString()));
+ }
+
+ return connectionFactory;
+ }
+
+
+}
diff --git a/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/runtime/RabbitmqPublisher.java b/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/runtime/RabbitmqPublisher.java
new file mode 100644
index 0000000..67c2d45
--- /dev/null
+++ b/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/runtime/RabbitmqPublisher.java
@@ -0,0 +1,74 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package org.apache.edgent.connectors.rabbitmq.runtime;
+
+import org.apache.edgent.function.Consumer;
+import org.apache.edgent.function.Function;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * a publisher for RabbitMQ connector
+ */
+public class RabbitmqPublisher<T> implements Consumer<T>, AutoCloseable {
+
+ private static final long serialVersionUID = 1L;
+ private static final Logger logger = LoggerFactory.getLogger(RabbitmqPublisher.class);
+
+ private final RabbitmqConnector connector;
+ private final Function<T, byte[]> msgFn;
+ private final String queue;
+ private String id;
+
+ public RabbitmqPublisher(RabbitmqConnector connector, String queue, Function<T, byte[]> msgFn) {
+ this.connector = connector;
+ this.queue = queue;
+ this.msgFn = msgFn;
+ }
+
+ @Override
+ public synchronized void close() throws Exception {
+ logger.info("{} is closing.", id());
+ connector.close();
+ logger.info("{} is closed.", id());
+ }
+
+ @Override
+ public void accept(T value) {
+ byte[] msg = msgFn.apply(value);
+ try {
+ connector.channel().basicPublish("", queue, null, msg);
+ } catch (IOException e) {
+ logger.error("publish exception : {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ public String id() {
+ if (id == null) {
+ // include our short object Id
+ id = connector.id() + " PUB " + toString().substring(toString().indexOf('@') + 1);
+ }
+
+ return id;
+ }
+}
diff --git a/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/runtime/RabbitmqSubscriber.java b/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/runtime/RabbitmqSubscriber.java
new file mode 100644
index 0000000..40c2d45
--- /dev/null
+++ b/connectors/rabbitmq/src/main/java/org/apache/edgent/connectors/rabbitmq/runtime/RabbitmqSubscriber.java
@@ -0,0 +1,111 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package org.apache.edgent.connectors.rabbitmq.runtime;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.client.Envelope;
+import org.apache.edgent.function.Consumer;
+import org.apache.edgent.function.Function;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+/**
+ * a subscriber for RabbitMQ connector
+ */
+public class RabbitmqSubscriber<T> implements Consumer<Consumer<T>>, AutoCloseable {
+
+ private static final long serialVersionUID = 1L;
+ private static final Logger logger = LoggerFactory.getLogger(RabbitmqSubscriber.class);
+
+ private final RabbitmqConnector connector;
+ private Function<byte[], T> toTupleFn;
+ private Consumer<T> eventSubmitter;
+ private ExecutorService executor;
+ private String queue;
+ private String id;
+
+ public RabbitmqSubscriber(RabbitmqConnector connector, String queue, Function<byte[], T> toTupleFn) {
+ this.connector = connector;
+ this.queue = queue;
+ this.toTupleFn = toTupleFn;
+ }
+
+ @Override
+ public synchronized void close() throws Exception {
+ logger.info("{} is closing.", id());
+ if (executor != null && !executor.isShutdown()) {
+ executor.shutdownNow();
+ }
+
+ connector.close();
+ logger.info("{} is closed", id());
+ }
+
+ @Override
+ public void accept(Consumer<T> eventSubmitter) {
+ this.eventSubmitter = eventSubmitter;
+
+ executor = Executors.newFixedThreadPool(1);
+
+ executor.submit(() -> {
+ boolean autoAck = false;
+ try {
+ connector.channel().basicConsume(queue, autoAck,
+ new DefaultConsumer(connector.channel()) {
+ @Override
+ public void handleDelivery(String consumerTag,
+ Envelope envelope,
+ AMQP.BasicProperties properties,
+ byte[] body)
+ throws IOException {
+ long deliveryTag = envelope.getDeliveryTag();
+
+ acceptCallback(body);
+
+ connector.channel().basicAck(deliveryTag, false);
+ }
+ });
+ } catch (IOException e) {
+ logger.error("Consumer exception : {}", e);
+ }
+ });
+
+ }
+
+ private void acceptCallback(byte[] msg) {
+ T tuple = toTupleFn.apply(msg);
+ eventSubmitter.accept(tuple);
+ }
+
+ public String id() {
+ if (id == null) {
+ // include our short object Id
+ id = connector.id() + " SUB " + toString().substring(toString().indexOf('@') + 1);
+ }
+
+ return id;
+ }
+
+}
diff --git a/connectors/rabbitmq/src/test/java/org/apache/edgent/test/connectors/rabbitmq/RabbitmqStreamsGlobalTestManual.java b/connectors/rabbitmq/src/test/java/org/apache/edgent/test/connectors/rabbitmq/RabbitmqStreamsGlobalTestManual.java
new file mode 100644
index 0000000..9565bf9
--- /dev/null
+++ b/connectors/rabbitmq/src/test/java/org/apache/edgent/test/connectors/rabbitmq/RabbitmqStreamsGlobalTestManual.java
@@ -0,0 +1,38 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package org.apache.edgent.test.connectors.rabbitmq;
+
+/**
+ * RabbitmqStreams connector globalization tests.
+ */
+public class RabbitmqStreamsGlobalTestManual extends RabbitmqStreamsTestManual {
+
+ private static final String globalMsg1 = "你好";
+ private static final String globalMsg2 = "你在嗎";
+
+ public String getMsg1() {
+ return globalMsg1;
+ }
+
+ public String getMsg2() {
+ return globalMsg2;
+ }
+
+}
diff --git a/connectors/rabbitmq/src/test/java/org/apache/edgent/test/connectors/rabbitmq/RabbitmqStreamsTestManual.java b/connectors/rabbitmq/src/test/java/org/apache/edgent/test/connectors/rabbitmq/RabbitmqStreamsTestManual.java
new file mode 100644
index 0000000..02ea0bc
--- /dev/null
+++ b/connectors/rabbitmq/src/test/java/org/apache/edgent/test/connectors/rabbitmq/RabbitmqStreamsTestManual.java
@@ -0,0 +1,118 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package org.apache.edgent.test.connectors.rabbitmq;
+
+import org.apache.edgent.connectors.rabbitmq.RabbitmqConfigKeyConstants;
+import org.apache.edgent.connectors.rabbitmq.RabbitmqConsumer;
+import org.apache.edgent.connectors.rabbitmq.RabbitmqProducer;
+import org.apache.edgent.test.connectors.common.ConnectorTestBase;
+import org.apache.edgent.topology.TSink;
+import org.apache.edgent.topology.TStream;
+import org.apache.edgent.topology.Topology;
+import org.apache.edgent.topology.plumbing.PlumbingStreams;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertNotNull;
+
+/**
+ * A {@link RabbitmqConsumer} manual test case.
+ * Please follow there steps
+ *
+ * step 1 :
+ * Install RabbitMQ server.
+ * For Mac os x, you can use homebrew to install it, the simple command is : `brew install rabbitmq`
+ * For other system, please follow RabbitMQ's offical document :
+ * <a href="http://www.rabbitmq.com/download.html">http://www.rabbitmq.com/download.html</a>
+ *
+ * step 2 :
+ * Start the RabbitMQ server.
+ * For Mac os x, if you install it with homebrew, you can start it with : `brew services start rabbitmq`
+ * For other system, please follow RabbitMQ's offical document :
+ * <a href="http://www.rabbitmq.com/download.html">http://www.rabbitmq.com/download.html</a>
+ * Note : the default port which RabbitMQ server listen is 5672
+ * and it will start a web console which listen port 15672
+ * so you can access it with your web browser, just enter : http://localhost:15672/
+ * you will see a authorization page, the default user name and password are both `guest`
+ *
+ * step 3 :
+ * Create a test queue use the web console.
+ * after starting the server, you can access `http://localhost:15672/#/queues`
+ * then you will find a button named `Add a new queue`, click it and input `testQueue` into `name` field
+ * and click `Add queue` button to create the queue.
+ *
+ * step 4 :
+ * run this test case to test the publish and subscribe method.
+ */
+public class RabbitmqStreamsTestManual extends ConnectorTestBase {
+
+ private static final int PUB_DELAY_MSEC = 15*1000; // have seen 12sec 1st test's consumer startup delay
+ private static final int SEC_TIMEOUT = 20;
+
+ private final String msg1 = "Hello";
+ private final String msg2 = "Are you there?";
+
+ public String getMsg1() {
+ return msg1;
+ }
+
+ public String getMsg2() {
+ return msg2;
+ }
+
+ private Map<String, Object> initRabbitmqConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put(RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_HOST, "127.0.0.1");
+ config.put(RabbitmqConfigKeyConstants.RABBITMQ_CONFIG_KEY_PORT, 5672);
+
+ return config;
+ }
+
+ @Test
+ public void testSimple() throws Exception {
+ Topology t = newTopology("testSimple");
+ Map<String, Object> configMap = initRabbitmqConfig();
+ MsgGenerator generator = new MsgGenerator(t.getName());
+ String queue = "testQueue";
+
+ List<String> msgs = createMsgs(generator, queue, getMsg1(), getMsg2());
+
+ TStream<String> stream = PlumbingStreams.blockingOneShotDelay(
+ t.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
+
+ RabbitmqConsumer consumer = new RabbitmqConsumer(t, () -> configMap);
+
+ TStream<String> receivedStream = consumer.subscribe((byte[] bytes) -> new String(bytes), queue);
+
+ RabbitmqProducer producer = new RabbitmqProducer(t, () -> configMap);
+
+ TSink<String> sink = producer.publish(stream, queue, (String s) -> s.getBytes());
+
+ completeAndValidate("", t, receivedStream, generator, SEC_TIMEOUT, msgs.toArray(new String[0]));
+
+ assertNotNull(sink);
+ }
+
+
+}
