X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fcollector.py;h=68c95bfb77d0ddbb75e8da64d25f6972d7e94847;hb=4c173a5448990cd4da398be1bf3479bef17b1048;hp=d8ca86c0791a2b8c0cda3dae053153b649c863ce;hpb=4c9597cff07443042f51c2330a050b7ee5bc9bcb;p=loctrkd.git diff --git a/gps303/collector.py b/gps303/collector.py index d8ca86c..68c95bf 100644 --- a/gps303/collector.py +++ b/gps303/collector.py @@ -1,55 +1,115 @@ +""" TCP server that communicates with terminals """ + from getopt import getopt -from logging import getLogger, StreamHandler, DEBUG, INFO +from logging import getLogger from logging.handlers import SysLogHandler from socket import socket, AF_INET, SOCK_STREAM, SOL_SOCKET, SO_REUSEADDR from time import time -import sys +from struct import pack import zmq -from .config import readconfig -from .GT06mod import handle_packet, make_response, LOGIN, set_config - -CONF = "/etc/gps303.conf" +from . import common +from .gps303proto import HIBERNATION, LOGIN, parse_message, proto_of_message 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 = ( + pack("B", proto_of_message(msg)) + + ("0000000000000000" if imei is None else imei).encode() + + msg + ) -class Zmsg: - def __init__(self, msg): - self.imei = msg[:16].decode() - self.payload = msg[16:] +class Resp: + """Zmq message received from a third party to send to the terminal""" + + def __init__(self, *args, **kwargs): + if not kwargs and len(args) == 1 and isinstance(args[0], bytes): + self.imei = msg[:16].decode() + self.payload = msg[16:] + elif len(args) == 0: + self.imei = kwargs["imei"] + self.payload = kwargs["payload"] class Client: - def __init__(self, clntsock, clntaddr): - self.clntsock = clntsock - self.clntaddr = clntaddr + """Connected socket to the terminal plus buffer and metadata""" + + def __init__(self, sock, addr): + self.sock = sock + self.addr = addr self.buffer = b"" self.imei = None def close(self): - self.clntsock.close() + log.debug("Closing fd %d (IMEI %s)", self.sock.fileno(), self.imei) + self.sock.close() + self.buffer = b"" + self.imei = None def recv(self): - packet = self.clntsock.recv(4096) - if not packet: + """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 += packet - # implement framing properly - msg = handle_packet(packet, self.clntaddr, when) - self.buffer = self.buffer[len(packet):] - if isinstance(msg, LOGIN): - self.imei = msg.imei - return msg + self.buffer += segment + 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 + packet = self.buffer[2:frameend] + self.buffer = self.buffer[frameend + 2 :] + 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 + ) + msgs.append(packet) + return msgs def send(self, buffer): - self.clntsock.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: @@ -63,25 +123,31 @@ 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() + if msgs is None: + return None + result = [] + for msg in msgs: + if proto_of_message(msg) == LOGIN.PROTO: # Could do blindly... + 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): +def runserver(conf): zctx = zmq.Context() zpub = zctx.socket(zmq.PUB) zpub.bind(conf.get("collector", "publishurl")) @@ -107,38 +173,37 @@ def runserver(opts, conf): while True: try: msg = zsub.recv(zmq.NOBLOCK) - tosend.append(Zmsg(msg)) + tosend.append(Resp(msg)) 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 proto_of_message(msg) == HIBERNATION.PROTO + ): + log.debug( + "HIBERNATION from fd %d (IMEI %s)", sk, imei + ) + tostop.append(sk) + elif proto_of_message(msg) == LOGIN.PROTO: + clients.response(Resp(imei=imei, payload=LOGIN.response())) # 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 if __name__.endswith("__main__"): - opts, _ = getopt(sys.argv[1:], "c:d") - opts = dict(opts) - conf = readconfig(opts["-c"] if "-c" in opts else CONF) - if sys.stdout.isatty(): - log.addHandler(StreamHandler(sys.stderr)) - else: - log.addHandler(SysLogHandler(address="/dev/log")) - log.setLevel(DEBUG if "-d" in opts else INFO) - log.info("starting with options: %s", opts) - runserver(opts, conf) + runserver(common.init(log))