X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fcollector.py;h=345d0b99a0de7048fc75f8a1a515007d2233318d;hb=2c60838ac44f02802480eeca626040231014caaa;hp=28c636d33642764118067bcadf0355377999b2c2;hpb=b613110bb16c95d9b641882c2ad6869e3ced1a0c;p=loctrkd.git

diff --git a/gps303/collector.py b/gps303/collector.py
index 28c636d..345d0b9 100644
--- a/gps303/collector.py
+++ b/gps303/collector.py
@@ -9,7 +9,7 @@ import sys
 import zmq
 
 from .config import readconfig
-from .gps303proto import handle_packet, make_response, LOGIN, set_config
+from .gps303proto import parse_message, HIBERNATION, LOGIN, set_config
 
 CONF = "/etc/gps303.conf"
 
@@ -19,7 +19,7 @@ log = getLogger("gps303/collector")
 class Bcast:
     """Zmq message to broadcast what was received from the terminal"""
     def __init__(self, imei, msg):
-        self.as_bytes = imei.encode() + msg.encode()
+        self.as_bytes = imei.encode() + msg.to_packet()
 
 
 class Resp:
@@ -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[2: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):
@@ -70,22 +97,26 @@ class Clients:
         return fd
 
     def stop(self, fd):
-        clnt = by_fd[fd]
+        clnt = self.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]
         del self.by_fd[fd]
 
     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
+        clnt = self.by_fd[fd]
+        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):
@@ -118,22 +149,23 @@ def runserver(opts, conf):
                         except zmq.Again:
                             break
                 elif sk == tcpfd:
-                    clntsock, clntaddr = ctlsock.accept()
+                    clntsock, clntaddr = tcpl.accept()
                     topoll.append((clntsock, clntaddr))
                 else:
-                    imei, msg = clients.recv(sk)
-                    zpub.send(Bcast(imei, msg).as_bytes)
-                    if msg is None or isinstance(msg, HIBERNATION):
-                        tostop.append(sk)
+                    for imei, msg in 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:
+                poller.unregister(fd)
                 clients.stop(fd)
-                pollset.unregister(fd)
             for zmsg in tosend:
                 clients.response(zmsg)
             for clntsock, clntaddr in topoll:
                 fd = clients.add(clntsock, clntaddr)
-                pollset.register(fd)
+                poller.register(fd)
     except KeyboardInterrupt:
         pass