You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by zh...@apache.org on 2022/03/01 13:45:13 UTC
[rocketmq-connect] 02/12: init complete
This is an automated email from the ASF dual-hosted git repository.
zhoubo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-connect.git
commit c7c0c04a7fd28d1067ec2c8700d404ac8dd2f391
Author: laohu <23...@qq.com>
AuthorDate: Sun Jun 2 21:10:20 2019 +0800
init complete
---
pom.xml | 259 +++++++++++++++++++++
.../java/io/openmessaging/activemq/Config.java | 133 +++++++++++
.../java/io/openmessaging/activemq/Replicator.java | 72 ++++++
.../activemq/connector/ActivemqConnector.java | 72 ++++++
.../activemq/connector/ActivemqTask.java | 87 +++++++
.../activemq/pattern/PatternProcessor.java | 83 +++++++
6 files changed, 706 insertions(+)
diff --git a/pom.xml b/pom.xml
new file mode 100644
index 0000000..b021330
--- /dev/null
+++ b/pom.xml
@@ -0,0 +1,259 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!-- 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. -->
+
+<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">
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-connect-activemq</artifactId>
+ <version>1.0.0</version>
+
+ <name>connect-activemq</name>
+ <description>Redis Replicator is a redis RDB and Command parser written in java.
+ It can parse,filter,broadcast the RDB and Command events in a real time manner
+ and resent to Apache RocketMQ, then consumer could subscribe topic to receive data.
+ </description>
+ <url>https://github.com/apache/incubator-rocketmq-externals/tree/master/rocketmq-connect-activemq</url>
+
+ <licenses>
+ <license>
+ <name>The Apache Software License, Version 2.0</name>
+ <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+ </license>
+ </licenses>
+
+ <developers>
+ <developer>
+ <name>Leon Chen</name>
+ <email>chen.bao.yi@gmail.com</email>
+ <organization>moilioncircle</organization>
+ <organizationUrl>http://www.moilioncircle.com/</organizationUrl>
+ <roles>
+ <role>Developer</role>
+ </roles>
+ <timezone>+8</timezone>
+ </developer>
+
+ <developer>
+ <name>Adrian Yao</name>
+ <email>adrianyaofly@gmail.com</email>
+ <organization>unstudy</organization>
+ <timezone>+8</timezone>
+ </developer>
+
+ <developer>
+ <name>Rick Zhang</name>
+ <email>zhangke.huangshan@gmail.com</email>
+ <organization>treefinance.com.cn</organization>
+ <roles>
+ <role>Developer</role>
+ </roles>
+ <timezone>+8</timezone>
+ </developer>
+ </developers>
+
+ <issueManagement>
+ <system>jira</system>
+ <url>https://issues.apache.org/jira/browse/RocketMQ</url>
+ </issueManagement>
+
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+
+ <!-- Compiler settings properties -->
+ <maven.compiler.source>1.8</maven.compiler.source>
+ <maven.compiler.target>1.8</maven.compiler.target>
+ </properties>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>versions-maven-plugin</artifactId>
+ <version>2.3</version>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>clirr-maven-plugin</artifactId>
+ <version>2.7</version>
+ </plugin>
+ <plugin>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.6.1</version>
+ <configuration>
+ <source>${maven.compiler.source}</source>
+ <target>${maven.compiler.target}</target>
+ <compilerVersion>${maven.compiler.source}</compilerVersion>
+ <showDeprecation>true</showDeprecation>
+ <showWarnings>true</showWarnings>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>2.19.1</version>
+ <configuration>
+ <argLine>-Xms512m -Xmx1024m</argLine>
+ <forkMode>always</forkMode>
+ <includes>
+ <include>**/*Test.java</include>
+ </includes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-site-plugin</artifactId>
+ <version>3.6</version>
+ <configuration>
+ <locales>en_US</locales>
+ <outputEncoding>UTF-8</outputEncoding>
+ <inputEncoding>UTF-8</inputEncoding>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-source-plugin</artifactId>
+ <version>3.0.1</version>
+ <executions>
+ <execution>
+ <id>attach-sources</id>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-javadoc-plugin</artifactId>
+ <version>2.10.4</version>
+ <configuration>
+ <charset>UTF-8</charset>
+ <locale>en_US</locale>
+ <excludePackageNames>io.openmessaging.internal</excludePackageNames>
+ </configuration>
+ <executions>
+ <execution>
+ <id>aggregate</id>
+ <goals>
+ <goal>aggregate</goal>
+ </goals>
+ <phase>site</phase>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-resources-plugin</artifactId>
+ <version>3.0.2</version>
+ <configuration>
+ <encoding>${project.build.sourceEncoding}</encoding>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>findbugs-maven-plugin</artifactId>
+ <version>3.0.4</version>
+ </plugin>
+ </plugins>
+ </build>
+
+ <dependencies>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>4.11</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <version>2.6.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <version>2.6.3</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>4.5.5</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.openmessaging</groupId>
+ <artifactId>openmessaging-connect-runtime</artifactId>
+ <version>0.0.1-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>io.openmessaging</groupId>
+ <artifactId>openmessaging-connector</artifactId>
+ <version>0.1.0-beta</version>
+ </dependency>
+ <dependency>
+ <groupId>io.openmessaging</groupId>
+ <artifactId>openmessaging-api</artifactId>
+ <version>0.3.1-alpha</version>
+ </dependency>
+ <dependency>
+ <groupId>com.alibaba</groupId>
+ <artifactId>fastjson</artifactId>
+ <version>1.2.51</version>
+ </dependency>
+ <!-- <dependency> <groupId>io.javalin</groupId> <artifactId>javalin</artifactId>
+ <version>1.3.0</version> </dependency> -->
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <version>1.7.7</version>
+ </dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ <version>1.0.13</version>
+ </dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-core</artifactId>
+ <version>1.0.13</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-openmessaging</artifactId>
+ <version>4.3.2</version>
+ </dependency>
+ <dependency>
+ <groupId>commons-cli</groupId>
+ <artifactId>commons-cli</artifactId>
+ <version>1.2</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.activemq</groupId>
+ <artifactId>activemq-all</artifactId>
+ <version>5.9.0</version>
+ </dependency>
+ <dependency>
+ <groupId>javax.jms</groupId>
+ <artifactId>javax.jms-api</artifactId>
+ <version>2.0</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.glassfish.main.javaee-api</groupId>
+ <artifactId>javax.jms</artifactId>
+ <version>3.1.2.2</version>
+ </dependency>
+
+ </dependencies>
+
+</project>
diff --git a/src/main/java/io/openmessaging/activemq/Config.java b/src/main/java/io/openmessaging/activemq/Config.java
new file mode 100644
index 0000000..10a5d9f
--- /dev/null
+++ b/src/main/java/io/openmessaging/activemq/Config.java
@@ -0,0 +1,133 @@
+/*
+ * 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 io.openmessaging.activemq;
+
+import java.lang.reflect.Method;
+import java.util.HashSet;
+import java.util.Set;
+
+import io.openmessaging.KeyValue;
+
+public class Config {
+
+ public static final Set<String> REQUEST_CONFIG = new HashSet<String>() {
+ {
+ add("activemqUrl");
+ add("activemqUsername");
+ add("activemqPassword");
+ add("destinationType");
+ add("destinationName");
+ }
+ };
+
+ public String activemqUrl;
+
+ public String activemqUsername;
+
+ public String activemqPassword;
+
+ public String destinationType;
+
+ public String destinationName;
+
+ public void load(KeyValue props) {
+
+ properties2Object(props, this);
+ }
+
+ private void properties2Object(final KeyValue p, final Object object) {
+
+ Method[] methods = object.getClass().getMethods();
+ for (Method method : methods) {
+ String mn = method.getName();
+ if (mn.startsWith("set")) {
+ try {
+ String tmp = mn.substring(4);
+ String first = mn.substring(3, 4);
+
+ String key = first.toLowerCase() + tmp;
+ String property = p.getString(key);
+ if (property != null) {
+ Class<?>[] pt = method.getParameterTypes();
+ if (pt != null && pt.length > 0) {
+ String cn = pt[0].getSimpleName();
+ Object arg;
+ if (cn.equals("int") || cn.equals("Integer")) {
+ arg = Integer.parseInt(property);
+ } else if (cn.equals("long") || cn.equals("Long")) {
+ arg = Long.parseLong(property);
+ } else if (cn.equals("double") || cn.equals("Double")) {
+ arg = Double.parseDouble(property);
+ } else if (cn.equals("boolean") || cn.equals("Boolean")) {
+ arg = Boolean.parseBoolean(property);
+ } else if (cn.equals("float") || cn.equals("Float")) {
+ arg = Float.parseFloat(property);
+ } else if (cn.equals("String")) {
+ arg = property;
+ } else {
+ continue;
+ }
+ method.invoke(object, arg);
+ }
+ }
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+ }
+
+ public String getActivemqUrl() {
+ return activemqUrl;
+ }
+
+ public void setActivemqUrl(String activemqUrl) {
+ this.activemqUrl = activemqUrl;
+ }
+
+ public String getActivemqUsername() {
+ return activemqUsername;
+ }
+
+ public void setActivemqUsername(String activemqUsername) {
+ this.activemqUsername = activemqUsername;
+ }
+
+ public String getActivemqPassword() {
+ return activemqPassword;
+ }
+
+ public void setActivemqPassword(String activemqPassword) {
+ this.activemqPassword = activemqPassword;
+ }
+
+ public String getDestinationType() {
+ return destinationType;
+ }
+
+ public void setDestinationType(String destinationType) {
+ this.destinationType = destinationType;
+ }
+
+ public String getDestinationName() {
+ return destinationName;
+ }
+
+ public void setDestinationName(String destinationName) {
+ this.destinationName = destinationName;
+ }
+}
\ No newline at end of file
diff --git a/src/main/java/io/openmessaging/activemq/Replicator.java b/src/main/java/io/openmessaging/activemq/Replicator.java
new file mode 100644
index 0000000..51ca6c1
--- /dev/null
+++ b/src/main/java/io/openmessaging/activemq/Replicator.java
@@ -0,0 +1,72 @@
+/*
+ * 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 io.openmessaging.activemq;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import javax.jms.Message;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.openmessaging.activemq.pattern.PatternProcessor;
+
+public class Replicator {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(Replicator.class);
+
+ private static final Logger POSITION_LOGGER = LoggerFactory.getLogger("PositionLogger");
+
+ private PatternProcessor processor;
+
+ private Config config;
+ private Object lock = new Object();
+ private BlockingQueue<Message> queue = new LinkedBlockingQueue<>();
+
+ public Replicator(Config config){
+ this.config = config;
+ }
+
+ public void start() {
+
+ try {
+ processor = new PatternProcessor(this);
+ processor.start();
+
+ } catch (Exception e) {
+ LOGGER.error("Start error.", e);
+ }
+ }
+
+ public void stop(){
+ processor.stop();
+ }
+
+ public void commit(Message message, boolean isComplete) {
+ queue.add(message);
+ }
+
+ public Config getConfig() {
+ return this.config;
+ }
+
+ public BlockingQueue<Message> getQueue() {
+ return queue;
+ }
+}
diff --git a/src/main/java/io/openmessaging/activemq/connector/ActivemqConnector.java b/src/main/java/io/openmessaging/activemq/connector/ActivemqConnector.java
new file mode 100644
index 0000000..1c9a530
--- /dev/null
+++ b/src/main/java/io/openmessaging/activemq/connector/ActivemqConnector.java
@@ -0,0 +1,72 @@
+/*
+ * 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 io.openmessaging.activemq.connector;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.activemq.Config;
+import io.openmessaging.connector.api.Task;
+import io.openmessaging.connector.api.source.SourceConnector;
+import java.util.ArrayList;
+import java.util.List;
+
+public class ActivemqConnector extends SourceConnector {
+
+ private KeyValue config;
+
+ @Override
+ public String verifyAndSetConfig(KeyValue config) {
+
+ for(String requestKey : Config.REQUEST_CONFIG){
+ if(!config.containsKey(requestKey)){
+ return "Request config key: " + requestKey;
+ }
+ }
+ this.config = config;
+ return "";
+ }
+
+ @Override
+ public void start() {
+
+ }
+
+ @Override
+ public void stop() {
+
+ }
+
+ @Override public void pause() {
+
+ }
+
+ @Override public void resume() {
+
+ }
+
+ @Override
+ public Class<? extends Task> taskClass() {
+ return ActivemqTask.class;
+ }
+
+ @Override
+ public List<KeyValue> taskConfigs() {
+ List<KeyValue> config = new ArrayList<>();
+ config.add(this.config);
+ return config;
+ }
+}
diff --git a/src/main/java/io/openmessaging/activemq/connector/ActivemqTask.java b/src/main/java/io/openmessaging/activemq/connector/ActivemqTask.java
new file mode 100644
index 0000000..a04fc50
--- /dev/null
+++ b/src/main/java/io/openmessaging/activemq/connector/ActivemqTask.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 io.openmessaging.activemq.connector;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import javax.jms.Message;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.fastjson.JSON;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.activemq.Config;
+import io.openmessaging.activemq.Replicator;
+import io.openmessaging.connector.api.data.SourceDataEntry;
+import io.openmessaging.connector.api.source.SourceTask;
+
+public class ActivemqTask extends SourceTask {
+
+ private static final Logger log = LoggerFactory.getLogger(ActivemqTask.class);
+
+ private Replicator replicator;
+
+ private Config config;
+
+ @Override
+ public Collection<SourceDataEntry> poll() {
+
+ List<SourceDataEntry> res = new ArrayList<>();
+
+ try {
+ Message message = replicator.getQueue().poll(1000, TimeUnit.MILLISECONDS);
+ SourceDataEntry sourceDataEntry = null;
+
+ res.add(sourceDataEntry);
+ } catch (Exception e) {
+ log.error("Mysql task poll error, current config:" + JSON.toJSONString(config), e);
+ }
+ return res;
+ }
+
+ @Override
+ public void start(KeyValue props) {
+
+ try {
+ this.config = new Config();
+ this.config.load(props);
+ this.replicator = new Replicator(config);
+ } catch (Exception e) {
+ log.error("Mysql task start failed.", e);
+ }
+ this.replicator.start();
+ }
+
+ @Override
+ public void stop() {
+ replicator.stop();
+ }
+
+ @Override public void pause() {
+
+ }
+
+ @Override public void resume() {
+
+ }
+}
diff --git a/src/main/java/io/openmessaging/activemq/pattern/PatternProcessor.java b/src/main/java/io/openmessaging/activemq/pattern/PatternProcessor.java
new file mode 100644
index 0000000..4b4b450
--- /dev/null
+++ b/src/main/java/io/openmessaging/activemq/pattern/PatternProcessor.java
@@ -0,0 +1,83 @@
+package io.openmessaging.activemq.pattern;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.Session;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.commons.lang3.StringUtils;
+
+import io.openmessaging.activemq.Config;
+import io.openmessaging.activemq.Replicator;
+
+public class PatternProcessor {
+
+ private Replicator replicator;
+
+ private Config config;
+
+ Connection connection;
+
+ Session session;
+
+ MessageConsumer consumer;
+
+ public PatternProcessor(Replicator replicator) {
+ this.replicator = replicator;
+ this.config = replicator.getConfig();
+ }
+
+ public void start() {
+ try {
+ ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(config.getActivemqUrl());
+
+ //2、使用连接工厂创建一个连接对象
+ if(StringUtils.isNotBlank(config.getActivemqUsername()) && StringUtils.isNotBlank(config.getActivemqPassword()) ) {
+ connection = connectionFactory.createConnection(config.getActivemqUsername() , config.getActivemqPassword());
+ }else {
+ connection = connectionFactory.createConnection();
+ }
+ //3、开启连接
+ connection.start();
+ //4、使用连接对象创建会话（session）对象
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ //5、使用会话对象创建目标对象，包含queue和topic（一对一和一对多）
+ Destination destination = null;
+ if(StringUtils.equals("topic", config.getDestinationType())) {
+ destination = session.createTopic(config.getDestinationName());
+ }else if(StringUtils.equals("queue", config.getDestinationType())){
+ destination = session.createQueue(config.getDestinationName());
+ }else {
+ throw new RuntimeException("");
+ }
+ consumer = session.createConsumer(destination);
+ //6、使用会话对象创建生产者对象
+ //7、向consumer对象中设置一个messageListener对象，用来接收消息
+ consumer.setMessageListener(new MessageListener() {
+ @Override
+ public void onMessage(Message message) {
+ replicator.commit(message, true);
+ }
+ });
+ }catch(Exception e) {
+
+ }
+ }
+
+ public void stop() {
+ try {
+ consumer.close();
+ session.close();
+ connection.close();
+ } catch (JMSException e) {
+ e.printStackTrace();
+ }
+ }
+
+
+}