transformer-extras

docker-compose.yml.patch

--- docker-compose.yml.original	2018-05-01 00:35:11.000000000 +0100
+++ docker-compose.yml	2018-05-01 00:44:49.000000000 +0100
@@ -1,31 +1,38 @@
 version: '3'
 
 services:
-  kafka:
-    image: wurstmeister/kafka:0.10.1.0
+  rabbitmq:
+    image: rabbitmq:3.7
     expose:
-      - "9092"
+      - "5672"
+  mysql:
+    image: mariadb:10.2
     environment:
-      - KAFKA_ADVERTISED_PORT=9092
-      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
-    depends_on:
-      - zookeeper
-  zookeeper:
-    image: wurstmeister/zookeeper
+      MYSQL_DATABASE: dataflow
+      MYSQL_USER: root
+      MYSQL_ROOT_PASSWORD: rootpw
     expose:
-      - "2181"
-    environment:
-      - KAFKA_ADVERTISED_HOST_NAME=zookeeper
+      - 3306
+  redis:
+    image: redis:2.8
+    expose:
+      - "6379"
   dataflow-server:
     image: springcloud/spring-cloud-dataflow-server-local:1.4.0.RELEASE
     container_name: dataflow-server
     ports:
       - "9393:9393"
     environment:
-      - spring.cloud.dataflow.applicationProperties.stream.spring.cloud.stream.kafka.binder.brokers=kafka:9092
-      - spring.cloud.dataflow.applicationProperties.stream.spring.cloud.stream.kafka.binder.zkNodes=zookeeper:2181
+      - spring_datasource_url=jdbc:mysql://mysql:3306/dataflow
+      - spring_datasource_username=root
+      - spring_datasource_password=rootpw
+      - spring_datasource_driver-class-name=org.mariadb.jdbc.Driver
+      - spring.cloud.dataflow.applicationProperties.stream.spring.rabbitmq.host=rabbitmq
+      - spring.cloud.dataflow.applicationProperties.stream.spring.redis.host=redis
+      - spring_redis_host=redis
     depends_on:
-      - kafka
+      - rabbitmq
+      - mysql
   app-import:
     image: alpine:3.7
     depends_on:
@@ -36,7 +43,7 @@
         do
           sleep 1;
         done;
-        wget -qO- 'http://dataflow-server:9393/apps' --post-data='uri=http://bit.ly/Celsius-SR1-stream-applications-kafka-10-maven&force=true';
+        wget -qO- 'http://dataflow-server:9393/apps' --post-data='uri=http://bit.ly/Celsius-SR1-stream-applications-rabbit-maven&force=true';
         echo 'Stream apps imported'
         wget -qO- 'http://dataflow-server:9393/apps'  --post-data='uri=http://bit.ly/Clark-GA-task-applications-maven&force=true';
         echo 'Task apps imported'"

pom.xml.patch

--- pom.xml.original	2018-04-30 23:07:31.000000000 +0100
+++ pom.xml	2018-04-30 23:07:38.000000000 +0100
@@ -28,10 +28,14 @@
 	<dependencies>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-stream</artifactId>
+			<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
 		</dependency>
 
 		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-web</artifactId>
+		</dependency>
+		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>
 			<scope>test</scope>

Transformer.java

package io.spring.stream.sample;

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;

import java.util.HashMap;
import java.util.Map;

@EnableBinding(Processor.class)
public class Transformer {

    @StreamListener(Processor.INPUT)
    @Output(Processor.OUTPUT)
    public Map<String, Object> transform(Map<String, Object> doc) {
        Map<String, Object> map = new HashMap<>();
        map.put("sensor_id", doc.getOrDefault("id", "-1"));
        map.put("temp_val", doc.getOrDefault("temperature", "-999"));
        return map;
    }
}

TransformerApplicationTests.java.patch

--- TransformerApplicationTests.java.original	2018-04-30 21:40:01.000000000 +0100
+++ TransformerApplicationTests.java	2018-04-30 21:40:35.000000000 +0100
@@ -2,15 +2,35 @@
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringRunner;
 
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.entry;
+
 @RunWith(SpringRunner.class)
-@SpringBootTest
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
 public class TransformerApplicationTests {
 
-	@Test
-	public void contextLoads() {
-	}
+    @Autowired
+    private Transformer transformer;
+
+    @Test
+    public void simpleTest() {
+        Map<String, Object> resultMap = transformer.transform(createInputData());
+        assertThat(resultMap).hasSize(2)
+                .contains(entry("sensor_id", "1"))
+                .contains(entry("temp_val", "100"));
+    }
 
+    private Map<String, Object> createInputData() {
+        HashMap<String, Object> inputData = new HashMap<>();
+        inputData.put("id", "1");
+        inputData.put("temperature", "100");
+        return inputData;
+    }
 }