Submit

MessageProcessorModule/pom.xml

@@ -0,0 +1,74 @@
+<?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">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>com.lotus.iot.mpm</groupId>
+    <artifactId>message-processor-module</artifactId>
+    <version>1.0.0</version>
+
+    <parent>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>2.0.6.RELEASE</version>
+    </parent>
+
+    <properties>
+        <java.version>1.8</java.version>
+        <junit.version>4.12</junit.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.jayway.jsonpath</groupId>
+            <artifactId>json-path</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <!-- https://mvnrepository.com/artifact/com.google.code.gson/gson -->
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
+            <version>2.8.5</version>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>${junit.version}</version>
+        </dependency>
+
+    </dependencies>
+
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+    <repositories>
+        <repository>
+            <id>spring-releases</id>
+            <url>https://repo.spring.io/libs-release</url>
+        </repository>
+    </repositories>
+    <pluginRepositories>
+        <pluginRepository>
+            <id>spring-releases</id>
+            <url>https://repo.spring.io/libs-release</url>
+        </pluginRepository>
+    </pluginRepositories>
+
+</project>

MessageProcessorModule/src/main/java/com/lotus/iot/mpm/Application.java

@@ -0,0 +1,150 @@
+package com.lotus.iot.mpm;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonParser;
+import com.lotus.iot.mpm.api.objectmodel.request.SubscribeRequest;
+import com.lotus.iot.mpm.api.objectmodel.response.AllAssetsInStateResponse;
+import com.lotus.iot.mpm.api.objectmodel.response.AssetCurrentStatusResponse;
+import com.lotus.iot.mpm.api.objectmodel.response.GeneralResponse;
+import com.lotus.iot.mpm.eventhandler.AssetStatusChangeQueue;
+import com.lotus.iot.mpm.eventhandler.EventHandlerFacade;
+import com.lotus.iot.mpm.eventhandler.objectmodel.AssetStatusEvent;
+import com.lotus.iot.mpm.eventhandler.objectmodel.SampleListener;
+import com.lotus.iot.mpm.handler.MessageHandler;
+import com.lotus.iot.mpm.objectmodel.AssetStatusMessage;
+import com.lotus.iot.mpm.objectmodel.type.StatusType;
+import com.lotus.iot.mpm.report.ReportHandler;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
+
+import java.io.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * @author Niloufar Mazloumpour
+ * @mail niloufar@mazloumpour.net
+ * @since 2018-05-22
+ */
+@SpringBootApplication
+public class Application {
+    public static void main(String[] args) {
+        ConfigurableApplicationContext applicationContext = SpringApplication.run(Application.class);
+
+        subscribeSomeListeners(applicationContext);
+
+        if (args.length > 0) {
+            File file = new File(args[0]);
+            sendMessages(file, applicationContext);
+        }
+
+        waitForProcessToFinish(applicationContext);
+
+        System.out.println();
+        System.out.println("-------------------");
+        System.out.println("Do some Reporting..");
+        System.out.println("-------------------");
+        System.out.println();
+
+        currentStatusReport(applicationContext);
+
+        assetsInStateReport(applicationContext);
+    }
+
+    private static void subscribeSomeListeners(ConfigurableApplicationContext applicationContext) {
+        EventHandlerFacade eventHandlerFacade = applicationContext.getBean(EventHandlerFacade.class);
+        List<String> rows = loadFromFile("subscribe.txt");
+        for (String row : rows) {
+            String[] subscribeInfo = row.split(",");
+            try {
+                AssetStatusEvent event = AssetStatusEvent.valueOf(subscribeInfo[1]);
+                String assetId = subscribeInfo[0];
+                SubscribeRequest subscribeRequest = new SubscribeRequest(assetId, event, new SampleListener());
+                eventHandlerFacade.subscribe(subscribeRequest);
+            } catch (IllegalArgumentException e) {
+                System.out.println(String.format("AssetStatusEvent %s does not exist", subscribeInfo[1]));
+            }
+        }
+    }
+
+    private static void assetsInStateReport(ConfigurableApplicationContext applicationContext) {
+        ReportHandler reportHandler = applicationContext.getBean(ReportHandler.class);
+        List<String> statusTypeList = loadFromFile("all_assets_in_state_report.txt");
+        for (String status : statusTypeList) {
+            GeneralResponse response = reportHandler.assetsInState(status);
+            try {
+                AllAssetsInStateResponse allAssetsInStateResponse = (AllAssetsInStateResponse) response;
+                List<String> assetIdList = allAssetsInStateResponse.getAssetIdList();
+                System.out.println(String.format("\nThere are %s assets which currently have an %s status:", assetIdList.size(), status));
+                assetIdList.forEach(id -> System.out.println("'" + id + "'"));
+            } catch (ClassCastException e) {
+                System.out.println(String.format("\nstatus: %s, %s", status, response.getMessage()));
+            }
+        }
+    }
+
+    private static void currentStatusReport(ConfigurableApplicationContext applicationContext) {
+        ReportHandler reportHandler = applicationContext.getBean(ReportHandler.class);
+        List<String> assetIdList = loadFromFile("asset_status_report.txt");
+        for (String assetId : assetIdList) {
+            GeneralResponse response = reportHandler.assetCurrentStatus(assetId);
+            try {
+                AssetCurrentStatusResponse assetCurrentStatusResponse = (AssetCurrentStatusResponse) response;
+                StatusType currentStatus = assetCurrentStatusResponse.getStatusType();
+                System.out.println(String.format("Current status for asset '%s' is '%s'", assetId, currentStatus));
+            } catch (ClassCastException e) {
+                System.out.println(String.format("assetId: %s, %s", assetId, response.getMessage()));
+
+            }
+        }
+    }
+
+    private static List<String> loadFromFile(String fileName) {
+        try {
+            Resource resource = new ClassPathResource("input/" + fileName);
+            InputStream inputStream = resource.getInputStream();
+            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
+            Stream<String> lines = bufferedReader.lines();
+            return lines.collect(Collectors.toList());
+        } catch (IOException e) {
+            return new ArrayList<>();
+        }
+    }
+
+    private static void waitForProcessToFinish(ConfigurableApplicationContext applicationContext) {
+        AssetStatusChangeQueue assetStatusChangeQueue = applicationContext.getBean(AssetStatusChangeQueue.class);
+
+        while (assetStatusChangeQueue.size() > 0) {
+            try {
+                TimeUnit.MILLISECONDS.sleep(10);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+    private static void sendMessages(File file, ConfigurableApplicationContext applicationContext) {
+        try {
+            Gson gson = new Gson();
+            FileReader fileReader = new FileReader(file);
+            JsonParser parser = new JsonParser();
+            Object obj = parser.parse(fileReader);
+            JsonArray jsonArray = (JsonArray) obj;
+            for (JsonElement element : jsonArray) {
+                AssetStatusMessage message = gson.fromJson(element, AssetStatusMessage.class);
+                MessageHandler messageHandler = applicationContext.getBean(MessageHandler.class);
+                messageHandler.assetStatus(message);
+            }
+        } catch (FileNotFoundException e) {
+            System.out.println(String.format("file not found, %s", file));
+        }
+    }
+}

MessageProcessorModule/src/main/java/com/lotus/iot/mpm/api/AbstractController.java

@@ -0,0 +1,34 @@
+package com.lotus.iot.mpm.api;
+
+import com.lotus.iot.mpm.api.objectmodel.response.GeneralResponse;
+import com.lotus.iot.mpm.api.objectmodel.response.GeneralResponseCode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+import org.springframework.web.bind.annotation.ResponseBody;
+import org.springframework.web.bind.annotation.ResponseStatus;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @author Niloufar Mazloumpour
+ * @mail niloufar@mazloumpour.net
+ * @since 2018-05-22
+ */
+@RestController
+public abstract class AbstractController {
+    private static final Logger LOGGER = LoggerFactory.getLogger(AbstractController.class);
+
+    ResponseEntity generateResponse(GeneralResponse response) {
+        return new ResponseEntity<>(response, HttpStatus.valueOf(response.getStatus()));
+    }
+
+    @ExceptionHandler(value = Exception.class)
+    @ResponseBody
+    @ResponseStatus(value = HttpStatus.BAD_REQUEST)
+    public ResponseEntity handleException(Exception e) {
+        LOGGER.error(e.getMessage());
+        return generateResponse(GeneralResponseCode.INVALID_INPUT);
+    }
+}

MessageProcessorModule/src/main/java/com/lotus/iot/mpm/api/ReportController.java

@@ -0,0 +1,38 @@
+package com.lotus.iot.mpm.api;
+
+import com.lotus.iot.mpm.api.objectmodel.response.GeneralResponse;
+import com.lotus.iot.mpm.report.ReportHandler;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @author Niloufar Mazloumpour
+ * @mail niloufar@mazloumpour.net
+ * @since 2018-05-22
+ */
+@RestController
+@RequestMapping("report")
+public class ReportController extends AbstractController {
+    private final ReportHandler reportHandler;
+
+    @Autowired
+    public ReportController(ReportHandler reportHandler) {
+        this.reportHandler = reportHandler;
+    }
+
+    @RequestMapping(value = "/asset/{asset_id}/status", method = RequestMethod.GET)
+    public ResponseEntity assetCurrentStatus(@PathVariable("asset_id") String assetId) {
+        GeneralResponse response = reportHandler.assetCurrentStatus(assetId);
+        return generateResponse(response);
+    }
+
+    @RequestMapping(value = "/asset/status/{status_type}/all", method = RequestMethod.GET)
+    public ResponseEntity assetsInStatus(@PathVariable("status_type") String statusType) {
+        GeneralResponse response = reportHandler.assetsInState(statusType);
+        return generateResponse(response);
+    }
+}

MessageProcessorModule/src/main/java/com/lotus/iot/mpm/api/SubscriptionController.java

@@ -0,0 +1,34 @@
+package com.lotus.iot.mpm.api;
+
+import com.lotus.iot.mpm.api.objectmodel.request.SubscribeRequest;
+import com.lotus.iot.mpm.api.objectmodel.response.GeneralResponseCode;
+import com.lotus.iot.mpm.eventhandler.EventHandlerFacade;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @author Niloufar Mazloumpour
+ * @mail niloufar@mazloumpour.net
+ * @since 2018-05-22
+ */
+@RestController
+@RequestMapping("subscription")
+public class SubscriptionController extends AbstractController {
+    private EventHandlerFacade eventHandlerFacade;
+
+    @Autowired
+    public SubscriptionController(EventHandlerFacade eventHandlerFacade) {
+        this.eventHandlerFacade = eventHandlerFacade;
+    }
+
+    @RequestMapping(value = "asset/event/subscribe", method = RequestMethod.POST, consumes = MediaType.APPLICATION_JSON_VALUE)
+    public ResponseEntity subscribeToAssetStatusEvent(@RequestBody SubscribeRequest subscribeRequest) {
+        eventHandlerFacade.subscribe(subscribeRequest);
+        return generateResponse(GeneralResponseCode.SUCCESS);
+    }
+}

MessageProcessorModule/src/main/java/com/lotus/iot/mpm/api/objectmodel/request/SubscribeRequest.java

@@ -0,0 +1,38 @@
+package com.lotus.iot.mpm.api.objectmodel.request;
+
+import com.lotus.iot.mpm.eventhandler.objectmodel.AssetStatusEvent;
+import com.lotus.iot.mpm.eventhandler.objectmodel.SampleListener;
+import com.lotus.iot.mpm.objectmodel.Observer;
+
+
+/**
+ * @author Niloufar Mazloumpour
+ * @mail niloufar@mazloumpour.net
+ * @since 2018-05-22
+ */
+public class SubscribeRequest {
+    private String assetId;
+    private AssetStatusEvent statusEvent;
+    private SampleListener observer;
+
+    public SubscribeRequest() {
+    }
+
+    public SubscribeRequest(String assetId, AssetStatusEvent statusEvent, SampleListener observer) {
+        this.assetId = assetId;
+        this.statusEvent = statusEvent;
+        this.observer = observer;
+    }
+
+    public String getAssetId() {
+        return assetId;
+    }
+
+    public AssetStatusEvent getStatusEvent() {
+        return statusEvent;
+    }
+
+    public Observer getObserver() {
+        return observer;
+    }
+}