fill in `when` in Resp packet

[loctrkd.git] / gps303 / collector.py

diff --git a/gps303/collector.py b/gps303/collector.py
index 4cb7956d2809c4dfd57782d7a15b9bf57abfdbeb..b8cc379a16a5cbac14615ead70e84b841e7425ab 100644 (file)
--- a/gps303/collector.py
+++ b/gps303/collector.py
@@ -1,7 +1,8 @@
 """ TCP server that communicates with terminals """
 
 from logging import getLogger
-from socket import socket, AF_INET, SOCK_STREAM, SOL_SOCKET, SO_REUSEADDR
+from os import umask
+from socket import socket, AF_INET6, SOCK_STREAM, SOL_SOCKET, SO_REUSEADDR
 from time import time
 from struct import pack
 import zmq
@@ -32,7 +33,6 @@ class Client:
         log.debug("Closing fd %d (IMEI %s)", self.sock.fileno(), self.imei)
         self.sock.close()
         self.buffer = b""
-        self.imei = None
 
     def recv(self):
         """Read from the socket and parse complete messages"""
@@ -101,6 +101,7 @@ class Clients:
 
     def add(self, clntsock, clntaddr):
         fd = clntsock.fileno()
+        log.info("Start serving fd %d from %s", fd, clntaddr)
         self.by_fd[fd] = Client(clntsock, clntaddr)
         return fd
 
@@ -122,26 +123,36 @@ class Clients:
             if proto_of_message(packet) == LOGIN.PROTO:  # Could do blindly...
                 self.by_imei[clnt.imei] = clnt
             result.append((clnt.imei, when, peeraddr, packet))
+            log.debug(
+                "Received from %s (IMEI %s): %s",
+                peeraddr,
+                clnt.imei,
+                packet.hex(),
+            )
         return result
 
     def response(self, resp):
         if resp.imei in self.by_imei:
             self.by_imei[resp.imei].send(resp.packet)
+        else:
+            log.info("Not connected (IMEI %s)", resp.imei)
 
 
 def runserver(conf):
     zctx = zmq.Context()
     zpub = zctx.socket(zmq.PUB)
+    zpull = zctx.socket(zmq.PULL)
+    oldmask = umask(0o117)
     zpub.bind(conf.get("collector", "publishurl"))
-    zsub = zctx.socket(zmq.SUB)
-    zsub.connect(conf.get("collector", "listenurl"))
-    tcpl = socket(AF_INET, SOCK_STREAM)
+    zpull.bind(conf.get("collector", "listenurl"))
+    umask(oldmask)
+    tcpl = socket(AF_INET6, SOCK_STREAM)
     tcpl.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
     tcpl.bind(("", conf.getint("collector", "port")))
     tcpl.listen(5)
     tcpfd = tcpl.fileno()
     poller = zmq.Poller()
-    poller.register(zsub, flags=zmq.POLLIN)
+    poller.register(zpull, flags=zmq.POLLIN)
     poller.register(tcpfd, flags=zmq.POLLIN)
     clients = Clients()
     try:
@@ -149,19 +160,29 @@ def runserver(conf):
             tosend = []
             topoll = []
             tostop = []
-            events = poller.poll(10)
+            events = poller.poll(1000)
             for sk, fl in events:
-                if sk is zsub:
+                if sk is zpull:
                     while True:
                         try:
-                            msg = zsub.recv(zmq.NOBLOCK)
-                            tosend.append(Resp(msg))
+                            msg = zpull.recv(zmq.NOBLOCK)
+                            zmsg = Resp(msg)
+                            zpub.send(
+                                Bcast(
+                                    is_incoming=False,
+                                    proto=proto_of_message(zmsg.packet),
+                                    when=zmsg.when,
+                                    imei=zmsg.imei,
+                                    packet=zmsg.packet,
+                                ).packed
+                            )
+                            tosend.append(zmsg)
                         except zmq.Again:
                             break
                 elif sk == tcpfd:
                     clntsock, clntaddr = tcpl.accept()
                     topoll.append((clntsock, clntaddr))
-                else:
+                elif fl & zmq.POLLIN:
                     received = clients.recv(sk)
                     if received is None:
                         log.debug(
@@ -190,17 +211,20 @@ def runserver(conf):
                             respmsg = inline_response(packet)
                             if respmsg is not None:
                                 clients.response(
-                                    Resp(imei=imei, packet=respmsg)
+                                    Resp(imei=imei, when=when, packet=respmsg)
                                 )
+                else:
+                    log.debug("Stray event: %s on socket %s", fl, sk)
             # poll queue consumed, make changes now
             for fd in tostop:
                 poller.unregister(fd)
                 clients.stop(fd)
             for zmsg in tosend:
+                log.debug("Sending to the client: %s", zmsg)
                 clients.response(zmsg)
             for clntsock, clntaddr in topoll:
                 fd = clients.add(clntsock, clntaddr)
-                poller.register(fd)
+                poller.register(fd, flags=zmq.POLLIN)
     except KeyboardInterrupt:
         pass