Fix mqtt client

agrirouter/auth/dto.py

@@ -4,44 +4,6 @@
 from agrirouter.messaging.exceptions import WrongFieldError
 
 
-class AuthorizationResultUrl:
-    def __init__(self,
-                 *,
-                 state: str = None,
-                 signature: str = None,
-                 token: str = None,
-                 error: str = None
-                 ):
-        self.state = state
-        self.signature = signature
-        self.token = token
-        self.error = error
-
-    def get_state(self) -> str:
-        return self.state
-
-    def set_state(self, state: str) -> None:
-        self.state = state
-
-    def get_signature(self) -> str:
-        return self.signature
-
-    def set_signature(self, signature: str) -> None:
-        self.signature = signature
-
-    def get_token(self) -> str:
-        return self.token
-
-    def set_token(self, token: str) -> None:
-        self.token = token
-
-    def get_error(self) -> str:
-        return self.error
-
-    def set_error(self, error: str) -> None:
-        self.error = error
-
-
 class AuthorizationToken:
     ACCOUNT = 'account'
     REGISTRATION_CODE = 'regcode'
@@ -88,6 +50,52 @@ def set_expires(self, expires: str) -> None:
         self.expires = expires
 
 
+class AuthorizationResultUrl:
+    def __init__(self,
+                 *,
+                 state: str = None,
+                 signature: str = None,
+                 token: str = None,
+                 decoded_token: AuthorizationToken = None,
+                 error: str = None
+                 ):
+        self.state = state
+        self.signature = signature
+        self.token = token
+        self.decoded_token = decoded_token
+        self.error = error
+
+    def get_state(self) -> str:
+        return self.state
+
+    def set_state(self, state: str) -> None:
+        self.state = state
+
+    def get_signature(self) -> str:
+        return self.signature
+
+    def set_signature(self, signature: str) -> None:
+        self.signature = signature
+
+    def get_token(self) -> str:
+        return self.token
+
+    def set_token(self, token: str) -> None:
+        self.token = token
+
+    def get_error(self) -> str:
+        return self.error
+
+    def set_error(self, error: str) -> None:
+        self.error = error
+
+    def get_decoded_token(self) -> AuthorizationToken:
+        return self.decoded_token
+
+    def set_decoded_token(self, decoded_token: AuthorizationToken) -> None:
+        self.decoded_token = decoded_token
+
+
 class AuthorizationResult:
     def __init__(self,
                  *,

agrirouter/auth/response.py

@@ -5,7 +5,7 @@
 
 from cryptography.exceptions import InvalidSignature
 
-from agrirouter.auth.dto import AuthorizationToken
+from agrirouter.auth.dto import AuthorizationToken, AuthorizationResultUrl
 from agrirouter.onboarding.signature import verify_signature
 
 
@@ -69,16 +69,16 @@ def decode_token(token: Union[str, bytes]) -> AuthorizationToken:
         auth_token.json_deserialize(json.loads(decoded_token))
         return auth_token
 
-    def get_auth_result(self) -> dict:
-        if not self.is_successful:
-            return {self.ERROR_KEY: self.error}
+    def get_auth_result(self) -> AuthorizationResultUrl:
         decoded_token = self.decode_token(self.token)
-        return {
-            self.SIGNATURE_KEY: self.signature,
-            self.STATE_KEY: self.state,
-            self.TOKEN_KEY: self.token,
-            self.CRED_KEY: decoded_token
-        }
+
+        return AuthorizationResultUrl(
+            signature=self.signature,
+            state=self.state,
+            token=self.token,
+            decoded_token=decoded_token,
+            error=self.error
+        )
 
     def get_signature(self):
         return self.signature

agrirouter/messaging/clients/constants.py

@@ -0,0 +1,2 @@
+ASYNC = "ASYNC"
+SYNC = "SYNC"

agrirouter/messaging/clients/mqtt.py

@@ -1,16 +1,22 @@
+import time
+import ssl
 from typing import Any, List, Tuple
 
-from paho.mqtt import client as mqtt_client
+import paho.mqtt.client as mqtt_client
 from paho.mqtt.client import MQTTv31, MQTTMessageInfo
 
+from agrirouter.messaging.certification import create_certificate_file_from_pen
+from agrirouter.messaging.clients.constants import SYNC, ASYNC
+
 
 class MqttClient:
 
     def __init__(self,
-                 client_id,
+                 onboard_response,
+                 client_id: str,
                  on_message_callback: callable = None,
                  userdata: Any = None,
-                 clean_session: bool = True
+                 clean_session: bool = False
                  ):
         # TODO: Implement on_message_callback parameter validation:
         #  must take params as described at https://pypi.org/project/paho-mqtt/#callbacks
@@ -22,26 +28,54 @@ def __init__(self,
             protocol=MQTTv31,
             transport="tcp"
         )
+
         self.mqtt_client.on_message = on_message_callback if on_message_callback else self._get_on_message_callback()
-        self.mqtt_client.on_connect = self._get_on_connect_callback()
+        self.mqtt_client.on_connect = self._get_on_connect_callback(onboard_response)
         self.mqtt_client.on_disconnect = self._get_on_disconnect_callback()
         self.mqtt_client.on_subscribe = self._get_on_subscribe_callback()
         self.mqtt_client.on_unsubscribe = self._get_on_unsubscribe_callback()
 
+        certificate_file_path = create_certificate_file_from_pen(onboard_response)
+        context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
+        context.load_cert_chain(
+            certfile=certificate_file_path,
+            keyfile=certificate_file_path,
+            password=onboard_response.get_authentication().get_secret(),
+        )
+        self.mqtt_client.tls_set_context(context)
+
+        self._mode = None
+
     def connect(self, host: str, port: str) -> None:
+        self.mqtt_client.connect(
+            host=host,
+            port=int(port)
+        )
+        self.mqtt_client.loop()
+
+        self._mode = SYNC
+
+    def connect_async(self, host: str, port: str):
         self.mqtt_client.connect_async(
             host=host,
-            port=port
+            port=int(port)
         )
         self.mqtt_client.loop_start()
 
+        self._mode = ASYNC
+
+        while self.mqtt_client._state == 0:
+            time.sleep(1)
+
     def disconnect(self):
         self.mqtt_client.loop_stop()
         self.mqtt_client.disconnect()
 
-    def publish(self, topic, payload, qos=0) -> MQTTMessageInfo:
-        """
+    def receive_outbox_messages(self):
+        self.mqtt_client.loop()
 
+    def publish(self, topic, payload, qos=2) -> MQTTMessageInfo:
+        """
         :param topic: str representing unique name of the topic that the message should be published on
         :param payload: The actual message to send
         :param qos: int representing the quality of service level to use. May be [0, 1, 2]
@@ -52,6 +86,10 @@ def publish(self, topic, payload, qos=0) -> MQTTMessageInfo:
             payload=payload,
             qos=qos
         )
+        if self._mode == SYNC:
+            self.mqtt_client.loop()
+            time.sleep(3)
+            self.mqtt_client.loop()
         return message_info
 
     def subscribe(self, topics: List[Tuple[str, int]]) -> tuple:
@@ -67,7 +105,7 @@ def subscribe(self, topics: List[Tuple[str, int]]) -> tuple:
 
         :return: tuple
         """
-        result, mid = self.mqtt_client.subscribe(topics)
+        result, mid = self.mqtt_client.subscribe(topics, qos=2)
         return result, mid
 
     def unsubscribe(self, topics: List[str]) -> tuple:
@@ -85,61 +123,43 @@ def unsubscribe(self, topics: List[str]) -> tuple:
         return result, mid
 
     @staticmethod
-    def _get_on_connect_callback() -> callable:
-
-        def on_connect(client, userdata, flags, rc, properties=None):
-            print("Connection started")
-            with open("connection.txt", "w") as file:
-                file.write("Connection started")
-                if rc == 0:
-                    file.write("Connected!!")
-                else:
-                    file.write("Do not Connected!!")
-            if rc == 0:
-                print("Connected to MQTT Broker!")
-            else:
-                print(f"Failed to connect, return code: {rc}")
+    def _get_on_connect_callback(onboard_response) -> callable:
 
-            return client, userdata, flags, rc, properties
+        def on_connect(client: mqtt_client.Client, userdata, flags, rc, properties=None):
+            if rc == 0:
+                client.subscribe(topic=onboard_response.connection_criteria.commands)
+                time.sleep(3)
 
         return on_connect
 
     @staticmethod
     def _get_on_message_callback() -> callable:
 
         def on_message(client, userdata, msg):
-            # print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")
-
             return client, userdata, msg
 
         return on_message
 
     @staticmethod
     def _get_on_subscribe_callback() -> callable:
 
-        def on_subscribe(client, userdata, mid, granted_qos, properties=None):
-            # print(f"Subscribed {userdata} to `{properties}`")
-
-            return client, userdata, mid, granted_qos, properties
+        def on_subscribe(*args, **kwargs):
+            return args, kwargs
 
         return on_subscribe
 
     @staticmethod
     def _get_on_disconnect_callback() -> callable:
 
-        def on_disconnect(client, userdata, rc):
-            # print(f"Disconnected from from `{properties}`")
-
-            return client, userdata, rc
+        def on_disconnect(*args, **kwargs):
+            return args, kwargs
 
         return on_disconnect
 
     @staticmethod
     def _get_on_unsubscribe_callback() -> callable:
 
-        def on_unsubscribe(client, userdata, mid):
-            # print(f"Unsubscribed `{userdata}` from `{properties}`")
-
-            return client, userdata, mid
+        def on_unsubscribe(*args, **kwargs):
+            return args, kwargs
 
         return on_unsubscribe

agrirouter/messaging/services/commons.py

@@ -11,6 +11,7 @@
 from agrirouter.messaging.request import MessageRequest
 from agrirouter.messaging.result import MessagingResult
 from agrirouter.onboarding.exceptions import BadMessagingResult
+from agrirouter.onboarding.response import SoftwareOnboardingResponse
 
 
 class AbstractMessagingClient(ABC):
@@ -50,33 +51,35 @@ def send(self, parameters) -> MessagingResult:
 class MqttMessagingService(AbstractMessagingClient):
 
     def __init__(self,
-                 client_id,
-                 onboarding_response,
+                 onboarding_response: SoftwareOnboardingResponse,
                  on_message_callback: callable = None,
+                 client_async: bool = True
                  ):
 
         self.onboarding_response = onboarding_response
         self.client = MqttClient(
-            client_id=client_id,
+            onboard_response=onboarding_response,
+            client_id=onboarding_response.get_connection_criteria().get_client_id(),
             on_message_callback=on_message_callback,
         )
-        self.client.connect(
-            self.onboarding_response.get_connection_criteria().get_host(),
-            self.onboarding_response.get_connection_criteria().get_port()
-        )
+        if client_async:
+            self.client.connect_async(
+                self.onboarding_response.get_connection_criteria().get_host(),
+                self.onboarding_response.get_connection_criteria().get_port()
+            )
+        else:
+            self.client.connect(
+                self.onboarding_response.get_connection_criteria().get_host(),
+                self.onboarding_response.get_connection_criteria().get_port()
+            )
 
     def send(self, parameters, qos: int = 0) -> MessagingResult:
         message_request = self.create_message_request(parameters)
         mqtt_payload = message_request.json_serialize()
         self.client.publish(
-            self.onboarding_response.get_connection_criteria().get_measures(), json.dumps(mqtt_payload),
+            topic=self.onboarding_response.get_connection_criteria().get_measures(),
+            payload=json.dumps(mqtt_payload),
             qos=qos
         )
         result = MessagingResult([parameters.get_application_message_id()])
         return result
-
-    def subscribe(self):
-        pass
-
-    def unsubscribe(self):
-        pass