Merge pull request #70 from samson0v/master

Added latency check methods

Author: imbeacon

tb_device_mqtt.py

@@ -15,6 +15,7 @@
 import logging
 from inspect import signature
 from time import sleep
+from time import time as timestamp
 
 import paho.mqtt.client as paho
 from math import ceil
@@ -737,8 +738,12 @@ def _send_request(self, _type, kwargs, timeout=DEFAULT_TIMEOUT, device=None,
             if msg_rate_limit.has_limit():
                 return self.__send_publish_with_limitations(kwargs, timeout, device, msg_rate_limit, dp_rate_limit)
             else:
+                if self.__is_test_latency_message(kwargs['payload']):
+                    kwargs = self.__convert_test_latency_message(kwargs)
+
                 if "payload" in kwargs and not isinstance(kwargs["payload"], str):
                     kwargs["payload"] = dumps(kwargs["payload"])
+
                 return TBPublishInfo(self._client.publish(**kwargs))
         elif _type == TBSendMethod.SUBSCRIBE:
             return self._client.subscribe(**kwargs)
@@ -788,6 +793,8 @@ def __send_publish_with_limitations(self, kwargs, timeout, device=None, msg_rate
                                                    message_rate_limit=msg_rate_limit,
                                                    dp_rate_limit=dp_rate_limit,
                                                    amount=dp_rate_limit.get_minimal_limit())
+                if self.__is_test_latency_message(kwargs['payload']):
+                    kwargs = self.__convert_test_latency_message(kwargs)
                 kwargs["payload"] = dumps(part['message'])
                 self.wait_until_current_queued_messages_processed()
                 results.append(self._client.publish(**kwargs))
@@ -799,6 +806,8 @@ def __send_publish_with_limitations(self, kwargs, timeout, device=None, msg_rate
                                                    message_rate_limit=msg_rate_limit,
                                                    dp_rate_limit=dp_rate_limit,
                                                    amount=datapoints)
+            if self.__is_test_latency_message(kwargs['payload']):
+                kwargs = self.__convert_test_latency_message(kwargs)
             kwargs["payload"] = dumps(payload)
             return TBPublishInfo(self._client.publish(**kwargs))
 
@@ -1076,3 +1085,24 @@ def _split_message(message_pack, max_size, max_payload_size):
         if add_last_item:
             split_messages.append(final_message_item)
         return split_messages
+
+    @staticmethod
+    def __is_test_latency_message(payload):
+        if isinstance(payload, list) and payload[0].get('values', {}).get('isTestLatencyMessageType', False):
+            return True
+
+        return False
+
+    @staticmethod
+    def __convert_test_latency_message(kwargs):
+        try:
+            values = kwargs['payload'][0]['values']
+            payload = {
+                values['connectorName']: {'receivedTs': values['receivedTs'], 'publishedTs': int(timestamp() * 1000)}}
+
+            kwargs['payload'] = payload
+            kwargs['topic'] = 'v1/gateway/latency'
+            return kwargs
+        except Exception as e:
+            log.error(e)
+            return kwargs