Merge pull request #73 from samson0v/master

Deleted redundant methods

Author: imbeacon

tb_device_mqtt.py

@@ -15,7 +15,6 @@
 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
@@ -738,8 +737,6 @@ 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"])
@@ -761,12 +758,12 @@ def __get_rate_limits_by_topic(self, topic, device=None, msg_rate_limit=None, dp
 
     def __send_publish_with_limitations(self, kwargs, timeout, device=None, msg_rate_limit: RateLimit = None,
                                         dp_rate_limit: RateLimit = None):
-        data_for_analysis = data = kwargs.get("payload")
+        data = kwargs.get("payload")
         if isinstance(data, str):
-            data_for_analysis = loads(data)
+            data = loads(data)
         datapoints = -1
         if dp_rate_limit.has_limit():
-            datapoints = self._count_datapoints_in_message(data_for_analysis, device=device)
+            datapoints = self._count_datapoints_in_message(data, device=device)
         payload = data
         if dp_rate_limit.has_limit() and datapoints >= 0 and dp_rate_limit.get_minimal_limit() < datapoints:
             log.debug("Rate limit is too low, cannot send message with %i datapoints, "
@@ -782,19 +779,19 @@ def __send_publish_with_limitations(self, kwargs, timeout, device=None, msg_rate
                 device_split_messages = self._split_message(device_data, dp_rate_limit.get_minimal_limit(),
                                                             self.max_payload_size)
                 split_messages = [
-                    {'message': {device: [split_message['data']]}, 'datapoints': split_message['datapoints']}
-                                  for split_message in device_split_messages]
+                    {'message': {device: [split_message['data']]}, 'datapoints': split_message['datapoints'],
+                     'metadata': split_message.get('metadata')} for split_message in device_split_messages]
+
             if len(split_messages) == 0:
                 log.debug("Cannot split message to smaller parts!")
+
             results = []
             for part in split_messages:
                 dp_rate_limit.increase_rate_limit_counter(part['datapoints'])
                 self._wait_for_rate_limit_released(timeout,
                                                    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))
@@ -806,8 +803,6 @@ 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))
 
@@ -1043,6 +1038,7 @@ def _split_message(message_pack, max_size, max_payload_size):
                     values = message.get("values")
                 else:
                     values = message
+
                 values_data_keys = tuple(values.keys())
                 if len(values_data_keys) == 1:
                     if ts is not None:
@@ -1072,7 +1068,8 @@ def _split_message(message_pack, max_size, max_payload_size):
                         or current_data_key_index == len(values_data_keys) - 1) or len(
                             str(message_item_values_with_allowed_size)) >= max_payload_size:
                         if ts is not None:
-                            final_message_item['data'] = {"ts": ts, "values": message_item_values_with_allowed_size}
+                            final_message_item['data'] = {"ts": ts, "values": message_item_values_with_allowed_size,
+                                                          'metadata': message.get('metadata')}
                         else:
                             final_message_item['data'] = message_item_values_with_allowed_size
                         final_message_item['datapoints'] = len(message_item_values_with_allowed_size)
@@ -1085,24 +1082,3 @@ 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/metrics'
-            return kwargs
-        except Exception as e:
-            log.error(e)
-            return kwargs