Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/websocket/_app.py: 11%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import inspect
2import socket
3import threading
4import time
5from typing import Any, Callable, List, Optional, Tuple, Union
7from ._logging import debug, error, info, warning
8from ._abnf import ABNF
9from ._core import WebSocket, getdefaulttimeout
10from ._exceptions import (
11 WebSocketConnectionClosedException,
12 WebSocketException,
13 WebSocketTimeoutException,
14)
15from ._ssl_compat import SSLError
16from ._url import parse_url
17from ._dispatcher import Dispatcher, DispatcherBase, SSLDispatcher, WrappedDispatcher
19"""
20_app.py
21websocket - WebSocket client library for Python
23Copyright 2026 engn33r
25Licensed under the Apache License, Version 2.0 (the "License");
26you may not use this file except in compliance with the License.
27You may obtain a copy of the License at
29 http://www.apache.org/licenses/LICENSE-2.0
31Unless required by applicable law or agreed to in writing, software
32distributed under the License is distributed on an "AS IS" BASIS,
33WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
34See the License for the specific language governing permissions and
35limitations under the License.
36"""
38__all__ = ["WebSocketApp"]
40RECONNECT = 0
43def set_reconnect(reconnectInterval: int) -> None:
44 global RECONNECT
45 RECONNECT = reconnectInterval
48class WebSocketApp:
49 """
50 Higher level of APIs are provided. The interface is like JavaScript WebSocket object.
51 """
53 def __init__(
54 self,
55 url: str,
56 header: Optional[
57 Union[
58 list[str],
59 dict[str, str],
60 Callable[[], Union[list[str], dict[str, str]]],
61 ]
62 ] = None,
63 on_open: Optional[Callable[["WebSocketApp"], None]] = None,
64 on_reconnect: Optional[Callable[["WebSocketApp"], None]] = None,
65 on_message: Optional[Callable[["WebSocketApp", Any], None]] = None,
66 on_error: Optional[Callable[["WebSocketApp", Any], None]] = None,
67 on_close: Optional[Callable[["WebSocketApp", Any, Any], None]] = None,
68 on_ping: Optional[Callable] = None,
69 on_pong: Optional[Callable] = None,
70 on_cont_message: Optional[Callable] = None,
71 keep_running: bool = True,
72 get_mask_key: Optional[Callable] = None,
73 cookie: Optional[str] = None,
74 subprotocols: Optional[list[str]] = None,
75 on_data: Optional[Callable] = None,
76 socket: Optional[socket.socket] = None,
77 ) -> None:
78 """
79 WebSocketApp initialization
81 Parameters
82 ----------
83 url: str
84 Websocket url.
85 header: list or dict or Callable
86 Custom header for websocket handshake.
87 If the parameter is a callable object, it is called just before the connection attempt.
88 The returned dict or list is used as custom header value.
89 This could be useful in order to properly setup timestamp dependent headers.
90 on_open: function
91 Callback object which is called at opening websocket.
92 on_open has one argument.
93 The 1st argument is this class object.
94 on_reconnect: function
95 Callback object which is called at reconnecting websocket.
96 on_reconnect has one argument.
97 The 1st argument is this class object.
98 on_message: function
99 Callback object which is called when received data.
100 on_message has 2 arguments.
101 The 1st argument is this class object.
102 The 2nd argument is utf-8 data received from the server.
103 on_error: function
104 Callback object which is called when we get error.
105 on_error has 2 arguments.
106 The 1st argument is this class object.
107 The 2nd argument is exception object.
108 Called for every error, including errors during reconnect
109 attempts (since 1.10.0). To distinguish reconnect-phase
110 errors from the initial failure, track state with the
111 on_reconnect and on_open callbacks.
112 on_close: function
113 Callback object which is called when connection is closed.
114 on_close has 3 arguments.
115 The 1st argument is this class object.
116 The 2nd argument is close_status_code.
117 The 3rd argument is close_msg.
118 on_cont_message: function
119 Callback object which is called when a continuation
120 frame is received.
121 on_cont_message has 3 arguments.
122 The 1st argument is this class object.
123 The 2nd argument is utf-8 string which we get from the server.
124 The 3rd argument is continue flag. if 0, the data continue
125 to next frame data
126 on_data: function
127 Callback object which is called when a message received.
128 This is called before on_message or on_cont_message,
129 and then on_message or on_cont_message is called.
130 on_data has 4 argument.
131 The 1st argument is this class object.
132 The 2nd argument is utf-8 string which we get from the server.
133 The 3rd argument is data type. ABNF.OPCODE_TEXT or ABNF.OPCODE_BINARY will be came.
134 The 4th argument is continue flag. If 0, the data continue
135 keep_running: bool
136 This parameter is obsolete and ignored.
137 get_mask_key: function
138 A callable function to get new mask keys, see the
139 WebSocket.set_mask_key's docstring for more information.
140 cookie: str
141 Cookie value.
142 subprotocols: list
143 List of available sub protocols. Default is None.
144 socket: socket
145 Pre-initialized stream socket. With a wss:// URL, a socket that is
146 not already an ssl.SSLSocket is wrapped in TLS using sslopt.
147 """
148 self.url = url
149 self.header = header if header is not None else []
150 self.cookie = cookie
152 self.on_open = on_open
153 self.on_reconnect = on_reconnect
154 self.on_message = on_message
155 self.on_data = on_data
156 self.on_error = on_error
157 self.on_close = on_close
158 self.on_ping = on_ping
159 self.on_pong = on_pong
160 self.on_cont_message = on_cont_message
161 self.keep_running = False
162 self.get_mask_key = get_mask_key
163 self.sock: Optional[WebSocket] = None
164 self.last_ping_tm = float(0)
165 self.last_pong_tm = float(0)
166 self.ping_thread: Optional[threading.Thread] = None
167 self.stop_ping: Optional[threading.Event] = None
168 self.ping_interval = float(0)
169 self.ping_timeout: Optional[Union[float, int]] = None
170 self.ping_payload = ""
171 self.subprotocols = subprotocols
172 self.prepared_socket = socket
173 self.has_errored = False
174 self.has_done_teardown = False
175 self.has_done_teardown_lock = threading.Lock()
176 self.last_close_frame: Optional[ABNF] = None
178 def send(self, data: Union[bytes, str], opcode: int = ABNF.OPCODE_TEXT) -> None:
179 """
180 send message
182 Parameters
183 ----------
184 data: str
185 Message to send. If you set opcode to OPCODE_TEXT,
186 data must be utf-8 string or unicode.
187 opcode: int
188 Operation code of data. Default is OPCODE_TEXT.
189 """
191 if not self.sock or self.sock.send(data, opcode) == 0:
192 raise WebSocketConnectionClosedException("Connection is already closed.")
194 def send_text(self, text_data: str) -> None:
195 """
196 Sends UTF-8 encoded text.
197 """
198 if not self.sock or self.sock.send(text_data, ABNF.OPCODE_TEXT) == 0:
199 raise WebSocketConnectionClosedException("Connection is already closed.")
201 def send_bytes(self, data: Union[bytes, bytearray]) -> None:
202 """
203 Sends a sequence of bytes.
204 """
205 if not self.sock or self.sock.send(data, ABNF.OPCODE_BINARY) == 0:
206 raise WebSocketConnectionClosedException("Connection is already closed.")
208 def close(self, **kwargs: Any) -> None:
209 """
210 Close websocket connection.
211 """
212 self.keep_running = False
213 sock = self.sock
214 if sock:
215 sock.close(**kwargs)
216 # Capture the peer's close frame before clearing socket reference
217 if sock.close_frame is not None:
218 self.last_close_frame = sock.close_frame
219 self.sock = None
221 def _start_ping_thread(self) -> None:
222 self.last_ping_tm = self.last_pong_tm = float(0)
223 self.stop_ping = threading.Event()
224 self.ping_thread = threading.Thread(target=self._send_ping)
225 self.ping_thread.daemon = True
226 self.ping_thread.start()
228 def _stop_ping_thread(self) -> None:
229 if self.stop_ping:
230 self.stop_ping.set()
231 if self.ping_thread and self.ping_thread.is_alive():
232 self.ping_thread.join(3)
233 # Handle thread leak - if thread doesn't terminate within timeout,
234 # force cleanup and log warning instead of abandoning the thread
235 if self.ping_thread.is_alive():
236 warning(
237 "Ping thread failed to terminate within 3 seconds, "
238 "forcing cleanup. Thread may be blocked."
239 )
240 # Force cleanup by clearing references even if thread is still alive
241 # The daemon thread will eventually be cleaned up by Python's GC
242 # but we prevent resource leaks by not holding references
244 # Always clean up references regardless of thread state
245 self.ping_thread = None
246 self.stop_ping = None
247 self.last_ping_tm = self.last_pong_tm = float(0)
249 def _send_ping(self) -> None:
250 if self.stop_ping is None:
251 return
252 if self.keep_running is False:
253 return
254 while not self.stop_ping.wait(self.ping_interval) and self.keep_running is True:
255 if self.sock:
256 self.last_ping_tm = time.time()
257 try:
258 debug("Sending ping")
259 self.sock.ping(self.ping_payload)
260 except Exception as e:
261 debug(f"Failed to send ping: {e}")
263 def ready(self):
264 return self.sock and self.sock.connected
266 def run_forever(
267 self,
268 sockopt: Optional[list] = None,
269 sslopt: Optional[dict] = None,
270 ping_interval: Union[float, int] = 0,
271 ping_timeout: Optional[Union[float, int]] = None,
272 ping_payload: str = "",
273 http_proxy_host: Optional[str] = None,
274 http_proxy_port: Optional[Union[int, str]] = None,
275 http_no_proxy: Optional[list] = None,
276 http_proxy_auth: Optional[tuple] = None,
277 http_proxy_timeout: Optional[float] = None,
278 skip_utf8_validation: bool = False,
279 host: Optional[str] = None,
280 origin: Optional[str] = None,
281 dispatcher: Any = None,
282 suppress_origin: bool = False,
283 suppress_host: bool = False,
284 proxy_type: Optional[str] = None,
285 reconnect: Optional[int] = None,
286 redirect_limit: int = 3,
287 ) -> bool:
288 """
289 Run event loop for WebSocket framework.
291 This loop is an infinite loop and is alive while websocket is available.
293 Parameters
294 ----------
295 sockopt: tuple
296 Values for socket.setsockopt.
297 sockopt must be tuple
298 and each element is argument of sock.setsockopt.
299 sslopt: dict
300 Optional dict object for ssl socket option.
301 ping_interval: int or float
302 Automatically send "ping" command
303 every specified period (in seconds).
304 If set to 0, no ping is sent periodically.
305 ping_timeout: int or float
306 Timeout (in seconds) if the pong message is not received.
307 ping_payload: str
308 Payload message to send with each ping.
309 http_proxy_host: str
310 HTTP proxy host name.
311 http_proxy_port: int or str
312 HTTP proxy port. Required when http_proxy_host is set. Proxies
313 from environment variables default to port 80.
314 http_no_proxy: list
315 Whitelisted host names that don't use the proxy.
316 http_proxy_timeout: int or float
317 HTTP proxy timeout, default is 60 sec as per python-socks.
318 http_proxy_auth: tuple
319 HTTP proxy auth information. tuple of username and password. Default is None.
320 skip_utf8_validation: bool
321 skip utf8 validation.
322 host: str
323 update host header.
324 origin: str
325 update origin header.
326 dispatcher: Dispatcher object
327 customize reading data from socket.
328 suppress_origin: bool
329 suppress outputting origin header.
330 suppress_host: bool
331 suppress outputting host header.
332 proxy_type: str
333 type of proxy from: http, socks4, socks4a, socks5, socks5h
334 reconnect: int
335 delay interval when reconnecting
336 redirect_limit: int
337 Number of HTTP redirects to follow during each opening handshake,
338 the same as ``WebSocket.connect(redirect_limit=...)``. Default is 3.
339 Set it to 0 to refuse redirects, for example when the URL was
340 validated against an allowlist and a redirect could point the
341 connection at another host. A refused redirect is reported to
342 ``on_error`` as a ``WebSocketException``.
344 Returns
345 -------
346 teardown: bool
347 False if the `WebSocketApp` is closed gracefully,
348 True if an error was handled during the loop.
350 Raises
351 ------
352 KeyboardInterrupt
353 If interrupted during the loop. The connection is torn down before
354 the interrupt is propagated to the caller.
355 """
357 if reconnect is None:
358 reconnect = RECONNECT
360 if ping_timeout is not None and ping_timeout <= 0:
361 raise WebSocketException("Ensure ping_timeout > 0")
362 if ping_interval is not None and ping_interval < 0:
363 raise WebSocketException("Ensure ping_interval >= 0")
364 if ping_timeout and ping_interval and ping_interval <= ping_timeout:
365 raise WebSocketException("Ensure ping_interval > ping_timeout")
366 if redirect_limit < 0:
367 raise WebSocketException("Ensure redirect_limit >= 0")
368 if not sockopt:
369 sockopt = []
370 if not sslopt:
371 sslopt = {}
372 if self.sock:
373 raise WebSocketException("socket is already opened")
375 self.ping_interval = ping_interval
376 self.ping_timeout = ping_timeout
377 self.ping_payload = ping_payload
378 self.has_done_teardown = False
379 self.has_errored = False
380 self.keep_running = True
382 def teardown(close_frame: Optional[ABNF] = None) -> None:
383 """
384 Tears down the connection.
386 Parameters
387 ----------
388 close_frame: ABNF frame
389 If close_frame is set, the on_close handler is invoked
390 with the statusCode and reason from the provided frame.
391 """
393 # teardown() is called in many code paths to ensure resources are cleaned up and on_close is fired.
394 # To ensure the work is only done once, we use this bool and lock.
395 with self.has_done_teardown_lock:
396 if self.has_done_teardown:
397 return
398 self.has_done_teardown = True
400 self._stop_ping_thread()
401 self.keep_running = False
403 if self.sock:
404 # in cases like handleDisconnect, the "on_error" callback is called first. If the WebSocketApp
405 # is being used in a multithreaded application, we nee to make sure that "self.sock" is cleared
406 # before calling close, otherwise logic built around the sock being set can cause issues -
407 # specifically calling "run_forever" again, since is checks if "self.sock" is set.
408 current_sock = self.sock
409 self.sock = None
410 current_sock.close()
412 # Use stored close frame as fallback if none provided (e.g., client-initiated close)
413 effective_close_frame = (
414 close_frame if close_frame else self.last_close_frame
415 )
416 close_status_code, close_reason = self._get_close_args(
417 effective_close_frame
418 )
419 # Finally call the callback AFTER all teardown is complete
420 self._callback(self.on_close, close_status_code, close_reason)
422 def initialize_socket(reconnecting: bool = False) -> None:
423 if reconnecting and self.sock:
424 self.sock.shutdown()
426 # Reset close frame to avoid stale data from previous connections
427 self.last_close_frame = None
429 self.sock = WebSocket(
430 self.get_mask_key,
431 sockopt=sockopt,
432 sslopt=sslopt,
433 fire_cont_frame=self.on_cont_message is not None,
434 skip_utf8_validation=skip_utf8_validation,
435 enable_multithread=True,
436 dispatcher=dispatcher,
437 )
439 self.sock.settimeout(getdefaulttimeout())
440 try:
441 header = self.header() if callable(self.header) else self.header
443 self.sock.connect(
444 self.url,
445 header=header,
446 cookie=self.cookie,
447 http_proxy_host=http_proxy_host,
448 http_proxy_port=http_proxy_port,
449 http_no_proxy=http_no_proxy,
450 http_proxy_auth=http_proxy_auth,
451 http_proxy_timeout=http_proxy_timeout,
452 subprotocols=self.subprotocols,
453 host=host,
454 origin=origin,
455 suppress_origin=suppress_origin,
456 suppress_host=suppress_host,
457 proxy_type=proxy_type,
458 socket=self.prepared_socket,
459 redirect_limit=redirect_limit,
460 )
462 info("Websocket connected")
464 if self.ping_interval:
465 self._start_ping_thread()
467 if reconnecting and self.on_reconnect:
468 self._callback(self.on_reconnect)
469 else:
470 self._callback(self.on_open)
472 assert dispatcher is not None
473 dispatcher.read(self.sock.sock, read, check)
474 except (
475 WebSocketConnectionClosedException,
476 ConnectionRefusedError,
477 KeyboardInterrupt,
478 SystemExit,
479 Exception,
480 ) as e:
481 handleDisconnect(e)
483 def read() -> bool:
484 if not self.keep_running:
485 teardown()
486 return False
488 if self.sock is None:
489 return False
491 try:
492 op_code, frame = self.sock.recv_data_frame(True)
493 except (
494 WebSocketConnectionClosedException,
495 KeyboardInterrupt,
496 SSLError,
497 ConnectionResetError,
498 WebSocketTimeoutException,
499 ) as e:
500 if custom_dispatcher:
501 return closed(e)
502 else:
503 raise e
505 if op_code == ABNF.OPCODE_CLOSE:
506 return closed(frame)
507 elif op_code == ABNF.OPCODE_PING:
508 self._callback(self.on_ping, frame.data)
509 elif op_code == ABNF.OPCODE_PONG:
510 self.last_pong_tm = time.time()
511 self._callback(self.on_pong, frame.data)
512 elif op_code == ABNF.OPCODE_CONT and self.on_cont_message:
513 self._callback(self.on_data, frame.data, frame.opcode, frame.fin)
514 self._callback(self.on_cont_message, frame.data, frame.fin)
515 else:
516 data = frame.data
517 if op_code == ABNF.OPCODE_TEXT and not skip_utf8_validation:
518 data = data.decode("utf-8")
519 self._callback(self.on_data, data, frame.opcode, True)
520 self._callback(self.on_message, data)
522 return True
524 def check() -> bool:
525 if self.ping_timeout:
526 has_timeout_expired = (
527 time.time() - self.last_ping_tm > self.ping_timeout
528 )
529 has_pong_not_arrived_after_last_ping = (
530 self.last_pong_tm - self.last_ping_tm < 0
531 )
532 has_pong_arrived_too_late = (
533 self.last_pong_tm - self.last_ping_tm > self.ping_timeout
534 )
536 if (
537 self.last_ping_tm
538 and has_timeout_expired
539 and (
540 has_pong_not_arrived_after_last_ping
541 or has_pong_arrived_too_late
542 )
543 ):
544 raise WebSocketTimeoutException("ping/pong timed out")
545 return True
547 def closed(
548 e: Union[
549 WebSocketConnectionClosedException,
550 ConnectionRefusedError,
551 KeyboardInterrupt,
552 SystemExit,
553 Exception,
554 str,
555 "ABNF", # Now explicitly handle ABNF frame objects
556 ] = "closed unexpectedly",
557 ) -> bool:
558 close_frame: Optional[ABNF] = None
559 if type(e) is str:
560 e = WebSocketConnectionClosedException(e)
561 elif isinstance(e, ABNF) and e.opcode == ABNF.OPCODE_CLOSE:
562 close_frame = e
563 # Convert close frames to a descriptive exception for on_error callback
564 close_status_code, close_reason = self._parse_close_frame(e)
565 reason_parts: List[str] = []
566 if close_status_code is None:
567 message = "Connection closed"
568 elif close_status_code == 1000:
569 message = "Connection closed normally (code 1000)"
570 else:
571 message = f"Connection closed (code {close_status_code})"
572 if close_reason:
573 reason_parts.append(close_reason)
574 if reason_parts:
575 message = f"{message}: {'; '.join(reason_parts)}"
576 converted = WebSocketConnectionClosedException(message)
577 setattr(converted, "status_code", close_status_code)
578 setattr(converted, "reason", close_reason)
579 e = converted
580 return handleDisconnect(e, close_frame=close_frame) # type: ignore[arg-type]
582 def handleDisconnect(
583 e: Union[
584 WebSocketConnectionClosedException,
585 ConnectionRefusedError,
586 KeyboardInterrupt,
587 SystemExit,
588 Exception,
589 ],
590 close_frame: Optional[ABNF] = None,
591 ) -> bool:
592 if close_frame is None:
593 self.has_errored = True
594 self._stop_ping_thread()
595 self._callback(self.on_error, e)
597 if isinstance(e, (KeyboardInterrupt, SystemExit)):
598 teardown(close_frame)
599 # Propagate further
600 raise
602 if reconnect:
603 info(f"{e} - reconnect")
604 if custom_dispatcher:
605 debug(
606 f"Calling custom dispatcher reconnect [{len(inspect.stack())} frames in stack]"
607 )
608 assert dispatcher is not None
609 dispatcher.reconnect(reconnect, initialize_socket)
610 else:
611 error(f"{e} - goodbye")
612 teardown(close_frame)
613 return self.has_errored
615 custom_dispatcher = bool(dispatcher)
616 try:
617 dispatcher = self.create_dispatcher(
618 ping_timeout, dispatcher, parse_url(self.url)[3], closed
619 )
620 except KeyboardInterrupt:
621 teardown()
622 raise
624 try:
625 initialize_socket()
626 if not custom_dispatcher and reconnect:
627 while self.keep_running:
628 debug(
629 f"Calling dispatcher reconnect [{len(inspect.stack())} frames in stack]"
630 )
631 dispatcher.reconnect(reconnect, initialize_socket)
632 except (KeyboardInterrupt, Exception) as e:
633 info(f"tearing down on exception {e!r}")
634 teardown()
635 if isinstance(e, KeyboardInterrupt):
636 raise
637 finally:
638 if not custom_dispatcher:
639 # Ensure teardown was called before returning from run_forever
640 teardown()
642 return self.has_errored
644 def create_dispatcher(
645 self,
646 ping_timeout: Optional[Union[float, int]],
647 dispatcher: Optional[DispatcherBase] = None,
648 is_ssl: bool = False,
649 handleDisconnect: Optional[Callable] = None,
650 ) -> Union[Dispatcher, SSLDispatcher, WrappedDispatcher]:
651 if dispatcher: # If custom dispatcher is set, use WrappedDispatcher
652 return WrappedDispatcher(self, ping_timeout, dispatcher, handleDisconnect)
653 timeout = ping_timeout or 10
654 if is_ssl:
655 return SSLDispatcher(self, timeout)
656 return Dispatcher(self, timeout)
658 def _get_close_args(
659 self, close_frame: Optional[ABNF]
660 ) -> List[Optional[Union[int, str]]]:
661 """
662 _get_close_args extracts the close code and reason from the close body
663 if it exists (RFC6455 says WebSocket Connection Close Code is optional)
664 """
665 # Need to catch the case where close_frame is None
666 # Otherwise the following if statement causes an error
667 if not close_frame:
668 return [None, None]
669 close_status_code, reason = self._parse_close_frame(close_frame)
670 if not self.on_close:
671 return [None, None]
672 return [close_status_code, reason]
674 def _parse_close_frame(
675 self, close_frame: Optional[ABNF]
676 ) -> Tuple[Optional[int], Optional[str]]:
677 """
678 Parse a close frame into status code and UTF-8 reason text.
679 """
680 if not close_frame or not getattr(close_frame, "data", None):
681 return (None, None)
683 data = close_frame.data
684 if isinstance(data, bytes):
685 data_bytes = data
686 elif isinstance(data, str):
687 data_bytes = data.encode("utf-8")
688 else:
689 data_bytes = bytes(data)
691 if len(data_bytes) < 2:
692 return (None, None)
694 close_status_code = 256 * int(data_bytes[0]) + int(data_bytes[1])
695 reason_bytes = data_bytes[2:]
697 reason: Optional[str]
698 if not reason_bytes:
699 reason = None
700 else:
701 try:
702 reason = reason_bytes.decode("utf-8")
703 except UnicodeDecodeError:
704 reason = reason_bytes.decode("utf-8", errors="replace")
706 return (close_status_code, reason)
708 def _callback(self, callback: Optional[Callable], *args: Any) -> None:
709 if callback:
710 try:
711 callback(self, *args)
713 except Exception as e:
714 error(f"error from callback {callback}: {e}", exc_info=True)
715 # Bug fix: Prevent infinite recursion by not calling on_error
716 # when the failing callback IS on_error itself
717 if self.on_error and callback is not self.on_error:
718 self.on_error(self, e)