X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=loctrkd%2Fcollector.py;h=26d37f33f60af39a123c8aafa043c50244125ef0;hb=c6343a2c2886a35c3a1361a0bf96fe9428ad5aa2;hp=136ecba99aeeb602ee85028caadc0caa0caf744d;hpb=63a086cf3956b93f760b1a0344afd757e0d0392f;p=loctrkd.git

diff --git a/loctrkd/collector.py b/loctrkd/collector.py
index 136ecba..26d37f3 100644
--- a/loctrkd/collector.py
+++ b/loctrkd/collector.py
@@ -14,7 +14,7 @@ from socket import (
 )
 from struct import pack
 from time import time
-from typing import Any, cast, Dict, List, Optional, Tuple, Union
+from typing import Any, cast, Dict, List, Optional, Set, Tuple, Union
 import zmq
 
 from . import common
@@ -27,16 +27,16 @@ MAXBUFFER: int = 4096
 
 class ProtoModule:
     class Stream:
-        @staticmethod
-        def enframe(buffer: bytes, imei: Optional[str] = None) -> bytes:
-            ...
-
         def recv(self, segment: bytes) -> List[Union[bytes, str]]:
             ...
 
         def close(self) -> bytes:
             ...
 
+    @staticmethod
+    def enframe(buffer: bytes, imei: Optional[str] = None) -> bytes:
+        ...
+
     @staticmethod
     def probe_buffer(buffer: bytes) -> bool:
         ...
@@ -61,10 +61,6 @@ class ProtoModule:
     def proto_of_message(packet: bytes) -> str:
         ...
 
-    @staticmethod
-    def proto_by_name(name: str) -> int:
-        ...
-
 
 pmods: List[ProtoModule] = []
 
@@ -139,9 +135,9 @@ class Client:
         return msgs
 
     def send(self, buffer: bytes) -> None:
-        assert self.stream is not None
+        assert self.stream is not None and self.pmod is not None
         try:
-            self.sock.send(self.stream.enframe(buffer, imei=self.imei))
+            self.sock.send(self.pmod.enframe(buffer, imei=self.imei))
         except OSError as e:
             log.error(
                 "Sending to fd %d (IMEI %s): %s",
@@ -155,6 +151,7 @@ class Clients:
     def __init__(self) -> None:
         self.by_fd: Dict[int, Client] = {}
         self.by_imei: Dict[str, Client] = {}
+        self.tostop: Set[int] = set()
 
     def add(self, clntsock: socket, clntaddr: Tuple[str, int]) -> int:
         fd = clntsock.fileno()
@@ -166,7 +163,7 @@ class Clients:
         clnt = self.by_fd[fd]
         log.info("Stop serving fd %d (IMEI %s)", clnt.sock.fileno(), clnt.imei)
         clnt.close()
-        if clnt.imei:
+        if clnt.imei and self.by_imei[clnt.imei] == clnt:  # could be replaced
             del self.by_imei[clnt.imei]
         del self.by_fd[fd]
 
@@ -189,18 +186,11 @@ class Clients:
                     clnt.imei = imei
                     oldclnt = self.by_imei.get(clnt.imei)
                     if oldclnt is not None:
-                        log.info(
-                            "Orphaning fd %d with the same IMEI",
-                            oldclnt.sock.fileno(),
-                        )
+                        oldfd = oldclnt.sock.fileno()
+                        log.info("Removing stale connection on fd %d", oldfd)
                         oldclnt.imei = None
+                        self.tostop.add(oldfd)
                     self.by_imei[clnt.imei] = clnt
-                else:
-                    log.warning(
-                        "Login message from %s: %s, but client imei unfilled",
-                        peeraddr,
-                        packet,
-                    )
             result.append((clnt.pmod, clnt.imei, when, peeraddr, packet))
             log.debug(
                 "Received from %s (IMEI %s): %s",
@@ -247,7 +237,7 @@ def runserver(conf: ConfigParser, handle_hibernate: bool = True) -> None:
         while True:
             tosend = []
             topoll = []
-            tostop = []
+            clients.tostop = set()
             events = poller.poll(1000)
             for sk, fl in events:
                 if sk is zpull:
@@ -266,7 +256,7 @@ def runserver(conf: ConfigParser, handle_hibernate: bool = True) -> None:
                     received = clients.recv(sk)
                     if received is None:
                         log.debug("Terminal gone from fd %d", sk)
-                        tostop.append(sk)
+                        clients.tostop.add(sk)
                     else:
                         for pmod, imei, when, peeraddr, packet in received:
                             proto = pmod.proto_of_message(packet)
@@ -288,7 +278,7 @@ def runserver(conf: ConfigParser, handle_hibernate: bool = True) -> None:
                                     sk,
                                     imei,
                                 )
-                                tostop.append(sk)
+                                clients.tostop.add(sk)
                             respmsg = pmod.inline_response(packet)
                             if respmsg is not None:
                                 tosend.append(
@@ -310,7 +300,7 @@ def runserver(conf: ConfigParser, handle_hibernate: bool = True) -> None:
                             packet=zmsg.packet,
                         ).packed
                     )
-            for fd in tostop:
+            for fd in clients.tostop:
                 poller.unregister(fd)  # type: ignore
                 clients.stop(fd)
             for clntsock, clntaddr in topoll: