X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fwsgateway.py;h=79937d8e5b1f5d1835e1dfab8a627f2844701f31;hb=311d3cc7b0692e66edb9b9bb9285b2bfc094d571;hp=c7f96524e98573463ac25318d0599421ccab28cb;hpb=ec0679813aad0a64eb9751e2f423c137d84e364a;p=loctrkd.git diff --git a/gps303/wsgateway.py b/gps303/wsgateway.py index c7f9652..79937d8 100644 --- a/gps303/wsgateway.py +++ b/gps303/wsgateway.py @@ -1,5 +1,7 @@ """ Websocket Gateway """ +from datetime import datetime, timezone +from json import dumps, loads from logging import getLogger from socket import socket, AF_INET6, SOCK_STREAM, SOL_SOCKET, SO_REUSEADDR from time import time @@ -16,12 +18,41 @@ from wsproto.utilities import RemoteProtocolError import zmq from . import common -from .zmsg import LocEvt +from .evstore import initdb, fetch +from .gps303proto import ( + GPS_POSITIONING, + WIFI_POSITIONING, + parse_message, +) +from .zmsg import Bcast, topic log = getLogger("gps303/wsgateway") htmlfile = None +def backlog(imei, numback): + result = [] + for is_incoming, timestamp, packet in fetch( + imei, + ((True, GPS_POSITIONING.PROTO), (False, WIFI_POSITIONING.PROTO)), + numback, + ): + msg = parse_message(packet, is_incoming=is_incoming) + result.append( + { + "imei": imei, + "timestamp": str( + datetime.fromtimestamp(timestamp).astimezone( + tz=timezone.utc + ) + ), + "longitude": msg.longitude, + "latitude": msg.latitude, + } + ) + return result + + def try_http(data, fd, e): global htmlfile try: @@ -63,7 +94,7 @@ def try_http(data, fd, e): return ( f"{proto} 500 File not found\r\n" f"Content-Type: text/plain\r\n\r\n" - f'HTML file could not be opened\r\n'.encode() + f"HTML file could not be opened\r\n".encode() ) else: return ( @@ -90,6 +121,8 @@ class Client: self.addr = addr self.ws = WSConnection(ConnectionType.SERVER) self.ws_data = b"" + self.ready = False + self.imeis = set() def close(self): log.debug("Closing fd %d", self.sock.fileno()) @@ -122,6 +155,7 @@ class Client: e, ) self.ws_data = try_http(data, self.sock.fileno(), e) + self.write() # TODO this is a hack log.debug("Sending HTTP response to %d", self.sock.fileno()) msgs = None else: @@ -131,25 +165,38 @@ class Client: log.debug("WebSocket upgrade on fd %d", self.sock.fileno()) # self.ws_data += self.ws.send(event.response()) # Why not?! self.ws_data += self.ws.send(AcceptConnection()) + self.ready = True elif isinstance(event, (CloseConnection, Ping)): log.debug("%s on fd %d", event, self.sock.fileno()) self.ws_data += self.ws.send(event.response()) elif isinstance(event, TextMessage): # TODO: save imei "subscription" log.debug("%s on fd %d", event, self.sock.fileno()) - msgs.append(event.data) + msg = loads(event.data) + msgs.append(msg) + if msg.get("type", None) == "subscribe": + self.imeis = set(msg.get("imei", [])) + log.debug( + "subs list on fd %s is %s", + self.sock.fileno(), + self.imeis, + ) else: log.warning("%s on fd %d", event, self.sock.fileno()) - if self.ws_data: # Temp hack - self.write() return msgs def wants(self, imei): + log.debug( + "wants %s? set is %s on fd %d", + imei, + self.imeis, + self.sock.fileno(), + ) return True # TODO: check subscriptions def send(self, message): - # TODO: filter only wanted imei got from the client - self.ws_data += self.ws.send(Message(data=message.json)) + if self.ready and message["imei"] in self.imeis: + self.ws_data += self.ws.send(Message(data=dumps(message))) def write(self): if self.ws_data: @@ -184,18 +231,12 @@ class Clients: def recv(self, fd): clnt = self.by_fd[fd] - msgs = clnt.recv() - if msgs is None: - return None - result = [] - for msg in msgs: - log.debug("Received: %s", msg) - return result + return clnt.recv() def send(self, msg): towrite = set() for fd, clnt in self.by_fd.items(): - if clnt.wants(msg.imei): + if clnt.wants(msg["imei"]): clnt.send(msg) towrite.add(fd) return towrite @@ -207,15 +248,21 @@ class Clients: waiting.add(fd) return waiting + def subs(self): + result = set() + for clnt in self.by_fd.values(): + result |= clnt.imeis + return result + def runserver(conf): global htmlfile + initdb(conf.get("storage", "dbfn")) htmlfile = conf.get("wsgateway", "htmlfile") zctx = zmq.Context() zsub = zctx.socket(zmq.SUB) - zsub.connect(conf.get("lookaside", "publishurl")) - zsub.setsockopt(zmq.SUBSCRIBE, b"") + zsub.connect(conf.get("collector", "publishurl")) tcpl = socket(AF_INET6, SOCK_STREAM) tcpl.setblocking(False) tcpl.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) @@ -226,9 +273,31 @@ def runserver(conf): poller.register(zsub, flags=zmq.POLLIN) poller.register(tcpfd, flags=zmq.POLLIN) clients = Clients() + activesubs = set() try: towait = set() while True: + neededsubs = clients.subs() + for imei in neededsubs - activesubs: + zsub.setsockopt( + zmq.SUBSCRIBE, + topic(GPS_POSITIONING.PROTO, True, imei), + ) + zsub.setsockopt( + zmq.SUBSCRIBE, + topic(WIFI_POSITIONING.PROTO, False, imei), + ) + for imei in activesubs - neededsubs: + zsub.setsockopt( + zmq.UNSUBSCRIBE, + topic(GPS_POSITIONING.PROTO, True, imei), + ) + zsub.setsockopt( + zmq.UNSUBSCRIBE, + topic(WIFI_POSITIONING.PROTO, False, imei), + ) + activesubs = neededsubs + log.debug("Subscribed to: %s", activesubs) tosend = [] topoll = [] tostop = [] @@ -238,8 +307,21 @@ def runserver(conf): if sk is zsub: while True: try: - zmsg = LocEvt(zsub.recv(zmq.NOBLOCK)) - tosend.append(zmsg) + zmsg = Bcast(zsub.recv(zmq.NOBLOCK)) + msg = parse_message(zmsg.packet, zmsg.is_incoming) + log.debug("Got %s with %s", zmsg, msg) + tosend.append( + { + "imei": zmsg.imei, + "timestamp": str( + datetime.fromtimestamp( + zmsg.when + ).astimezone(tz=timezone.utc) + ), + "longitude": msg.longitude, + "latitude": msg.latitude, + } + ) except zmq.Again: break elif sk == tcpfd: @@ -250,9 +332,16 @@ def runserver(conf): if received is None: log.debug("Client gone from fd %d", sk) tostop.append(sk) + towait.discard(fd) else: for msg in received: log.debug("Received from %d: %s", sk, msg) + if msg.get("type", None) == "subscribe": + imeis = msg.get("imei") + numback = msg.get("backlog", 5) + for imei in imeis: + tosend.extend(backlog(imei, numback)) + towrite.add(sk) elif fl & zmq.POLLOUT: log.debug("Write now open for fd %d", sk) towrite.add(sk)