WIP to reorganise to microservices

[loctrkd.git] / gps303 / collector.py

diff --git a/gps303/collector.py b/gps303/collector.py
index 28c636d33642764118067bcadf0355377999b2c2..7ffa7526cbce36f3ff98d4d88570b97ee8c84e94 100644 (file)
--- a/gps303/collector.py
+++ b/gps303/collector.py
@@ -38,26 +38,53 @@ class Client:
         self.imei = None
 
     def close(self):
+        log.debug("Closing fd %d (IMEI %s)", self.sock.fileno(), self.imei)
         self.sock.close()
         self.buffer = b""
         self.imei = None
 
     def recv(self):
-        segment = self.sock.recv(4096)
-        if not segment:
+        """ Read from the socket and parse complete messages """
+        try:
+            segment = self.sock.recv(4096)
+        except OSError:
+            log.warning("Reading from fd %d (IMEI %s): %s",
+                    self.sock.fileno(), self.imei, e)
+            return None
+        if not segment:  # Terminal has closed connection
+            log.info("EOF reading from fd %d (IMEI %s)",
+                    self.sock.fileno(), self.imei)
             return None
         when = time()
         self.buffer += segment
-        # implement framing properly
-        msg = handle_packet(packet, self.addr, when)
-        self.buffer = self.buffer[len(packet):]
-        if isinstance(msg, LOGIN):
-            self.imei = msg.imei
-        return msg
+        msgs = []
+        while True:
+            framestart = self.buffer.find(b"xx")
+            if framestart == -1:  # No frames, return whatever we have
+                break
+            if framestart > 0:  # Should not happen, report
+                log.warning("Undecodable data \"%s\" from fd %d (IMEI %s)",
+                        self.buffer[:framestart].hex(), self.sock.fileno(), self.imei)
+                self.buffer = self.buffer[framestart:]
+            # At this point, buffer starts with a packet
+            frameend = self.buffer.find(b"\r\n", 4)
+            if frameend == -1:  # Incomplete frame, return what we have
+                break
+            msg = parse_message(self.buffer[:frameend])
+            self.buffer = self.buffer[frameend+2:]
+            if isinstance(msg, LOGIN):
+                self.imei = msg.imei
+                log.info("LOGIN from fd %d: IMEI %s",
+                        self.sock.fileno(), self.imei)
+            msgs.append(msg)
+        return msgs
 
     def send(self, buffer):
-        self.sock.send(buffer)
-
+        try:
+            self.sock.send(b"xx" + buffer + b"\r\n")
+        except OSError as e:
+            log.error("Sending to fd %d (IMEI %s): %s",
+                    self.sock.fileno, self.imei, e)
 
 class Clients:
     def __init__(self):
@@ -71,6 +98,7 @@ class Clients:
 
     def stop(self, fd):
         clnt = by_fd[fd]
+        log.info("Stop serving fd %d (IMEI %s)", clnt.sock.fileno(), clnt.imei)
         clnt.close()
         if clnt.imei:
             del self.by_imei[clnt.imei]
@@ -78,14 +106,17 @@ class Clients:
 
     def recv(self, fd):
         clnt = by_fd[fd]
-        msg = clnt.recv()
-        if isinstance(msg, LOGIN):
-            self.by_imei[clnt.imei] = clnt
-        return clnt.imei, msg
+        msgs = clnt.recv()
+        result = []
+        for msg in msgs:
+            if isinstance(msg, LOGIN):
+                self.by_imei[clnt.imei] = clnt
+            result.append(clnt.imei, msg)
+        return result
 
-    def response(self, zmsg):
-        if zmsg.imei in self.by_imei:
-            clnt = self.by_imei[zmsg.imei].send(zmsg.payload)
+    def response(self, resp):
+        if resp.imei in self.by_imei:
+            self.by_imei[resp.imei].send(resp.payload)
 
 
 def runserver(opts, conf):
@@ -124,6 +155,7 @@ def runserver(opts, conf):
                     imei, msg = clients.recv(sk)
                     zpub.send(Bcast(imei, msg).as_bytes)
                     if msg is None or isinstance(msg, HIBERNATION):
+                        log.debug("HIBERNATION from fd %d", sk)
                         tostop.append(sk)
             # poll queue consumed, make changes now
             for fd in tostop: