问Sanic Webserver: Websocket处理程序返回时关闭套接字；循环中断其他请求处理程序

EN

import os
import random
import string
from functools import partial
from pathlib import Path

from sanic import Sanic

import asyncio
import websockets
from dataclasses import dataclass, field
from typing import Optional, Set

app = Sanic(__name__)

FILE = "/tmp/foobar"
TIMEOUT = 10
INTERVAL = 20


def generate_code(length=12, include_punctuation=False):
    characters = string.ascii_letters + string.digits
    if include_punctuation:
        characters += string.punctuation
    return "".join(random.choice(characters) for x in range(length))


@dataclass
class Client:
    interface: websockets.server.WebSocketServerProtocol = field(repr=False)
    sid: str = field(default_factory=partial(generate_code, 36))

    def __hash__(self):
        return hash(str(self))

    async def keep_alive(self) -> None:
        while True:
            try:
                try:
                    pong_waiter = await self.interface.ping()
                    await asyncio.wait_for(pong_waiter, timeout=TIMEOUT)
                except asyncio.TimeoutError:
                    print("NO PONG!!")
                    await self.feed.unregister(self)
                else:
                    print(f"ping: {self.sid} on <{self.feed.name}>")
                await asyncio.sleep(INTERVAL)
            except websockets.exceptions.ConnectionClosed:
                print(f"broken connection: {self.sid} on <{self.feed.name}>")
                await self.feed.unregister(self)
                break

    async def shutdown(self) -> None:
        self.interface.close()

    async def run(self) -> None:
        try:
            self.feed.app.add_task(self.keep_alive())
            while True:
                pass
        except websockets.exceptions.ConnectionClosed:
            print("connection closed")
        finally:
            await self.feed.unregister(self)


class Feed:
    app: Sanic
    clients: Set[Client]
    cached = None

    def __init__(self, app: Sanic):
        self.clients = set()
        self.app = app

    @classmethod
    async def get(cls, app: Sanic):
        is_existing = False

        if cls.cached:
            is_existing = True
            feed = cls.cached
        else:
            feed = cls(app)
            cls.cached = feed

        if not is_existing:
            feed.app.add_task(feed.receiver())

        return feed, is_existing

    async def receiver(self) -> None:
        print("Feed receiver started")
        mod_time = 0
        while True:
            try:
                stat = os.stat(FILE)
                print(f"times: {mod_time} | {stat.st_mtime}")
                if mod_time != stat.st_mtime:
                    content = self.get_file_contents()
                    for client in self.clients:
                        try:
                            print(f"\tSending to {client.sid}")
                            await client.interface.send(content)
                        except websockets.exceptions.ConnectionClosed:
                            print(f"ConnectionClosed. Client {client.sid}")
            except Exception as e:
                print("Exception while checking file: ", e)

    async def register(
        self, websocket: websockets.server.WebSocketServerProtocol
    ) -> Optional[Client]:
        client = Client(interface=websocket)
        print(f">>> register {client}")

        client.feed = self
        self.clients.add(client)

        # Send initial content
        content = self.get_file_contents()
        client.interface.send(content)

        print(f"\nAll clients\n{self.clients}\n\n")

        return client

    async def unregister(self, client: Client) -> None:
        print(f">>> unregister {client} on <{self.name}>")
        if client in self.clients:
            await client.shutdown()
            self.clients.remove(client)
            print(f"\nAll remaining clients\n{self.clients}\n\n")

    def get_file_contents(self):
        return Path(FILE).read_text()


@app.websocket("/reload")
async def feed(request, ws):
    feed, is_existing = await Feed.get(app)

    client = await feed.register(ws)
    await client.run()


if __name__ == "__main__":
    app.run(debug=True, port=7777)