[ISSUE-48] Support for Spring Cloud Stream (#318)

* Support for Spring Cloud Stream (#48)

joyqueue-client/joyqueue-client-samples/joyqueue-client-samples-springcloud-stream/pom.xml

@@ -0,0 +1,40 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Copyright 2019 The JoyQueue Authors.
+
+    Licensed 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">
+    <parent>
+        <artifactId>joyqueue-client-samples</artifactId>
+        <groupId>org.joyqueue</groupId>
+        <version>4.2.8-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>joyqueue-client-samples-springcloud-stream</artifactId>
+    <name>JoyQueue-Client-Samples-SpringCloud-Stream</name>
+    <description>Client samples using SpringCloud Stream</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.joyqueue</groupId>
+            <artifactId>openmessaging-spring-cloud-stream-binder</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+</project>

joyqueue-client/joyqueue-client-samples/joyqueue-client-samples-springcloud-stream/src/main/java/org/joyqueue/client/samples/springcloud/stream/CustomProcessor.java

@@ -0,0 +1,56 @@
+/**
+ * Copyright 2019 The JoyQueue Authors.
+ * <p>
+ * Licensed 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.joyqueue.client.samples.springcloud.stream;
+
+import org.springframework.cloud.stream.annotation.Input;
+import org.springframework.cloud.stream.annotation.Output;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.stereotype.Component;
+
+/**
+ * Custom Processor
+ */
+@Component
+public interface CustomProcessor {
+
+    String INPUT_ORDER = "inputOrder";
+
+    String OUTPUT_ORDER = "outputOrder";
+
+    /**
+     * 主题订阅通道
+     *  <p>
+     *     订阅消息通道{@link SubscribableChannel}为消息通道{@link MessageChannel}子类，该通道的所有消息被{@link org.springframework.messaging.MessageHandler}消息处理器所订阅
+     *  </p>
+     * @return {@link SubscribableChannel}
+     */
+    @Input(INPUT_ORDER)
+    SubscribableChannel inputOrder();
+
+    /**
+     * 主题消息发布通道
+     *
+     * <p>
+     *  消息通道{@link MessageChannel}用于接收消息，调用{@link MessageChannel#send(Message)}方法可以将消息发送至该消息通道中
+     * </p>
+     * @return {@link MessageChannel}
+     */
+    @Output(OUTPUT_ORDER)
+    MessageChannel outputOrder();
+
+}

joyqueue-client/joyqueue-client-samples/joyqueue-client-samples-springcloud-stream/src/main/java/org/joyqueue/client/samples/springcloud/stream/StreamBootstrap.java

@@ -0,0 +1,43 @@
+/**
+ * Copyright 2019 The JoyQueue Authors.
+ *
+ * Licensed 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.joyqueue.client.samples.springcloud.stream;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+
+@EnableBinding(CustomProcessor.class)
+@SpringBootApplication(scanBasePackages = {"org.joyqueue.client.samples.springcloud.stream"})
+public class StreamBootstrap {
+
+    public static void main(String[] args) throws InterruptedException {
+        System.setProperty("spring.profiles.active", "stream");
+        ConfigurableApplicationContext run = SpringApplication.run(StreamBootstrap.class, args);
+        CustomProcessor processor = run.getBean(CustomProcessor.class);
+
+        for (int i = 0; i < 100; i++) {
+            Message<String> message = new GenericMessage<>("Hello - " + i);
+            //Message<String> received = (Message<String>) messageCollector.forChannel(processor.output()).poll();
+            //Assert.assertThat(received.getPayload(), equalTo("hello world"));
+            processor.outputOrder().send(message);
+
+            Thread.sleep(1000);
+        }
+    }
+}

joyqueue-client/joyqueue-client-samples/joyqueue-client-samples-springcloud-stream/src/main/java/org/joyqueue/client/samples/springcloud/stream/StreamListenerService.java

@@ -0,0 +1,38 @@
+/**
+ * Copyright 2019 The JoyQueue Authors.
+ *
+ * Licensed 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.joyqueue.client.samples.springcloud.stream;
+
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.messaging.Message;
+import org.springframework.stereotype.Service;
+
+/**
+ * Test for Stream Listener
+ */
+@Service
+public class StreamListenerService {
+
+    @StreamListener(CustomProcessor.INPUT_ORDER)
+    public void receiveMessage(Message<String> message) {
+        System.out.println(String.format("接收到消息对象，headers=%s, payload=%s", message.getHeaders().toString(), message.getPayload()));
+    }
+
+    @StreamListener(CustomProcessor.INPUT_ORDER)
+    public void receiveMessageBody(String receiveMsg) {
+        System.out.println("接收到消息体: " + receiveMsg);
+    }
+
+}

joyqueue-client/joyqueue-client-samples/joyqueue-client-samples-springcloud-stream/src/main/resources/application-stream.yml

@@ -0,0 +1,53 @@
+#
+# Copyright 2019 The JoyQueue Authors.
+#
+# Licensed 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.
+#
+spring:
+  cloud:
+    stream:
+      bindings:
+        outputOrder:
+          binder: oms1
+          destination: jqtopic
+          content-type: text/plain
+          producer:
+            #通过该参数指定了分区键的表达式规则，可以根据实际的输出消息规则配置 SpEL 来生成合适的分区键
+            partitionKeyExpression: payload
+            partitionCount: 2
+        inputOrder:
+          binder: oms1
+          destination: jqtopic
+          content-type: text/plain
+          group: group1
+          consumer:
+            concurrency: 50
+      binders:
+        oms1:
+          type: oms
+      default-binder: oms
+
+      oms:
+        binder:
+          url: oms:joyqueue://jqbinder@test-nameserver.jmq.xx.local:50088/UNKNOWN
+          attributes:
+            ACCOUNT_KEY: xxxx
+        bindings:
+          outputOrder:
+            producer:
+              group: demo-group
+              sync: true
+          inputOrder:
+            consumer:
+              enable: true
+              batch: false

joyqueue-client/joyqueue-client-samples/joyqueue-client-samples-springcloud-stream/src/main/resources/application.yml

@@ -0,0 +1,21 @@
+#
+# Copyright 2019 The JoyQueue Authors.
+#
+# Licensed 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.
+#
+
+# 采用Spring Cloud Stream与Spring Boot配置冲突, Spring Boot优先级会更高
+#spring.oms:
+#  url: oms:joyqueue://test_app@127.0.0.1:50088/UNKNOWN
+#  attributes:
+#  - ACCOUNT_KEY: test_token

joyqueue-client/joyqueue-client-samples/pom.xml

@@ -33,6 +33,7 @@
         <module>joyqueue-client-samples-spring</module>
         <module>joyqueue-client-samples-springboot</module>
         <module>joyqueue-client-samples-kafka</module>
+        <module>joyqueue-client-samples-springcloud-stream</module>
     </modules>
 
     <dependencies>

joyqueue-openmessaging/openmessaging-spring-cloud-stream-binder/README.md

@@ -0,0 +1,84 @@
+# Spring Cloud Stream Binder for Openmessaging (JoyQueue)
+## 关于（About Spring Cloud Stream） 
+- 官方文档（Official document）: [docs.spring.io/spring-cloud-stream](https://docs.spring.io/spring-cloud-stream/docs/current/reference/html)
+
+## 如何使用（How to use）
+- 使用示例（Use sample） [Client Sample for Spring Cloud Stream](../../joyqueue-client/joyqueue-client-samples/joyqueue-client-samples-springcloud-stream)
+
+## 配置（Configuration）
+自定义消息处理接口
+```java
+@Component
+public interface CustomProcessor {
+
+    String INPUT_ORDER = "inputOrder";
+
+    String OUTPUT_ORDER = "outputOrder";
+
+    /**
+     * 主题订阅通道
+     *  <p>
+     *     订阅消息通道{@link SubscribableChannel}为消息通道{@link MessageChannel}子类，该通道的所有消息被{@link org.springframework.messaging.MessageHandler}消息处理器所订阅
+     *  </p>
+     * @return {@link SubscribableChannel}
+     */
+    @Input(INPUT_ORDER)
+    SubscribableChannel inputOrder();
+
+    /**
+     * 主题消息发布通道
+     *
+     * <p>
+     *  消息通道{@link MessageChannel}用于接收消息，调用{@link MessageChannel#send(Message)}方法可以将消息发送至该消息通道中
+     * </p>
+     * @return {@link MessageChannel}
+     */
+    @Output(OUTPUT_ORDER)
+    MessageChannel outputOrder();
+
+}
+
+```
+Configuration file
+```yaml
+spring:
+  cloud:
+    stream:
+      bindings:
+        #对应CustomProcessor接口定义的outputOrder()
+        outputOrder:
+          binder: oms1
+          destination: jqtopic
+          content-type: text/plain
+          producer:
+            #通过该参数指定了分区键的表达式规则，可以根据实际的输出消息规则配置 SpEL 来生成合适的分区键
+            partitionKeyExpression: payload
+            partitionCount: 2
+        inputOrder:
+          binder: oms1
+          destination: jqtopic
+          content-type: text/plain
+          group: group1
+          consumer:
+            concurrency: 50
+      binders:
+        oms1:
+          type: oms
+      default-binder: oms
+      #OMS(JoyQueue)对应的连接配置
+      oms:
+        binder:
+          url: oms:joyqueue://jqbinder@test-nameserver.jmq.xx.local:50088/UNKNOWN
+          attributes:
+            ACCOUNT_KEY: xxxx
+        bindings:
+          #OMS的自定义配置
+          outputOrder:
+            producer:
+              group: demo-group
+              sync: true
+          inputOrder:
+            consumer:
+              enable: true
+              batch: true
+```