add mongodb had issue

data-consumer/pom.xml

@@ -20,5 +20,9 @@
             <groupId>org.springframework.amqp</groupId>
             <artifactId>spring-rabbit</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-mongodb</artifactId>
+        </dependency>
     </dependencies>
 </project>

data-consumer/src/main/java/com/artgeektech/iotmicroservices/Constants.java

@@ -0,0 +1,13 @@
+package com.artgeektech.iotmicroservices;
+
+/**
+ * Created by guang on 5:49 AM 8/19/18.
+ */
+public class Constants {
+
+    public static final String EXCHANGE_NAME = "AirDataExchange";
+
+    public static final String QUEUE_NAME = "AirDataQueue";
+
+    public static final String ROUTING_KEY = "airdata.ingested";
+}

data-consumer/src/main/java/com/artgeektech/iotmicroservices/DataConsumerApplication.java

@@ -1,9 +1,16 @@
 package com.artgeektech.iotmicroservices;
 
+import com.artgeektech.iotmicroservices.model.AirData;
+import com.artgeektech.iotmicroservices.repository.AirDataRepository;
+import org.springframework.amqp.core.*;
 import org.springframework.amqp.rabbit.annotation.EnableRabbit;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
+import org.springframework.context.annotation.Bean;
+
+import java.sql.Timestamp;
 
 /**
  * Created by guang on 11:42 AM 8/18/18.
@@ -12,7 +19,52 @@
 @EnableDiscoveryClient
 @EnableRabbit
 public class DataConsumerApplication {
+
+    @Autowired
+    private static AirDataRepository airDataRepository;
+
+
+    @Bean
+    public Exchange airDataExchange() {
+        return new TopicExchange(Constants.EXCHANGE_NAME);
+    }
+
+    @Bean
+    public Queue queue() {
+        return new Queue(Constants.QUEUE_NAME);
+    }
+
+    @Bean
+    public Binding binding(Queue queue, Exchange dataExchange) {
+        return BindingBuilder
+            .bind(queue)
+            .to(dataExchange)
+            .with(Constants.ROUTING_KEY).noargs();
+    }
+
     public static void main(String[] args) {
         SpringApplication.run(DataConsumerApplication.class, args);
+        System.out.println("Main in Mongo");
+
+        // save to DB
+        AirData data1 = new AirData(
+            1d, 1d, 1d, 1d,
+            new Timestamp(System.currentTimeMillis()));
+        AirData data2 = new AirData(
+            1d, 1d, 1d, 1d,
+            new Timestamp(System.currentTimeMillis()));
+        AirData data3 = new AirData(
+            2d, 1d, 1d, 1d,
+            new Timestamp(System.currentTimeMillis()));
+        airDataRepository.save(data1);
+        airDataRepository.save(data2);
+        airDataRepository.save(data3);
+        airDataRepository.save(data3);
+
+        for (AirData data: airDataRepository.findAll()) {
+            System.out.println("printing Mongo DB");
+            System.out.println(data);
+            System.out.println(data.toString());
+        }
     }
 }

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

@@ -1,10 +1,11 @@
 package com.artgeektech.iotmicroservices.controller;
 
+import com.artgeektech.iotmicroservices.Constants;
+import com.artgeektech.iotmicroservices.model.AirData;
+import com.artgeektech.iotmicroservices.repository.AirDataRepository;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.amqp.core.Exchange;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
-import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
@@ -14,21 +15,37 @@
 @Component
 public class DataConsumerController {
     private static final Logger logger = LoggerFactory.getLogger(DataConsumerController.class);
-    private static final String routingKey = "airdata.ingested";
 
     @Autowired
-    private RabbitTemplate rabbitTemplate;
+    private AirDataRepository airDataRepository;
 
-    @Autowired
-    private Exchange exchange;
+    @RabbitListener(queues = Constants.QUEUE_NAME)
+    public void process(AirData airData) {
+
+        logger.info("Received message from MQ '{}'", airData);
 
-    @RabbitListener(queues="DataQueue")
-    public void receive1(String message) {
-        logger.info("Received message '{}'", message);
+        // 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!!!");
+        }
+
+        // save to DB
+        airDataRepository.save(airData);
+        logger.info("Saved message to Mongo DB '{}'", airData);
+        for (AirData data: airDataRepository.findAll()) {
+            System.out.println("printing Mongo DB");
+            System.out.println(data);
+            System.out.println(data.toString());
+        }
     }
 
-//    @RabbitListener(queues="DataQueue")
-//    public void receive2(AirData message) {
-//        logger.info("Received message '{}'", message.toString());
-//    }
+    private void triggerActionAlert(String msg) {
+        System.out.println("Sending the Email Alert: " + msg);
+    }
 }

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

@@ -1,12 +1,9 @@
-package com.artgeektech.iotmicroservices;
+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;
 import java.sql.Timestamp;
 
@@ -15,25 +12,9 @@
 @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 Timestamp timestamp;
 }

data-consumer/src/main/java/com/artgeektech/iotmicroservices/repository/AirDataRepository.java

@@ -0,0 +1,10 @@
+package com.artgeektech.iotmicroservices.repository;
+
+import com.artgeektech.iotmicroservices.model.AirData;
+import org.springframework.data.mongodb.repository.MongoRepository;
+
+/**
+ * Created by guang on 7:04 AM 8/19/18.
+ */
+public interface AirDataRepository extends MongoRepository<AirData, String> {
+}

data-consumer/src/main/resources/application.yml

@@ -6,6 +6,13 @@ spring:
     port: 5672
     username: guest
     password: guest
+  data:
+    mongodb:
+      host: localhost
+      port: 27072
+      database: airdata-db
+      username: root
+      password: password
 
 server:
   port: 9002                    # 指定该Eureka实例的端口
@@ -16,12 +23,3 @@ eureka:
   client:
     serviceUrl:
       defaultZone: http://service-registry:8761/eureka/
-
-app:
-  exchange:
-    name: DataExchange
-  queue:
-    name: DataQueue
-  routing:
-    key: airdata.ingested
-

data-consumer/target/classes/application.yml

@@ -6,6 +6,13 @@ spring:
     port: 5672
     username: guest
     password: guest
+  data:
+    mongodb:
+      host: localhost
+      port: 27072
+      database: airdata-db
+      username: root
+      password: password
 
 server:
   port: 9002                    # 指定该Eureka实例的端口
@@ -16,12 +23,3 @@ eureka:
   client:
     serviceUrl:
       defaultZone: http://service-registry:8761/eureka/
-
-app:
-  exchange:
-    name: DataExchange
-  queue:
-    name: DataQueue
-  routing:
-    key: airdata.ingested
-

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

@@ -0,0 +1,13 @@
+package com.artgeektech.iotmicroservices;
+
+/**
+ * Created by guang on 5:49 AM 8/19/18.
+ */
+public class Constants {
+
+    public static final String EXCHANGE_NAME = "AirDataExchange";
+
+    public static final String QUEUE_NAME = "AirDataQueue";
+
+    public static final String ROUTING_KEY = "airdata.ingested";
+}

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

@@ -1,14 +1,23 @@
 package com.artgeektech.iotmicroservices;
 
+import org.springframework.amqp.core.Exchange;
+import org.springframework.amqp.core.TopicExchange;
 import org.springframework.amqp.rabbit.annotation.EnableRabbit;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
+import org.springframework.context.annotation.Bean;
 
 @SpringBootApplication
 @EnableDiscoveryClient
 @EnableRabbit
 public class DataIngestApplication {
+
+    @Bean
+    public Exchange airDataExchange() {
+        return new TopicExchange(Constants.EXCHANGE_NAME);
+    }
+
     public static void main(String[] args) {
         SpringApplication.run(DataIngestApplication.class, args);
     }

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

@@ -1,5 +1,6 @@
 package com.artgeektech.iotmicroservices.controller;
 
+import com.artgeektech.iotmicroservices.Constants;
 import com.artgeektech.iotmicroservices.model.AirData;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -20,7 +21,6 @@
 public class DataIngestController {
 
     private static final Logger logger = LoggerFactory.getLogger(DataIngestController.class);
-    private static final String routingKey = "airdata.ingested";
 
     @Autowired
     private RabbitTemplate rabbitTemplate;
@@ -31,8 +31,8 @@ public class DataIngestController {
     @PostMapping("/ingest")
     public AirData ingest(@Valid @RequestBody AirData airData) {
         preprocess(airData);
-        rabbitTemplate.convertAndSend(exchange.getName(), routingKey, airData.toString());
-        logger.info(airData.toString());
+        rabbitTemplate.convertAndSend(exchange.getName(), Constants.ROUTING_KEY, airData);
+        logger.info("ingested data: " + airData.toString());
         return airData;
     }
 

data-ingest/src/main/resources/application.yml

@@ -16,4 +16,3 @@ eureka:
   client:
     serviceUrl:
       defaultZone: http://service-registry:8761/eureka/
-

data-ingest/target/classes/application.yml

@@ -16,4 +16,3 @@ eureka:
   client:
     serviceUrl:
       defaultZone: http://service-registry:8761/eureka/
-