Merge pull request #1015 from hashmapinc/Tempus-947-Publish-data-qual…

…ity-notification-to-events

Tempus 947 publish data quality notification to events

common/message/src/main/java/com/hashmapinc/server/common/msg/session/MsgType.java

@@ -25,6 +25,7 @@ public enum MsgType {
     SPARKPLUG_TELEMETRY_UPDATE_NOTIFICATION,
     SPARKPLUG_SUBSCRIBE_TERMINATE,
     SPARKPLUG_DEATH_SUBSCRIBE,
+    POST_DEVICE_EVENT,
 
     SUBSCRIBE_RPC_COMMANDS_REQUEST, UNSUBSCRIBE_RPC_COMMANDS_REQUEST,
     TO_DEVICE_RPC_REQUEST, TO_DEVICE_RPC_RESPONSE, TO_DEVICE_RPC_RESPONSE_ACK,

transport/mqtt/src/main/java/com/hashmapinc/server/transport/mqtt/MqttTopics.java

@@ -32,6 +32,7 @@ public class MqttTopics {
     public static final String DEVICE_TELEMETRY_TOPIC = BASE_DEVICE_API_TOPIC + "/telemetry";
     public static final String DEVICE_ATTRIBUTES_TOPIC = BASE_DEVICE_API_TOPIC + "/attributes";
     public static final String DEVICE_DEPTH_TELEMETRY_TOPIC = BASE_DEVICE_API_TOPIC + "/depth/telemetry";
+    public static final String DEVICE_EVENT_TOPIC = BASE_DEVICE_API_TOPIC + "/events";
 
     public static final String BASE_GATEWAY_API_TOPIC = "v1/gateway";
     public static final String GATEWAY_CONNECT_TOPIC = BASE_GATEWAY_API_TOPIC + "/connect";
@@ -42,6 +43,7 @@ public class MqttTopics {
     public static final String GATEWAY_RPC_TOPIC = BASE_GATEWAY_API_TOPIC + "/rpc";
     public static final String GATEWAY_ATTRIBUTES_REQUEST_TOPIC = BASE_GATEWAY_API_TOPIC + "/attributes/request";
     public static final String GATEWAY_ATTRIBUTES_RESPONSE_TOPIC = BASE_GATEWAY_API_TOPIC + "/attributes/response";
+    public static final String GATEWAY_EVENTS_TOPIC = BASE_GATEWAY_API_TOPIC + "/events";
 
 
     private MqttTopics() {

transport/mqtt/src/main/java/com/hashmapinc/server/transport/mqtt/MqttTransportHandler.java

@@ -17,7 +17,9 @@
 package com.hashmapinc.server.transport.mqtt;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.hashmapinc.server.common.data.Device;
+import com.hashmapinc.server.common.data.Event;
 import com.hashmapinc.server.common.data.security.DeviceTokenCredentials;
 import com.hashmapinc.server.common.data.security.DeviceX509Credentials;
 import com.hashmapinc.server.common.msg.session.AdaptorToSessionActorMsg;
@@ -32,28 +34,30 @@
 import com.hashmapinc.server.dao.asset.AssetService;
 import com.hashmapinc.server.dao.attributes.AttributesService;
 import com.hashmapinc.server.dao.device.DeviceService;
+import com.hashmapinc.server.dao.event.EventService;
 import com.hashmapinc.server.dao.mail.MailService;
 import com.hashmapinc.server.dao.relation.RelationService;
+import com.hashmapinc.server.transport.mqtt.adaptors.JsonMqttAdaptor;
 import com.hashmapinc.server.transport.mqtt.adaptors.MqttTransportAdaptor;
 import com.hashmapinc.server.transport.mqtt.session.DeviceSessionCtx;
 import com.hashmapinc.server.transport.mqtt.session.GatewaySessionCtx;
+import com.hashmapinc.server.transport.mqtt.sparkplug.SparkPlugDecodeService;
 import com.hashmapinc.server.transport.mqtt.sparkplug.SparkPlugMsgTypes;
+import com.hashmapinc.server.transport.mqtt.sparkplug.SparkPlugUtils;
+import com.hashmapinc.server.transport.mqtt.sparkplug.data.SparkPlugMetaData;
 import com.hashmapinc.server.transport.mqtt.util.SslUtil;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.handler.codec.mqtt.*;
-import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
-import io.netty.handler.codec.mqtt.MqttQoS;
 import io.netty.handler.ssl.SslHandler;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.util.StringUtils;
-import com.hashmapinc.server.transport.mqtt.sparkplug.data.SparkPlugMetaData;
-import com.hashmapinc.server.transport.mqtt.sparkplug.SparkPlugDecodeService;
-import com.hashmapinc.server.transport.mqtt.sparkplug.SparkPlugUtils;
+
 import javax.net.ssl.SSLPeerUnverifiedException;
 import javax.security.cert.X509Certificate;
+import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.List;
@@ -83,14 +87,16 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
     private final AssetService assetService;
     private final SslHandler sslHandler;
     private final MailService mailService;
+    private final EventService eventService;
     private static final String SPARK_PLUG_NAME_SPACE = "spBv1.0";
     private volatile boolean connected;
     private volatile GatewaySessionCtx gatewaySessionCtx;
     private SparkPlugDecodeService sparkPlugDecodeService;
     private SparkPlugUtils sparkPlugUtils;
 
     public MqttTransportHandler(SessionMsgProcessor processor, DeviceService deviceService, DeviceAuthService authService, RelationService relationService,
-                                MqttTransportAdaptor adaptor, SslHandler sslHandler, QuotaService quotaService , AttributesService attributesService ,AssetService assetService, MailService mailService) {
+                                MqttTransportAdaptor adaptor, SslHandler sslHandler, QuotaService quotaService , AttributesService attributesService,
+                                AssetService assetService, EventService eventService, MailService mailService) {
         this.processor = processor;
         this.deviceService = deviceService;
         this.relationService = relationService;
@@ -103,6 +109,7 @@ public MqttTransportHandler(SessionMsgProcessor processor, DeviceService deviceS
         this.attributesService = attributesService;
         this.assetService = assetService;
         this.mailService = mailService;
+        this.eventService = eventService;
     }
 
     @Override
@@ -184,8 +191,10 @@ private void handleMqttPublishMsg(String topicName, int msgId, MqttPublishMessag
         try {
             if (topicName.equals(MqttTopics.GATEWAY_TELEMETRY_TOPIC)) {
                 gatewaySessionCtx.onDeviceTelemetry(mqttMsg);
-            }else if (topicName.equals(MqttTopics.GATEWAY_DEPTH_TELEMETRY_TOPIC)) {
+            } else if (topicName.equals(MqttTopics.GATEWAY_DEPTH_TELEMETRY_TOPIC)) {
                 gatewaySessionCtx.onDeviceDepthTelemetry(mqttMsg);
+            } else if (topicName.equals(MqttTopics.GATEWAY_EVENTS_TOPIC)) {
+                gatewaySessionCtx.onDeviceEventMsg(mqttMsg);
             } else if (topicName.equals(MqttTopics.GATEWAY_ATTRIBUTES_TOPIC)) {
                 gatewaySessionCtx.onDeviceAttributes(mqttMsg);
             } else if (topicName.equals(MqttTopics.GATEWAY_ATTRIBUTES_REQUEST_TOPIC)) {
@@ -207,11 +216,12 @@ private void processDevicePublish(ChannelHandlerContext ctx, MqttPublishMessage
         try {
             if (topicName.equals(MqttTopics.DEVICE_TELEMETRY_TOPIC)) {
                 msg = adaptor.convertToActorMsg(deviceSessionCtx, MsgType.POST_TELEMETRY_REQUEST, mqttMsg);
-            }else if(topicName.equals(MqttTopics.DEVICE_DEPTH_TELEMETRY_TOPIC)){
+            } else if (topicName.equals(MqttTopics.DEVICE_DEPTH_TELEMETRY_TOPIC)){
                 msg = adaptor.convertToActorMsg(deviceSessionCtx, MsgType.POST_TELEMETRY_REQUEST_DEPTH, mqttMsg);
-            }
-            else if (topicName.equals(MqttTopics.DEVICE_ATTRIBUTES_TOPIC)) {
+            } else if (topicName.equals(MqttTopics.DEVICE_ATTRIBUTES_TOPIC)) {
                 msg = adaptor.convertToActorMsg(deviceSessionCtx, MsgType.POST_ATTRIBUTES_REQUEST, mqttMsg);
+            } else if (topicName.equals(MqttTopics.DEVICE_EVENT_TOPIC)) {
+                saveEventForDevice(mqttMsg);
             } else if (topicName.startsWith(MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX)) {
                 msg = adaptor.convertToActorMsg(deviceSessionCtx, MsgType.GET_ATTRIBUTES_REQUEST, mqttMsg);
                 if (msgId >= 0) {
@@ -239,6 +249,25 @@ else if (topicName.equals(MqttTopics.DEVICE_ATTRIBUTES_TOPIC)) {
         }
     }
 
+    private void saveEventForDevice(MqttPublishMessage inbound) throws AdaptorException {
+        String payload = JsonMqttAdaptor.validatePayload(deviceSessionCtx.getSessionId(), inbound.payload());
+        try {
+            ObjectMapper mapper = new ObjectMapper();
+            JsonNode eventInfo = mapper.readTree(payload);
+            Event event = new Event();
+            Device device = deviceSessionCtx.getDevice();
+            event.setEntityId(device.getId());
+            event.setTenantId(device.getTenantId());
+            event.setType("QUALITY_EVENT");
+            event.setBody(eventInfo);
+            eventService.save(event);
+
+        } catch (IOException ex) {
+            log.info("Execption occurred : {}", ex);
+            throw new AdaptorException(ex);
+        }
+    }
+
     private void processSubscribe(ChannelHandlerContext ctx, MqttSubscribeMessage mqttMsg) {
         if (!checkConnected(ctx)) {
             return;
@@ -459,7 +488,8 @@ private void checkGatewaySession(MqttConnectMessage msg) {
             JsonNode gatewayNode = infoNode.get("gateway");
             JsonNode topic = infoNode.get(TOPIC);
             if (gatewayNode != null && gatewayNode.asBoolean()) {
-                gatewaySessionCtx = new GatewaySessionCtx(processor, deviceService, authService, relationService, deviceSessionCtx,attributesService,assetService, mailService);
+                gatewaySessionCtx = new GatewaySessionCtx(processor, deviceService, authService, relationService, eventService,
+                        deviceSessionCtx, attributesService, assetService, mailService);
                 if((msg.payload().willTopic() != null) && msg.payload().willTopic().startsWith(SPARK_PLUG_NAME_SPACE)){
                     sparkPlugDecodeService = new SparkPlugDecodeService();
                 }

transport/mqtt/src/main/java/com/hashmapinc/server/transport/mqtt/MqttTransportServerInitializer.java

@@ -16,21 +16,22 @@
  */
 package com.hashmapinc.server.transport.mqtt;
 
+import com.hashmapinc.server.common.transport.SessionMsgProcessor;
+import com.hashmapinc.server.common.transport.auth.DeviceAuthService;
+import com.hashmapinc.server.common.transport.quota.QuotaService;
 import com.hashmapinc.server.dao.asset.AssetService;
 import com.hashmapinc.server.dao.attributes.AttributesService;
+import com.hashmapinc.server.dao.device.DeviceService;
+import com.hashmapinc.server.dao.event.EventService;
 import com.hashmapinc.server.dao.mail.MailService;
+import com.hashmapinc.server.dao.relation.RelationService;
+import com.hashmapinc.server.transport.mqtt.adaptors.MqttTransportAdaptor;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.mqtt.MqttDecoder;
 import io.netty.handler.codec.mqtt.MqttEncoder;
 import io.netty.handler.ssl.SslHandler;
-import com.hashmapinc.server.common.transport.SessionMsgProcessor;
-import com.hashmapinc.server.common.transport.auth.DeviceAuthService;
-import com.hashmapinc.server.common.transport.quota.QuotaService;
-import com.hashmapinc.server.dao.device.DeviceService;
-import com.hashmapinc.server.dao.relation.RelationService;
-import com.hashmapinc.server.transport.mqtt.adaptors.MqttTransportAdaptor;
 
 
 public class MqttTransportServerInitializer extends ChannelInitializer<SocketChannel> {
@@ -47,10 +48,11 @@ public class MqttTransportServerInitializer extends ChannelInitializer<SocketCha
     private final AttributesService attributesService;
     private final AssetService assetService;
     private final MailService mailService;
+    private final EventService eventService;
 
     public MqttTransportServerInitializer(SessionMsgProcessor processor, DeviceService deviceService, DeviceAuthService authService, RelationService relationService,
-                                          MqttTransportAdaptor adaptor, MqttSslHandlerProvider sslHandlerProvider,
-                                          QuotaService quotaService, AttributesService attributesService ,AssetService assetService,MailService mailService) {
+                                          MqttTransportAdaptor adaptor, MqttSslHandlerProvider sslHandlerProvider, QuotaService quotaService,
+                                          AttributesService attributesService ,AssetService assetService, EventService eventService, MailService mailService) {
         this.processor = processor;
         this.deviceService = deviceService;
         this.authService = authService;
@@ -61,6 +63,7 @@ public MqttTransportServerInitializer(SessionMsgProcessor processor, DeviceServi
         this.attributesService = attributesService;
         this.assetService = assetService;
         this.mailService = mailService;
+        this.eventService = eventService;
     }
 
     @Override
@@ -75,7 +78,7 @@ public void initChannel(SocketChannel ch) {
         pipeline.addLast("encoder", MqttEncoder.INSTANCE);
 
         MqttTransportHandler handler = new MqttTransportHandler(processor, deviceService, authService, relationService,
-                adaptor, sslHandler, quotaService, attributesService, assetService, mailService);
+                adaptor, sslHandler, quotaService, attributesService, assetService, eventService, mailService);
 
         pipeline.addLast(handler);
         ch.closeFuture().addListener(handler);

transport/mqtt/src/main/java/com/hashmapinc/server/transport/mqtt/MqttTransportService.java

@@ -16,9 +16,16 @@
  */
 package com.hashmapinc.server.transport.mqtt;
 
+import com.hashmapinc.server.common.transport.SessionMsgProcessor;
+import com.hashmapinc.server.common.transport.auth.DeviceAuthService;
+import com.hashmapinc.server.common.transport.quota.QuotaService;
 import com.hashmapinc.server.dao.asset.AssetService;
 import com.hashmapinc.server.dao.attributes.AttributesService;
+import com.hashmapinc.server.dao.device.DeviceService;
+import com.hashmapinc.server.dao.event.EventService;
 import com.hashmapinc.server.dao.mail.MailService;
+import com.hashmapinc.server.dao.relation.RelationService;
+import com.hashmapinc.server.transport.mqtt.adaptors.MqttTransportAdaptor;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.Channel;
 import io.netty.channel.EventLoopGroup;
@@ -31,12 +38,6 @@
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.ApplicationContext;
 import org.springframework.stereotype.Service;
-import com.hashmapinc.server.common.transport.SessionMsgProcessor;
-import com.hashmapinc.server.common.transport.auth.DeviceAuthService;
-import com.hashmapinc.server.common.transport.quota.QuotaService;
-import com.hashmapinc.server.dao.device.DeviceService;
-import com.hashmapinc.server.dao.relation.RelationService;
-import com.hashmapinc.server.transport.mqtt.adaptors.MqttTransportAdaptor;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
@@ -62,6 +63,9 @@ public class MqttTransportService {
     @Autowired(required = false)
     private RelationService relationService;
 
+    @Autowired(required = false)
+    private EventService eventService;
+
     @Autowired(required = false)
     private MqttSslHandlerProvider sslHandlerProvider;
 
@@ -112,7 +116,7 @@ public void init() throws InterruptedException {
         b.group(bossGroup, workerGroup)
                 .channel(NioServerSocketChannel.class)
                 .childHandler(new MqttTransportServerInitializer(processor, deviceService, authService, relationService,
-                        adaptor, sslHandlerProvider, quotaService,attributesService, assetService,mailService));
+                        adaptor, sslHandlerProvider, quotaService,attributesService, assetService, eventService, mailService));
 
         serverChannel = b.bind(host, port).sync().channel();
         log.info("Mqtt transport started!");

transport/mqtt/src/main/java/com/hashmapinc/server/transport/mqtt/adaptors/JsonMqttAdaptor.java

@@ -29,20 +29,16 @@
 import com.hashmapinc.server.common.transport.adaptor.AdaptorException;
 import com.hashmapinc.server.common.transport.adaptor.JsonConverter;
 import com.hashmapinc.server.transport.mqtt.MqttTopics;
+import com.hashmapinc.server.transport.mqtt.MqttTransportHandler;
 import com.hashmapinc.server.transport.mqtt.session.DeviceSessionCtx;
+import com.hashmapinc.server.transport.mqtt.sparkplug.SparkPlugEncodeService;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.UnpooledByteBufAllocator;
-import io.netty.handler.codec.mqtt.MqttPublishMessage;
-import io.netty.handler.codec.mqtt.MqttMessage;
-import io.netty.handler.codec.mqtt.MqttFixedHeader;
-import io.netty.handler.codec.mqtt.MqttQoS;
-import io.netty.handler.codec.mqtt.MqttMessageType;
-import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
+import io.netty.handler.codec.mqtt.*;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
-import com.hashmapinc.server.transport.mqtt.MqttTransportHandler;
-import com.hashmapinc.server.transport.mqtt.sparkplug.SparkPlugEncodeService;
+
 import java.nio.charset.Charset;
 import java.util.*;