Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

How to close a client connection using the threading interface while stuck in ClientConnection.recv #1449

Open
jmrbcu opened this issue Mar 1, 2024 · 0 comments

Comments

@jmrbcu
Copy link

jmrbcu commented Mar 1, 2024

Hello, I am using the threading client implementation. The problem I am facing is when I try to close the connection while stuck in the ClientConnection.recv method, the network connection goes down for a while and then gets back online.

Sometimes, I get an exception; other times, it has to wait until the closing time out.

Another issue is while receiving messages with ClientConnection.recv, if the network goes down, it stops receiving messages.

Here is the minimal code:

Steps to reproduce (eventually)
1- Run the code
2- Disable the computer network connection
3- wait for at least 60 seconds
4- stop the program with Ctrl+C (KeyboardInterrupt)
5- Output: Sometimes I get the expected behavior but other I get an exception like this:

Traceback (most recent call last):
  File "/Users/jmrbcu/.pyenv/versions/totaltrack3/lib/python3.11/site-packages/websockets/sync/connection.py", line 199, in recv
    return self.recv_messages.get(timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/jmrbcu/.pyenv/versions/totaltrack3/lib/python3.11/site-packages/websockets/sync/messages.py", line 102, in get
    raise EOFError("stream of frames ended")
EOFError: stream of frames ended

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/Users/jmrbcu/.pyenv/versions/3.11.1/lib/python3.11/concurrent/futures/_base.py", line 340, in _invoke_callbacks
    callback(self)
  File "/Volumes/Development/totaltrack/totaltrack/core/telephony/ringcentral_recorder1.py", line 185, in <lambda>
    future.add_done_callback(lambda f: future.result())
                                       ^^^^^^^^^^^^^^^
  File "/Users/jmrbcu/.pyenv/versions/3.11.1/lib/python3.11/concurrent/futures/_base.py", line 449, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/Users/jmrbcu/.pyenv/versions/3.11.1/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "/Users/jmrbcu/.pyenv/versions/3.11.1/lib/python3.11/concurrent/futures/thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Volumes/Development/totaltrack/totaltrack/core/telephony/ringcentral_recorder1.py", line 167, in start
    message = ws.recv()
              ^^^^^^^^^
  File "/Users/jmrbcu/.pyenv/versions/totaltrack3/lib/python3.11/site-packages/websockets/sync/connection.py", line 201, in recv
    raise self.protocol.close_exc from self.recv_events_exc
          ^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/jmrbcu/.pyenv/versions/totaltrack3/lib/python3.11/site-packages/websockets/protocol.py", line 223, in close_exc
    assert self.state is CLOSED, "connection isn't closed yet"
           ^^^^^^^^^^^^^^^^^^^^
AssertionError: connection isn't closed yet

Example Code:

from concurrent.futures import ThreadPoolExecutor, wait

fmt = "%(asctime)s [%(levelname)s] %(name)s.%(funcName)s:%(lineno)d %(message)s"
logging.basicConfig(level=logging.INFO, format=fmt)
stop = threading.Event()

def start():
    while True:
        try:
            message = ws.recv()
            logger.info("New message received: %s", message)
        except TimeoutError:
            logger.info("Timeout occurred")
        except websockets.ConnectionClosed as error:
            logger.info("Connection closed: %s", error)
            return

with ThreadPoolExecutor() as pool:
    ws = connect(api.get_websocket_endpoint())
    future = pool.submit(start)
    future.add_done_callback(lambda f: future.result())

    try:
        wait([future])
    except KeyboardInterrupt:
        print("Stopping")
        stop.set()
        ws.close()
        print("Stopped")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

No branches or pull requests

1 participant