Replace send and recv in connection with a concurrent synchronous implementation of request

mitel_ommclient2/connection.py

@@ -1,8 +1,12 @@
 #!/usr/bin/env python3
 
+import queue
+import select
 import socket
 import ssl
+import threading
 
+from . import messages
 
 class Connection:
     """
@@ -22,9 +26,11 @@ class Connection:
         self._host = host
         self._port = port
         self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        self._recv_buffer = b""
 
-        self._socket.settimeout(3)
+        self._seq = 0 # state of the sequence number generator
+        self._requests = {} # waiting area for pending responses
+
+        self._close = False
 
     def connect(self):
         """
@@ -32,51 +38,104 @@ class Connection:
         """
 
         self._socket.connect((self._host, self._port))
+        self._socket.setblocking(False)
 
-    def send(self, message):
+        threading.Thread(target=self._receive_loop, daemon=True).start()
+
+    def _receive_loop(self):
         """
-            Sends message string
+            Receives messages from socket and associates them to the responding request
 
-            :param message: Message string
+            This function is intended to be executed in thread.
         """
 
+        recv_buffer = b""
+
+        while not self._close:
+            if select.select([self._socket], [], []) != ([], [], []):
+                # wait for data availiable
+                while True:
+                    # fill buffer with one message
+                    data = self._socket.recv(1024)
+
+                    if not data:
+                        # buffer is empty
+                        break
+
+                    recv_buffer += data
+
+                    if b"\0" in recv_buffer:
+                        # there is a full message in buffer, handle that first
+                        break
+
+
+                if b"\0" not in recv_buffer:
+                    # no new messages
+                    break
+
+                # get one message from recv_buffer
+                message, buffer = recv_buffer.split(b"\0", 1)
+                recv_buffer = buffer
+
+                # parse the message
+                message = message.decode("utf-8")
+                response = messages.parse(message)
+
+                if response.seq in self._requests:
+                    # if this response belongs to a request, we return it and resolve the lock
+                    self._requests[response.seq]["response"] = response
+                    self._requests[response.seq]["event"].set()
+
+                # else the message will be ignored
+
+    def _generate_seq(self):
+        """
+            Returns new sequence number
+
+            This generates a number that tries to be unique during a session
+        """
+
+        seq = self._seq
+        self._seq += 1
+        return seq
+
+    def request(self, request):
+        """
+            Sends a request, waits for response and return response
+
+            :param request: Request object
+
+            Usage::
+                >>> r = c.request(mitel_ommclient2.messages.Ping())
+                >>> r.name
+                'PingResp'
+        """
+
+        # generate new sequence number and attach to request
+        seq = self._generate_seq()
+        request.seq = seq
+
+        # add request to waiting area
+        self._requests[seq] = {
+            "event": threading.Event(),
+        }
+
+        # send request
+        message = messages.construct(request)
         self._socket.send(message.encode("utf-8") + b"\0")
 
-    def recv(self):
+        # wait for response
-        """
+        self._requests[seq]["event"].wait()
-            Returns one message
 
-            Use multiple times to receive multiple messages
+        # return reponse and remove from waiting area
-        """
+        return self._requests.pop(seq, {"response": None})["response"]
-
-        data = b""
-        while True:
-            try:
-                new_data = self._socket.recv(1024)
-            except TimeoutError:
-                break
-
-            data += new_data
-
-            if b"\0" in new_data:
-                break
-
-        self._recv_buffer += data
-
-        if b"\0" not in self._recv_buffer:
-            # no new messages
-            return None
-
-        message, buffer = self._recv_buffer.split(b"\0", 1)
-        self._recv_buffer = buffer
-
-        return message.decode("utf-8")
 
     def close(self):
         """
-            Shout down connection
+            Shut down connection
         """
 
+        self._close = True
         return self._socket.close()
 
     def __del__(self):

mitel_ommclient2/messages/__init__.py

@@ -28,6 +28,10 @@ class Request:
     def seq(self):
         return self.attrs.get("seq")
 
+    @seq.setter
+    def seq(self, seq):
+        self.attrs["seq"] = seq
+
 
 class DictRequest(Request):
     """
@@ -79,7 +83,7 @@ class Response:
 
     @property
     def seq(self):
-        return self.attrs.get("seq")
+        return int(self.attrs.get("seq"))
 
     @property
     def errCode(self):

mitel_ommclient2/session.py

@@ -35,17 +35,6 @@ class Session:
 
         self._ensure_connection()
 
-    def _wait_for_respose(self):
-        """
-            Wait until data got received and return message string
-        """
-
-        while True:
-            r = self._connection.recv()
-            if r is not None:
-                return r
-            sleep(0.1)
-
     def _ensure_connection(self):
         """
             Make sure we are connected and logged in
@@ -65,19 +54,4 @@ class Session:
             r.raise_on_error()
 
     def request(self, request):
-        """
+        return self._connection.request(request)
-            Sends a request and waits for response
-
-            :param request: Request object
-
-            Usage::
-                >>> r = s.request(mitel_ommclient2.messages.Ping())
-                >>> r.name
-                'PingResp'
-        """
-
-        message = messages.construct(request)
-        self._connection.send(message)
-
-        res = self._wait_for_respose()
-        return messages.parse(res)