refactor code; add rabbitmq stomp

data-consumer/src/main/java/com/artgeektech/iotmicroservices/controller/DataConsumerController.java

@@ -3,46 +3,38 @@
 import com.artgeektech.iotmicroservices.Constants;
 import com.artgeektech.iotmicroservices.model.AirData;
 import com.artgeektech.iotmicroservices.repository.AirDataRepository;
+import com.artgeektech.iotmicroservices.service.RuleEngineService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
+import org.springframework.stereotype.Controller;
 
 /**
  * Created by guang on 11:56 AM 8/18/18.
  */
-@Component
+@Controller
 public class DataConsumerController {
     private static final Logger logger = LoggerFactory.getLogger(DataConsumerController.class);
 
     @Autowired
     private AirDataRepository airDataRepository;
 
-    @RabbitListener(queues = Constants.QUEUE_NAME)
+    @Autowired
+    private RuleEngineService ruleEngineService;
+
+    @RabbitListener(queues = Constants.QUEUE_NAME)  // Subscribe to the Message Queue
     public void process(AirData airData) {
 
         logger.info("Received message from MQ '{}'", airData);
 
-        // simple rule engine handling
-        if (airData.getTemperature() > 50) {
-            triggerActionAlert("Temperature too high!!");
-        }
-        if (airData.getCo2() > 50) {
-            triggerActionAlert("CO2 too high!!!");
-        }
-        if (airData.getPm2p5() > 50) {
-            triggerActionAlert("Too much Dust!!!");
-        }
+        // apply rule engine and trigger actions
+        ruleEngineService.applyRules(airData);
 
         // save to DB
         airDataRepository.save(airData);
+
         logger.info("Saved message to Mongo DB '{}'", airData);
         logger.info("Total message saved in Mongo DB is:  " + airDataRepository.findAll().size());
-
-    }
-
-    private void triggerActionAlert(String msg) {
-        System.out.println("\n\n!!!!!Sending the Email Alert: " + msg + "\n\n");
     }
 }

data-consumer/src/main/java/com/artgeektech/iotmicroservices/service/RuleEngineService.java

@@ -0,0 +1,27 @@
+package com.artgeektech.iotmicroservices.service;
+
+import com.artgeektech.iotmicroservices.model.AirData;
+import org.springframework.stereotype.Component;
+
+/**
+ * Created by guang on 7:16 PM 8/21/18.
+ */
+@Component
+public class RuleEngineService {
+
+    public void applyRules(AirData airData) {
+        if (airData.getTemperature() > 50) {
+            triggerActionAlert("Temperature too high!!");
+        }
+        if (airData.getCo2() > 50) {
+            triggerActionAlert("CO2 too high!!!");
+        }
+        if (airData.getPm2p5() > 50) {
+            triggerActionAlert("Too much Dust!!!");
+        }
+    }
+
+    private void triggerActionAlert(String msg) {
+        System.out.println("\n\n!!!!!Sending the Email Alert: " + msg + "\n\n");
+    }
+}

data-dashboard/data-dashboard.iml

@@ -0,0 +1,9 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="JAVA_MODULE" version="4">
+  <component name="NewModuleRootManager" inherit-compiler-output="true">
+    <exclude-output />
+    <content url="file://$MODULE_DIR$" />
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>

data-dashboard/pom.xml

@@ -0,0 +1,15 @@
+<?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>iotmicroservices</artifactId>
+        <groupId>com.artgeektech</groupId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>datadashboard</artifactId>
+
+
+</project>

data-dashboard/src/main/java/DataDashboardController.java

@@ -0,0 +1,7 @@
+package PACKAGE_NAME;
+
+/**
+ * Created by guang on 7:51 PM 8/20/18.
+ */
+public class DataDashboardController {
+}

data-ingest/pom.xml

@@ -20,5 +20,19 @@
             <groupId>org.springframework.amqp</groupId>
             <artifactId>spring-rabbit</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.springframework.boot</groupId>
+                    <artifactId>spring-boot-starter-tomcat</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-jetty</artifactId>
+        </dependency>
     </dependencies>
 </project>

data-ingest/src/main/java/com/artgeektech/iotmicroservices/DataIngestApplication.java

@@ -18,6 +18,11 @@ public Exchange airDataExchange() {
         return new TopicExchange(Constants.EXCHANGE_NAME);
     }
 
+//    @Bean
+//    public SimpMessagingTemplate simpMessagingTemplate() {
+//        return new SimpMessagingTemplate();
+//    }
+
     public static void main(String[] args) {
         SpringApplication.run(DataIngestApplication.class, args);
     }

data-ingest/src/main/java/com/artgeektech/iotmicroservices/controller/DataIngestController.java

@@ -2,43 +2,58 @@
 
 import com.artgeektech.iotmicroservices.Constants;
 import com.artgeektech.iotmicroservices.model.AirData;
+import com.artgeektech.iotmicroservices.model.AirRawData;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.amqp.core.Exchange;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
 import javax.validation.Valid;
 import java.util.Date;
 
 
 @RestController
-@RequestMapping("/iot/airdata")
 public class DataIngestController {
 
     private static final Logger logger = LoggerFactory.getLogger(DataIngestController.class);
 
+//    @Autowired
+//    private SimpMessagingTemplate messagingTemplate;
+
     @Autowired
     private RabbitTemplate rabbitTemplate;
 
     @Autowired
     private Exchange exchange;
 
-    @PostMapping("/ingest")
-    public AirData ingest(@Valid @RequestBody AirData airData) {
-        preprocess(airData);
+    @PostMapping("/airdata/ingest")  // validate payload from request body
+    public AirData ingest(@Valid @RequestBody AirRawData rawData) {
+        // preprocess
+        AirData airData = preprocess(rawData);
+        // publish to MQ
         rabbitTemplate.convertAndSend(exchange.getName(), Constants.ROUTING_KEY, airData);
         logger.info("ingested data: " + airData.toString());
         return airData;
     }
 
-    private void preprocess(AirData airData) {
+
+
+    private AirData preprocess(AirRawData rawData) {
+        AirData airData = new AirData();
+
+        // add more info from system
         airData.setTimestamp(new Date());
-        airData.setHumidity(Math.round(airData.getHumidity() * 100.0) / 100.0);
-        airData.setTemperature(Math.round(airData.getTemperature() * 100.0) / 100.0);
+
+        // standardize data format
+        airData.setHumidity(Math.round(rawData.getHumidity() * 100.0) / 100.0);
+        airData.setTemperature(Math.round(rawData.getTemperature() * 100.0) / 100.0);
+        airData.setCo2(Math.round(rawData.getCo2() * 100.0) / 100.0);
+        airData.setPm2p5(Math.round(rawData.getPm2p5() * 100.0) / 100.0);
+
+        return airData;
     }
 }

data-ingest/src/main/java/com/artgeektech/iotmicroservices/model/AirData.java

@@ -4,36 +4,16 @@
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
-import javax.validation.constraints.Max;
-import javax.validation.constraints.Min;
-import javax.validation.constraints.NotNull;
 import java.io.Serializable;
 import java.util.Date;
 
 @Data
 @AllArgsConstructor
 @NoArgsConstructor
 public class AirData implements Serializable {
-
-    @NotNull
-    @Min(1)
-    @Max(100)
     private Double temperature;
-
-    @NotNull
-    @Min(1)
-    @Max(100)
     private Double humidity;
-
-    @NotNull
-    @Min(1)
-    @Max(100)
     private Double pm2p5;
-
-    @NotNull
-    @Min(1)
-    @Max(100)
     private Double co2;
-
     private Date timestamp;
 }

data-ingest/src/main/java/com/artgeektech/iotmicroservices/model/AirRawData.java

@@ -0,0 +1,41 @@
+package com.artgeektech.iotmicroservices.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import javax.validation.constraints.Max;
+import javax.validation.constraints.Min;
+import javax.validation.constraints.NotNull;
+import java.io.Serializable;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class AirRawData implements Serializable {
+//    private Double temperature;
+//    private Double humidity;
+//    private Double pm2p5;
+//    private Double co2;
+
+    @NotNull
+    @Min(1)
+    @Max(100)
+    private Double temperature;
+
+    @NotNull
+    @Min(1)
+    @Max(100)
+    private Double humidity;
+
+    @NotNull
+    @Min(1)
+    @Max(100)
+    private Double pm2p5;
+
+    @NotNull
+    @Min(1)
+    @Max(100)
+    private Double co2;
+
+}

data-service/data-service-api-gateway.iml

@@ -0,0 +1,2 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="JAVA_MODULE" version="4" />

data-service/pom.xml

@@ -0,0 +1,15 @@
+<?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>iotmicroservices</artifactId>
+        <groupId>com.artgeektech</groupId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>dataservice</artifactId>
+
+
+</project>

data-simulator/src/main/java/com/artgeektech/iotmicroservices/DataSimulatorApplication.java

@@ -10,7 +10,7 @@
 
 public class DataSimulatorApplication {
 
-    private static final String resourceUrl = "http://data-ingest:9001/iot/airdata/ingest";
+    private static final String resourceUrl = "http://localhost:9001/airdata/ingest";
     private static final RestTemplate restTemplate = new RestTemplate();
     private static final Random random = new Random();
     private static double minVal = 10;
@@ -19,20 +19,21 @@ public class DataSimulatorApplication {
     private static int interval = 1000;
 
     public static void main(String[] args) {
+
         timer.scheduleAtFixedRate(new TimerTask() {
             @Override
             public void run() {
                 AirRawData payload = new AirRawData(genRandom(), genRandom(), genRandom(), genRandom());
                 HttpEntity<AirRawData> request = new HttpEntity<>(payload);
-                restTemplate.postForObject(resourceUrl, request, AirRawData.class);
+                restTemplate.postForObject(resourceUrl, request, Object.class);
             }
         }, 0, interval);
     }
 
     private static double genRandom() {
         int randInt = random.nextInt(5);
         if (randInt == 0) {
-            return 100;
+            return 99;
         } else {
             return minVal + random.nextDouble() * (maxVal - minVal);
         }

data-simulator/src/main/java/com/artgeektech/iotmicroservices/model/AirRawData.java

@@ -10,8 +10,8 @@
 @AllArgsConstructor
 @NoArgsConstructor
 public class AirRawData implements Serializable {
-    private double temperature;
-    private double humidity;
-    private double pm2p5;
-    private double co2;
+    private Double temperature;
+    private Double humidity;
+    private Double pm2p5;
+    private Double co2;
 }

docker-rabbitmq/docker-compose.yml

@@ -1,6 +1,10 @@
-rabbitmq:
-  image: rabbitmq:management
-  container_name: "rabbitmq"
-  ports:
-    - 5672:5672
-    - 15672:15672
+version: '2'
+services:
+    sonar:
+        container_name: aiwin-rabbit
+        image: aiwin/rabbitmq-stomp:latest
+        ports:
+            - 61613:61613
+            - 15674:15674
+            - 15672:15672
+            - 5672:5672

service-registry/src/main/resources/application.yml

@@ -7,7 +7,7 @@ server:
 
 eureka:
   instance:
-    hostname: service-registry         # 指定该Eureka实例的主机名
+    hostname: localhost         # 指定该Eureka实例的主机名
   client:
     registerWithEureka: false
     fetchRegistry: false