X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fwsgateway.py;h=c7f96524e98573463ac25318d0599421ccab28cb;hb=ec0679813aad0a64eb9751e2f423c137d84e364a;hp=a4a0b6d441d31d8a42800adc3e3d77eebd196547;hpb=117fd17ef103bb32940433955eba22f7fa457b99;p=loctrkd.git

diff --git a/gps303/wsgateway.py b/gps303/wsgateway.py
index a4a0b6d..c7f9652 100644
--- a/gps303/wsgateway.py
+++ b/gps303/wsgateway.py
@@ -19,11 +19,11 @@ from . import common
 from .zmsg import LocEvt
 
 log = getLogger("gps303/wsgateway")
-htmldata = None
+htmlfile = None
 
 
 def try_http(data, fd, e):
-    global htmldata
+    global htmlfile
     try:
         lines = data.decode().split("\r\n")
         request = lines[0]
@@ -43,24 +43,33 @@ def try_http(data, fd, e):
         except ValueError:
             pass
         if op == "GET":
-            if htmldata is None:
+            if htmlfile is None:
                 return (
                     f"{proto} 500 No data configured\r\n"
                     f"Content-Type: text/plain\r\n\r\n"
                     f"HTML data not configure on the server\r\n".encode()
                 )
             elif resource == "/":
-                length = len(htmldata.encode("utf-8"))
-                return (
-                    f"{proto} 200 Ok\r\n"
-                    f"Content-Type: text/html; charset=utf-8\r\n"
-                    f"Content-Length: {length:d}\r\n\r\n" + htmldata
-                ).encode("utf-8")
+                try:
+                    with open(htmlfile, "rb") as fl:
+                        htmldata = fl.read()
+                    length = len(htmldata)
+                    return (
+                        f"{proto} 200 Ok\r\n"
+                        f"Content-Type: text/html; charset=utf-8\r\n"
+                        f"Content-Length: {len(htmldata):d}\r\n\r\n"
+                    ).encode("utf-8") + htmldata
+                except OSError:
+                    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()
+                    )
             else:
                 return (
                     f"{proto} 404 File not found\r\n"
                     f"Content-Type: text/plain\r\n\r\n"
-                    f"We can only serve \"/\"\r\n".encode()
+                    f'We can only serve "/"\r\n'.encode()
                 )
         else:
             return (
@@ -143,16 +152,18 @@ class Client:
         self.ws_data += self.ws.send(Message(data=message.json))
 
     def write(self):
-        try:
-            sent = self.sock.send(self.ws_data)
-            self.ws_data = self.ws_data[sent:]
-        except OSError as e:
-            log.error(
-                "Sending to fd %d: %s",
-                self.sock.fileno(),
-                e,
-            )
-            self.ws_data = b""
+        if self.ws_data:
+            try:
+                sent = self.sock.send(self.ws_data)
+                self.ws_data = self.ws_data[sent:]
+            except OSError as e:
+                log.error(
+                    "Sending to fd %d: %s",
+                    self.sock.fileno(),
+                    e,
+                )
+                self.ws_data = b""
+        return bool(self.ws_data)
 
 
 class Clients:
@@ -182,21 +193,25 @@ class Clients:
         return result
 
     def send(self, msg):
-        for clnt in self.by_fd.values():
+        towrite = set()
+        for fd, clnt in self.by_fd.items():
             if clnt.wants(msg.imei):
                 clnt.send(msg)
-                clnt.write()
+                towrite.add(fd)
+        return towrite
+
+    def write(self, towrite):
+        waiting = set()
+        for fd, clnt in [(fd, self.by_fd.get(fd)) for fd in towrite]:
+            if clnt.write():
+                waiting.add(fd)
+        return waiting
 
 
 def runserver(conf):
-    global htmldata
-    try:
-        with open(
-            conf.get("wsgateway", "htmlfile"), encoding="utf-8"
-        ) as fl:
-            htmldata = fl.read()
-    except OSError:
-        pass
+    global htmlfile
+
+    htmlfile = conf.get("wsgateway", "htmlfile")
     zctx = zmq.Context()
     zsub = zctx.socket(zmq.SUB)
     zsub.connect(conf.get("lookaside", "publishurl"))
@@ -212,11 +227,13 @@ def runserver(conf):
     poller.register(tcpfd, flags=zmq.POLLIN)
     clients = Clients()
     try:
+        towait = set()
         while True:
             tosend = []
             topoll = []
             tostop = []
-            events = poller.poll(5000)
+            towrite = set()
+            events = poller.poll()
             for sk, fl in events:
                 if sk is zsub:
                     while True:
@@ -236,6 +253,10 @@ def runserver(conf):
                     else:
                         for msg in received:
                             log.debug("Received from %d: %s", sk, msg)
+                elif fl & zmq.POLLOUT:
+                    log.debug("Write now open for fd %d", sk)
+                    towrite.add(sk)
+                    towait.discard(sk)
                 else:
                     log.debug("Stray event: %s on socket %s", fl, sk)
             # poll queue consumed, make changes now
@@ -243,12 +264,26 @@ def runserver(conf):
                 poller.unregister(fd)
                 clients.stop(fd)
             for zmsg in tosend:
-                log.debug("Sending to the client: %s", zmsg)
-                clients.send(zmsg)
+                log.debug("Sending to the clients: %s", zmsg)
+                towrite |= clients.send(zmsg)
             for clntsock, clntaddr in topoll:
                 fd = clients.add(clntsock, clntaddr)
                 poller.register(fd, flags=zmq.POLLIN)
-            # TODO: Handle write overruns (register for POLLOUT)
+            # Deal with actually writing the data out
+            trywrite = towrite - towait
+            morewait = clients.write(trywrite)
+            log.debug(
+                "towait %s, tried %s, still busy %s",
+                towait,
+                trywrite,
+                morewait,
+            )
+            for fd in morewait - trywrite:  # new fds waiting for write
+                poller.modify(fd, flags=zmq.POLLIN | zmq.POLLOUT)
+            for fd in trywrite - morewait:  # no longer waiting for write
+                poller.modify(fd, flags=zmq.POLLIN)
+            towait &= trywrite
+            towait |= morewait
     except KeyboardInterrupt:
         pass