Add a quick start which contain some examples.

README.md

@@ -7,7 +7,7 @@
 [![Twitter Follow](https://img.shields.io/twitter/follow/ApacheRocketMQ?style=social)](https://twitter.com/intent/follow?screen_name=ApacheRocketMQ)
 
 ## [中文文档](./README-Chinese.md)
-## [Quick Start](./QUICKSTART.md)
+## [Quick Start](./quick_start.md)
 
 
 ## Features

pom.xml

@@ -69,7 +69,7 @@
         <commons-logging.version>1.1</commons-logging.version>
         <spring.version>3.2.13.RELEASE</spring.version>
         <auto-service.version>1.0-rc5</auto-service.version>
-        <mysql-connector.version>5.1.40</mysql-connector.version>
+        <mysql-connector.version>8.0.26</mysql-connector.version>
         <fastjson.version>1.2.78</fastjson.version>
         <quartz.version>2.2.1</quartz.version>
         <httpclient.version>4.5.2</httpclient.version>
@@ -112,13 +112,13 @@
                         <exclude>.asf.yaml</exclude>
                         <exclude>README.md</exclude>
                         <exclude>README-Chinese.md</exclude>
-                        <exclude>QUICKSTART.md</exclude>
+                        <exclude>quick_start.md</exclude>
                         <exclude>.github/**</exclude>
                         <exclude>*/target/**</exclude>
                         <exclude>*/*.iml</exclude>
                         <exclude>**/*.txt</exclude>
                         <exclude>**/*.cs</exclude>
-                        <exclude>src/test/resources/window_msg_*</exclude>
+                        <exclude>**/*.sql</exclude>
                     </excludes>
                 </configuration>
             </plugin>
@@ -283,6 +283,12 @@
                 <artifactId>rocketmq-streams-channel-rocketmq</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.apache.rocketmq</groupId>
+                <artifactId>rocketmq-streams-dbinit</artifactId>
+                <version>${project.version}</version>
+            </dependency>
+
 
             <!-- ================================================= -->
             <!-- rocketmq library -->

quick_start.md

@@ -0,0 +1,71 @@
+## rocketmq-streams 快速搭建
+---
+### 前言
+本文档主要介绍如何基于rocketmq-streams快速搭建流处理任务，搭建过程中某些例子会用到rocketmq，可以参考[rocketmq搭建文档](https://rocketmq.apache.org/docs/quick-start/)
+
+
+### 1、源码构建
+
+#### 1.1、构建环境
+ - JDK 1.8 and above
+ - Maven 3.2 and above
+
+#### 1.2、构建Rocketmq-streams
+
+`git clone https://github.com/apache/rocketmq-streams.git`
+`cd rocketmq-streams`
+`mvn clean -DskipTests  install -U`
+
+
+### 2、基于rocketmq-streams创建应用
+
+#### 2.1、pom依赖
+```xml
+<dependency>
+    <groupId>org.apache.rocketmq</groupId>
+    <artifactId>rocketmq-streams-clients</artifactId>
+</dependency>
+```
+#### 2.2、shade clients依赖包
+```xml
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <version>3.2.1</version>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                        <configuration>
+                            <minimizeJar>false</minimizeJar>
+                            <shadedArtifactAttached>true</shadedArtifactAttached>
+                            <artifactSet>
+                                <includes>
+                                    <include>org.apache.rocketmq:rocketmq-streams-clients</include>
+                                </includes>
+                            </artifactSet>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+```
+
+#### 2.3、编写业务代码
+Please see the [rocketmq-streams-examples](rocketmq-streams-examples/README.md)
+#### 2.4、运行
+- 前提：在从rocketmq中读取数据做流处理时，需要运行topic在rocketmq中自动创建，因为做groupBy操作时，需要用到rocketmq作为shuffle数据的读写目的地。
+- 命令：
+```
+   java -jar XXXX-shade.jar \ 
+        -Dlog4j.level=ERROR \
+        -Dlog4j.home=/logs  \
+        -Xms1024m \
+        -Xmx1024m 
+```
+

rocketmq-streams-channel-rocketmq/src/main/java/org/apache/rocketmq/streams/sink/RocketMQSink.java

@@ -40,6 +40,7 @@
 import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
 import org.apache.rocketmq.tools.command.CommandUtil;
 
+
 public class RocketMQSink extends AbstractSupportShuffleSink {
 
     private static final Log LOG = LogFactory.getLog(RocketMQSink.class);

rocketmq-streams-channel-rocketmq/src/main/java/org/apache/rocketmq/streams/source/RocketMQSource.java

@@ -41,12 +41,7 @@
 import org.apache.rocketmq.streams.common.channel.source.AbstractSupportOffsetResetSource;
 import org.apache.rocketmq.streams.common.channel.split.ISplit;
 import org.apache.rocketmq.streams.common.configurable.annotation.ENVDependence;
-import org.apache.rocketmq.streams.common.context.AbstractContext;
-import org.apache.rocketmq.streams.common.context.IMessage;
-import org.apache.rocketmq.streams.common.interfaces.IStreamOperator;
-import org.apache.rocketmq.streams.common.utils.DateUtil;
 import org.apache.rocketmq.streams.common.utils.ReflectUtil;
-import org.apache.rocketmq.streams.common.utils.RuntimeUtil;
 import org.apache.rocketmq.streams.debug.DebugWriter;
 import org.apache.rocketmq.streams.queue.RocketMQMessageQueue;
 import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
@@ -60,7 +55,6 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
 public class RocketMQSource extends AbstractSupportOffsetResetSource {
@@ -134,6 +128,8 @@ protected DefaultMQPushConsumer startConsumer() {
                     for (MessageExt msg : msgs) {
                         String data = new String(msg.getBody(), CHARSET);
                         JSONObject jsonObject = create(data);
+
+
                         String queueId = RocketMQMessageQueue.getQueueId(context.getMessageQueue());
                         String offset = msg.getQueueOffset() + "";
                         org.apache.rocketmq.streams.common.context.Message message = createMessage(jsonObject, queueId, offset, false);

rocketmq-streams-clients/pom.xml

@@ -52,6 +52,10 @@
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-streams-window</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-streams-dbinit</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>

rocketmq-streams-clients/src/main/java/org/apache/rocketmq/streams/client/DataStreamAction.java

@@ -18,11 +18,6 @@
 package org.apache.rocketmq.streams.client;
 
 import com.google.common.collect.Maps;
-
-import java.util.Map;
-import java.util.Properties;
-import java.util.Set;
-
 import org.apache.rocketmq.streams.client.strategy.Strategy;
 import org.apache.rocketmq.streams.client.transform.DataStream;
 import org.apache.rocketmq.streams.common.component.ComponentCreator;
@@ -32,6 +27,10 @@
 import org.apache.rocketmq.streams.common.topology.builder.PipelineBuilder;
 import org.apache.rocketmq.streams.configurable.ConfigurableComponent;
 
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
 public class DataStreamAction extends DataStream {
 
     private final Map<String, Object> properties = Maps.newHashMap();

rocketmq-streams-clients/src/test/java/org/apache/rocketmq/streams/client/DataStreamTest.java

@@ -28,7 +28,11 @@
 import org.junit.Test;
 
 import java.io.Serializable;
-import java.sql.*;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
 
 public class DataStreamTest implements Serializable {
 

rocketmq-streams-clients/src/test/java/org/apache/rocketmq/streams/client/windows/AbstractWindowTest.java

@@ -18,6 +18,16 @@
 package org.apache.rocketmq.streams.client.windows;
 
 import com.alibaba.fastjson.JSONObject;
+import org.apache.rocketmq.streams.client.transform.DataStream;
+import org.apache.rocketmq.streams.client.transform.window.Time;
+import org.apache.rocketmq.streams.client.transform.window.TumblingWindow;
+import org.apache.rocketmq.streams.common.functions.ForEachFunction;
+import org.apache.rocketmq.streams.common.functions.MapFunction;
+import org.apache.rocketmq.streams.common.utils.DateUtil;
+import org.apache.rocketmq.streams.common.utils.FileUtil;
+import org.apache.rocketmq.streams.common.utils.MapKeyUtil;
+import org.apache.rocketmq.streams.common.utils.StringUtil;
+
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -28,15 +38,6 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
-import org.apache.rocketmq.streams.client.transform.DataStream;
-import org.apache.rocketmq.streams.client.transform.window.Time;
-import org.apache.rocketmq.streams.client.transform.window.TumblingWindow;
-import org.apache.rocketmq.streams.common.functions.ForEachFunction;
-import org.apache.rocketmq.streams.common.functions.MapFunction;
-import org.apache.rocketmq.streams.common.utils.DateUtil;
-import org.apache.rocketmq.streams.common.utils.FileUtil;
-import org.apache.rocketmq.streams.common.utils.MapKeyUtil;
-import org.apache.rocketmq.streams.common.utils.StringUtil;
 
 import static junit.framework.TestCase.assertTrue;
 

rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/impl/OutputPrintChannel.java

@@ -16,11 +16,10 @@
  */
 package org.apache.rocketmq.streams.common.channel.impl;
 
-import java.util.List;
-
 import org.apache.rocketmq.streams.common.channel.sink.AbstractSink;
 import org.apache.rocketmq.streams.common.context.IMessage;
-import org.apache.rocketmq.streams.common.utils.PrintUtil;
+
+import java.util.List;
 
 /**
  * 测试使用，输出就是把消息打印出来
@@ -30,7 +29,7 @@ public class OutputPrintChannel extends AbstractSink {
     @Override
     protected boolean batchInsert(List<IMessage> messages) {
         for (IMessage msg : messages) {
-            //System.out.println(msg.getMessageValue());
+            System.out.println(msg.getMessageValue());
         }
         return false;
     }

rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/impl/file/FileSource.java

@@ -21,6 +21,7 @@
 import java.io.FileNotFoundException;
 import java.io.FileReader;
 import java.io.IOException;
+import java.net.URL;
 import java.util.Iterator;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
@@ -55,7 +56,7 @@ public FileSource(String filePath) {
     @Override
     protected boolean initConfigurable() {
         super.initConfigurable();
-        File file = new File(filePath);
+        File file = getFile(filePath);
         if (file.exists() && file.isDirectory()) {
             executorService = new ThreadPoolExecutor(maxThread, maxThread,
                 0L, TimeUnit.MILLISECONDS,
@@ -64,6 +65,23 @@ protected boolean initConfigurable() {
         return true;
     }
 
+    private File getFile(String filePath) {
+        File file = new File(filePath);
+        if (!file.exists()) {
+            ClassLoader loader = getClass().getClassLoader();
+            URL url = loader.getResource(filePath);
+
+            if (url != null) {
+                String path = url.getFile();
+                file = new File(path);
+                this.filePath = path;
+            }
+        }
+        return file;
+
+
+    }
+
     @Override
     protected boolean startSource() {
 
@@ -97,7 +115,7 @@ protected boolean startSource() {
      */
     protected LinkedBlockingQueue<FileIterator> createIteratorList() {
         LinkedBlockingQueue<FileIterator> iterators = new LinkedBlockingQueue<>(1000);
-        File file = new File(filePath);
+        File file = getFile(filePath);
         if (file.exists() == false) {
             return null;
         }