feature: rabbitmq as a central bus transport + protobuf as payload

.gitignore

@@ -25,4 +25,5 @@ nbdist/
 .nb-gradle/
 
 .DS_Store
-.env
+.env
+*/gen/

docker-compose-live.yml

@@ -28,9 +28,11 @@ services:
     image: adyach/mrbot-door-service:d41cce1
     environment:
       SPRING_DATA_MONGODB_URI: mongodb://door-service-mongo/test
+      RABBITMQ_HOST: rabbitmq
     links:
      - door-service-mongo
      - mrbot
+     - rabbitmq
 
   door-service-mongo:
     image: mongo:3.4.3
@@ -41,4 +43,7 @@ services:
     build:
       context: ./temp-service
     links:
-     - mrbot
+     - mrbot
+	 
+  rabbitmq:
+    image: rabbitmq:3.6.9-management-alpine

docker-compose.yml

@@ -20,26 +20,33 @@ services:
      - "443:443"
     volumes_from:
      - mrbot
-    volumes:
-     - /etc/letsencrypt:/etc/letsencrypt
-     - /etc/ssl/certs:/etc/ssl/certs
 
   door-service:
     build:
       context: ./door-service
     environment:
       SPRING_DATA_MONGODB_URI: mongodb://door-service-mongo/test
+      RABBITMQ_HOST: rabbitmq
     links:
      - door-service-mongo
      - mrbot
+     - rabbitmq
 
   door-service-mongo:
     image: mongo:3.4.3
     volumes:
      - /data/db
-
+    ports:
+      - "27017:27017"
+	  
   temp-service:
     build:
       context: ./temp-service
     links:
-     - mrbot
+     - mrbot
+
+  rabbitmq:
+    image: rabbitmq:3.6.9-management-alpine
+    ports:
+      - "5672:5672"
+      - "15672:15672"

door-service/build.gradle

@@ -10,14 +10,16 @@ buildscript {
 	}
 	dependencies {
 		classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
+		classpath 'com.google.protobuf:protobuf-gradle-plugin:0.8.1'
 	}
 }
 
 apply plugin: 'java'
-apply plugin: 'eclipse'
+apply plugin: 'idea'
 apply plugin: 'org.springframework.boot'
 apply plugin: 'io.spring.dependency-management'
 apply plugin: 'application'
+apply plugin: 'com.google.protobuf'
 
 mainClassName = 'org.crazycoder.door.service.DoorServiceApplication'
 
@@ -40,6 +42,35 @@ dependencies {
 	compile('org.springframework.boot:spring-boot-starter-webflux')
 	compile("org.springframework.boot:spring-boot-starter-data-mongodb")
 	compile("com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.4.0")
+	compile("org.springframework.boot:spring-boot-starter-amqp")
+	compile("org.springframework.boot:spring-boot-starter-actuator")
+	compile 'com.google.protobuf:protobuf-java:3.3.0'
 
 	testCompile('org.springframework.boot:spring-boot-starter-test')
 }
+
+sourceSets {
+	main {
+		proto {
+			srcDir 'src/main/java/proto'
+		}
+	}
+}
+
+protobuf {
+	generatedFilesBaseDir = "$projectDir/gen"
+	protoc {
+		artifact = 'com.google.protobuf:protoc:3.3.0'
+	}
+}
+
+
+clean {
+	delete protobuf.generatedFilesBaseDir
+}
+
+idea {
+	module {
+		sourceDirs += file("${protobuf.generatedFilesBaseDir}/main/java")
+	}
+}

door-service/src/main/java/org/crazycoder/door/service/DoorConfig.java

@@ -0,0 +1,33 @@
+package org.crazycoder.door.service;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class DoorConfig {
+
+    @Value("${door.service.queue}")
+    private String doorServiceQueue;
+    @Value("${door.service.topic}")
+    private String doorServiceTopic;
+
+    @Bean
+    Queue queue() {
+        return new Queue(doorServiceQueue, false);
+    }
+
+    @Bean
+    TopicExchange exchange() {
+        return new TopicExchange(doorServiceTopic);
+    }
+
+    @Bean
+    Binding binding(Queue queue, TopicExchange exchange) {
+        return BindingBuilder.bind(queue).to(exchange).with(doorServiceQueue);
+    }
+}

door-service/src/main/java/org/crazycoder/door/service/DoorController.java

@@ -4,8 +4,6 @@
 import org.crazycoder.door.service.domain.Heartbeat;
 import org.crazycoder.door.service.view.DoorStatusView;
 import org.crazycoder.door.service.view.HeartbeatView;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -23,43 +21,34 @@
 @RestController
 public class DoorController {
 
-    private static final Logger LOG = LoggerFactory.getLogger(DoorController.class);
 
     @Autowired
-    private HeartbeatRepository heartbeatRepository;
-    @Autowired
-    private DoorStatusRepository doorStatusRepository;
+    private DoorService doorService;
 
     @PostMapping("/home/security/door/status")
     public ResponseEntity status(@RequestBody DoorStatusView doorStatusView) {
-        LOG.info("Door status {}", doorStatusView);
-        doorStatusRepository.save(new DoorStatus(doorStatusView.getDeviceId(), DoorStatus.Status.valueOf(doorStatusView.getState())));
+        doorService.saveDoorStatus(new DoorStatus(doorStatusView.getDeviceId(), DoorStatus.Status.valueOf(doorStatusView.getState())));
         return ResponseEntity.ok().build();
     }
 
     @PostMapping("/home/security/door/heartbeat")
     public ResponseEntity heartbeat(@RequestBody HeartbeatView heartbeatView) {
-        LOG.info("Door status heartbeat {}", heartbeatView);
-        heartbeatRepository.save(new Heartbeat(heartbeatView.getDeviceId()));
+        doorService.saveHeartbeat(new Heartbeat(heartbeatView.getDeviceId()));
         return ResponseEntity.ok().build();
     }
 
     @GetMapping("/home/security/door/heartbeat/{device_id}")
     public ResponseEntity<List<HeartbeatView>> getHeartbeat(
             @PathVariable("device_id") String deviceId,
             @RequestParam(value = "limit", required = false, defaultValue = "5") int limit) {
-        return ResponseEntity.ok(heartbeatRepository.findByDeviceIdOrderByTimestampDesc(deviceId).stream()
-                .map(this::createHeartbeatView)
-                .limit(limit)
-                .collect(Collectors.toList()));
+        return ResponseEntity.ok(doorService.getHeartbeat(deviceId, limit).stream()
+                .map(this::createHeartbeatView).collect(Collectors.toList()));
     }
 
     @GetMapping("/home/security/door/heartbeats")
-    public ResponseEntity<List<HeartbeatView>> getHeartbeats(@RequestParam(value = "limit", required = false, defaultValue = "5") final int limit) {
-        return ResponseEntity.ok(heartbeatRepository.findAll().stream()
-                .map(this::createHeartbeatView)
-                .limit(limit)
-                .collect(Collectors.toList()));
+    public ResponseEntity<List<HeartbeatView>> getHeartbeats(@RequestParam(value = "limit", required = false, defaultValue = "5") int limit) {
+        return ResponseEntity.ok(doorService.getHeartbeats(limit).stream()
+                .map(this::createHeartbeatView).collect(Collectors.toList()));
     }
 
     private HeartbeatView createHeartbeatView(Heartbeat heartbeat) {
@@ -72,19 +61,15 @@ private HeartbeatView createHeartbeatView(Heartbeat heartbeat) {
     @GetMapping("/home/security/door/status/{device_id}")
     public ResponseEntity<List<DoorStatusView>> getDoorStatus(
             @PathVariable("device_id") String deviceId,
-            @RequestParam(value = "limit", required = false, defaultValue = "5") final int limit) {
-        return ResponseEntity.ok(doorStatusRepository.findByDeviceIdOrderByTimestampDesc(deviceId).stream()
-                .map(this::createDoorStatus)
-                .limit(limit)
-                .collect(Collectors.toList()));
+            @RequestParam(value = "limit", required = false, defaultValue = "5") int limit) {
+        return ResponseEntity.ok(doorService.getDoorStatus(deviceId, limit).stream()
+                .map(this::createDoorStatus).collect(Collectors.toList()));
     }
 
     @GetMapping("/home/security/door/statuses")
     public ResponseEntity<List<DoorStatusView>> getDoorStatuses(@RequestParam(value = "limit", required = false, defaultValue = "5") int limit) {
-        return ResponseEntity.ok(doorStatusRepository.findAllByOrderByTimestampDesc().stream()
-                .map(this::createDoorStatus)
-                .limit(limit)
-                .collect(Collectors.toList()));
+        return ResponseEntity.ok(doorService.getDoorStatuses(limit).stream()
+                .map(this::createDoorStatus).collect(Collectors.toList()));
     }
 
     private DoorStatusView createDoorStatus(DoorStatus doorStatus) {

door-service/src/main/java/org/crazycoder/door/service/DoorService.java

@@ -0,0 +1,73 @@
+package org.crazycoder.door.service;
+
+import org.crazycoder.door.service.domain.DoorStatus;
+import org.crazycoder.door.service.domain.Heartbeat;
+import org.crazycoder.door.service.protobuf.DoorSensorData;
+import org.crazycoder.door.service.protobuf.Status;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+@Service
+public class DoorService {
+
+    private static final Logger LOG = LoggerFactory.getLogger(DoorService.class);
+
+    @Autowired
+    private HeartbeatRepository heartbeatRepository;
+    @Autowired
+    private DoorStatusRepository doorStatusRepository;
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+    @Value("${door.service.queue}")
+    private String doorServiceQueue;
+    @Value("${door.service.topic}")
+    private String doorServiceTopic;
+
+    public void saveDoorStatus(DoorStatus doorStatus) {
+        LOG.info("Door status {}", doorStatus);
+        doorStatusRepository.save(doorStatus);
+        DoorSensorData doorSensorData = DoorSensorData.newBuilder()
+                .setDeviceId(doorStatus.getDeviceId())
+                .setTimestamp(doorStatus.getTimestamp())
+                .setStatus(Status.valueOf(doorStatus.getStatus().name()))
+                .build();
+        rabbitTemplate.convertAndSend(doorServiceTopic, doorServiceQueue, doorSensorData);
+    }
+
+    public void saveHeartbeat(Heartbeat heartbeat) {
+        LOG.info("Door status heartbeat {}", heartbeat);
+        heartbeatRepository.save(heartbeat);
+    }
+
+    public List<Heartbeat> getHeartbeat(String deviceId, int limit) {
+        return heartbeatRepository.findByDeviceIdOrderByTimestampDesc(deviceId).stream()
+                .limit(limit)
+                .collect(Collectors.toList());
+    }
+
+    public List<Heartbeat> getHeartbeats(int limit) {
+        return heartbeatRepository.findAll().stream()
+                .limit(limit)
+                .collect(Collectors.toList());
+    }
+
+    public List<DoorStatus> getDoorStatus(String deviceId, int limit) {
+        return doorStatusRepository.findByDeviceIdOrderByTimestampDesc(deviceId).stream()
+                .limit(limit)
+                .collect(Collectors.toList());
+    }
+
+    public List<DoorStatus> getDoorStatuses(int limit) {
+        return doorStatusRepository.findAllByOrderByTimestampDesc().stream()
+                .limit(limit)
+                .collect(Collectors.toList());
+    }
+
+}

door-service/src/main/java/proto/door-sensor-data.proto

@@ -0,0 +1,15 @@
+syntax = "proto3";
+
+option java_package = "org.crazycoder.door.service.protobuf";
+option java_multiple_files = true;
+
+message DoorSensorData {
+    string deviceId = 1;
+    int64 timestamp = 2;
+    Status status = 3;
+}
+
+enum Status {
+    OPENED = 0;
+    CLOSED = 1;
+}

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

@@ -0,0 +1,8 @@
+spring:
+  jackson:
+    serialization:
+      WRITE_DATES_AS_TIMESTAMPS: false
+
+door.service:
+  queue: "door.service.state"
+  topic: "door.service"

hardware/temp_sensor/temp_sensor.ino

@@ -34,7 +34,7 @@ void getDHTData() {
   pinMode(D0, WAKEUP_PULLUP);
 
   dht.begin();
-  delay(2000);
+  delay(3000);
 
   float humidity = dht.readHumidity();
   float temperature = dht.readTemperature();