Replace send and recv in connection with a concurrent synchronous implementation of request
This commit is contained in:
parent
8501aab7c3
commit
80fb8cd7cc
@ -27,7 +27,8 @@ class Connection:
|
|||||||
self._port = port
|
self._port = port
|
||||||
self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||||
|
|
||||||
self._received_messages = queue.Queue()
|
self._seq = 0 # state of the sequence number generator
|
||||||
|
self._requests = {} # waiting area for pending responses
|
||||||
|
|
||||||
self._close = False
|
self._close = False
|
||||||
|
|
||||||
@ -41,18 +42,13 @@ class Connection:
|
|||||||
|
|
||||||
threading.Thread(target=self._receive_loop, daemon=True).start()
|
threading.Thread(target=self._receive_loop, daemon=True).start()
|
||||||
|
|
||||||
def send(self, message):
|
|
||||||
"""
|
|
||||||
Sends message string
|
|
||||||
|
|
||||||
:param message: Message string
|
|
||||||
"""
|
|
||||||
|
|
||||||
message = messages.construct(message)
|
|
||||||
|
|
||||||
self._socket.send(message.encode("utf-8") + b"\0")
|
|
||||||
|
|
||||||
def _receive_loop(self):
|
def _receive_loop(self):
|
||||||
|
"""
|
||||||
|
Receives messages from socket and associates them to the responding request
|
||||||
|
|
||||||
|
This function is intended to be executed in thread.
|
||||||
|
"""
|
||||||
|
|
||||||
recv_buffer = b""
|
recv_buffer = b""
|
||||||
|
|
||||||
while not self._close:
|
while not self._close:
|
||||||
@ -63,11 +59,13 @@ class Connection:
|
|||||||
data = self._socket.recv(1024)
|
data = self._socket.recv(1024)
|
||||||
|
|
||||||
if not data:
|
if not data:
|
||||||
|
# buffer is empty
|
||||||
break
|
break
|
||||||
|
|
||||||
recv_buffer += data
|
recv_buffer += data
|
||||||
|
|
||||||
if b"\0" in recv_buffer:
|
if b"\0" in recv_buffer:
|
||||||
|
# there is a full message in buffer, handle that first
|
||||||
break
|
break
|
||||||
|
|
||||||
|
|
||||||
@ -75,24 +73,62 @@ class Connection:
|
|||||||
# no new messages
|
# no new messages
|
||||||
break
|
break
|
||||||
|
|
||||||
|
# get one message from recv_buffer
|
||||||
message, buffer = recv_buffer.split(b"\0", 1)
|
message, buffer = recv_buffer.split(b"\0", 1)
|
||||||
recv_buffer = buffer
|
recv_buffer = buffer
|
||||||
|
|
||||||
|
# parse the message
|
||||||
message = message.decode("utf-8")
|
message = message.decode("utf-8")
|
||||||
message = messages.parse(message)
|
response = messages.parse(message)
|
||||||
self._received_messages.put(message)
|
|
||||||
|
|
||||||
def recv(self):
|
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 one message
|
Returns new sequence number
|
||||||
|
|
||||||
Use multiple times to receive multiple messages
|
This generates a number that tries to be unique during a session
|
||||||
"""
|
"""
|
||||||
|
|
||||||
if self._received_messages.empty():
|
seq = self._seq
|
||||||
return None
|
self._seq += 1
|
||||||
|
return seq
|
||||||
|
|
||||||
return self._received_messages.get()
|
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")
|
||||||
|
|
||||||
|
# wait for response
|
||||||
|
self._requests[seq]["event"].wait()
|
||||||
|
|
||||||
|
# return reponse and remove from waiting area
|
||||||
|
return self._requests.pop(seq, {"response": None})["response"]
|
||||||
|
|
||||||
def close(self):
|
def close(self):
|
||||||
"""
|
"""
|
||||||
|
@ -28,6 +28,10 @@ class Request:
|
|||||||
def seq(self):
|
def seq(self):
|
||||||
return self.attrs.get("seq")
|
return self.attrs.get("seq")
|
||||||
|
|
||||||
|
@seq.setter
|
||||||
|
def seq(self, seq):
|
||||||
|
self.attrs["seq"] = seq
|
||||||
|
|
||||||
|
|
||||||
class DictRequest(Request):
|
class DictRequest(Request):
|
||||||
"""
|
"""
|
||||||
@ -79,7 +83,7 @@ class Response:
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def seq(self):
|
def seq(self):
|
||||||
return self.attrs.get("seq")
|
return int(self.attrs.get("seq"))
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def errCode(self):
|
def errCode(self):
|
||||||
|
@ -35,17 +35,6 @@ class Session:
|
|||||||
|
|
||||||
self._ensure_connection()
|
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):
|
def _ensure_connection(self):
|
||||||
"""
|
"""
|
||||||
Make sure we are connected and logged in
|
Make sure we are connected and logged in
|
||||||
@ -65,17 +54,4 @@ class Session:
|
|||||||
r.raise_on_error()
|
r.raise_on_error()
|
||||||
|
|
||||||
def request(self, request):
|
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'
|
|
||||||
"""
|
|
||||||
|
|
||||||
self._connection.send(request)
|
|
||||||
|
|
||||||
return self._wait_for_respose()
|
|
||||||
|
Loading…
Reference in New Issue
Block a user