X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fcollector.py;h=36a5f8a8efd9455b047da437ed60fbfe1c9853a3;hb=099a8195d469486fa9347cc5faaadd3614a2a445;hp=28ee4a8cd92321a8007747fea3dd7226066902b7;hpb=a7065cf3a3dcd36d0b47e9c25acdf30189019f9c;p=loctrkd.git diff --git a/gps303/collector.py b/gps303/collector.py index 28ee4a8..36a5f8a 100644 --- a/gps303/collector.py +++ b/gps303/collector.py @@ -3,7 +3,14 @@ from configparser import ConfigParser from logging import getLogger from os import umask -from socket import socket, AF_INET6, SOCK_STREAM, SOL_SOCKET, SO_REUSEADDR +from socket import ( + socket, + AF_INET6, + SOCK_STREAM, + SOL_SOCKET, + SO_KEEPALIVE, + SO_REUSEADDR, +) from struct import pack from time import time from typing import Dict, List, Optional, Tuple @@ -98,11 +105,18 @@ class Client: break packet = self.buffer[2:frameend] self.buffer = self.buffer[frameend + 2 :] + if len(packet) < 2: # frameend comes too early + log.warning("Packet too short: %s", packet) + break if proto_of_message(packet) == LOGIN.PROTO: - self.imei = parse_message(packet).imei - log.info( - "LOGIN from fd %d (IMEI %s)", self.sock.fileno(), self.imei - ) + msg = parse_message(packet) + if isinstance(msg, LOGIN): # Can be unparseable + self.imei = msg.imei + log.info( + "LOGIN from fd %d (IMEI %s)", + self.sock.fileno(), + self.imei, + ) msgs.append((when, self.addr, packet)) return msgs @@ -112,7 +126,7 @@ class Client: except OSError as e: log.error( "Sending to fd %d (IMEI %s): %s", - self.sock.fileno, + self.sock.fileno(), self.imei, e, ) @@ -171,7 +185,7 @@ class Clients: log.info("Not connected (IMEI %s)", resp.imei) -def runserver(conf: ConfigParser) -> None: +def runserver(conf: ConfigParser, handle_hibernate: bool = True) -> None: # Is this https://github.com/zeromq/pyzmq/issues/1627 still not fixed?! zctx = zmq.Context() # type: ignore zpub = zctx.socket(zmq.PUB) # type: ignore @@ -206,6 +220,7 @@ def runserver(conf: ConfigParser) -> None: break elif sk == tcpfd: clntsock, clntaddr = tcpl.accept() + clntsock.setsockopt(SOL_SOCKET, SO_KEEPALIVE, 1) topoll.append((clntsock, clntaddr)) elif fl & zmq.POLLIN: received = clients.recv(sk) @@ -224,7 +239,7 @@ def runserver(conf: ConfigParser) -> None: packet=packet, ).packed ) - if proto == HIBERNATION.PROTO: + if proto == HIBERNATION.PROTO and handle_hibernate: log.debug( "HIBERNATION from fd %d (IMEI %s)", sk, @@ -258,7 +273,10 @@ def runserver(conf: ConfigParser) -> None: fd = clients.add(clntsock, clntaddr) poller.register(fd, flags=zmq.POLLIN) except KeyboardInterrupt: - pass + zpub.close() + zpull.close() + zctx.destroy() # type: ignore + tcpl.close() if __name__.endswith("__main__"):