INNER CODE UNIT · Python

close

ImFeH2/huddol · core/src/fora/adapters/websocket/server.py:222

    def close(self) -> None:
        with self.lock:
            self.closed = True
            while self.items:
                frame = self.items.popleft()
                frame.text = ""
                self.release(frame)
                del frame
        for result in self.responses:
            if not result.done():
                result.set_exception(ConnectionError("Control connection closed"))
        self.responses.clear()
        self.available.set()
        self.closing.set()

    def wait_producers(self) -> None:
        with self.lock:
            self.lock.wait_for(lambda: self.producers == 0)

View source record →