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 on_close: function
109 Callback object which is called when connection is closed.
110 on_close has 3 arguments.
111 The 1st argument is this class object.
112 The 2nd argument is close_status_code.
113 The 3rd argument is close_msg.
114 on_cont_message: function
115 Callback object which is called when a continuation
116 frame is received.
117 on_cont_message has 3 arguments.
118 The 1st argument is this class object.
119 The 2nd argument is utf-8 string which we get from the server.
120 The 3rd argument is continue flag. if 0, the data continue
121 to next frame data
122 on_data: function
123 Callback object which is called when a message received.
124 This is called before on_message or on_cont_message,
125 and then on_message or on_cont_message is called.
126 on_data has 4 argument.
127 The 1st argument is this class object.
128 The 2nd argument is utf-8 string which we get from the server.
129 The 3rd argument is data type. ABNF.OPCODE_TEXT or ABNF.OPCODE_BINARY will be came.
130 The 4th argument is continue flag. If 0, the data continue
131 keep_running: bool
132 This parameter is obsolete and ignored.
133 get_mask_key: function
134 A callable function to get new mask keys, see the
135 WebSocket.set_mask_key's docstring for more information.
136 cookie: str
137 Cookie value.
138 subprotocols: list
139 List of available sub protocols. Default is None.
140 socket: socket
141 Pre-initialized stream socket.
142 """
143 self.url = url
144 self.header = header if header is not None else []
145 self.cookie = cookie
147 self.on_open = on_open
148 self.on_reconnect = on_reconnect
149 self.on_message = on_message
150 self.on_data = on_data
151 self.on_error = on_error
152 self.on_close = on_close
153 self.on_ping = on_ping
154 self.on_pong = on_pong
155 self.on_cont_message = on_cont_message
156 self.keep_running = False
157 self.get_mask_key = get_mask_key
158 self.sock: Optional[WebSocket] = None
159 self.last_ping_tm = float(0)
160 self.last_pong_tm = float(0)
161 self.ping_thread: Optional[threading.Thread] = None
162 self.stop_ping: Optional[threading.Event] = None
163 self.ping_interval = float(0)
164 self.ping_timeout: Optional[Union[float, int]] = None
165 self.ping_payload = ""
166 self.subprotocols = subprotocols
167 self.prepared_socket = socket
168 self.has_errored = False
169 self.has_done_teardown = False
170 self.has_done_teardown_lock = threading.Lock()
171 self.last_close_frame: Optional[ABNF] = None
173 def send(self, data: Union[bytes, str], opcode: int = ABNF.OPCODE_TEXT) -> None:
174 """
175 send message
177 Parameters
178 ----------
179 data: str
180 Message to send. If you set opcode to OPCODE_TEXT,
181 data must be utf-8 string or unicode.
182 opcode: int
183 Operation code of data. Default is OPCODE_TEXT.
184 """
186 if not self.sock or self.sock.send(data, opcode) == 0:
187 raise WebSocketConnectionClosedException("Connection is already closed.")
189 def send_text(self, text_data: str) -> None:
190 """
191 Sends UTF-8 encoded text.
192 """
193 if not self.sock or self.sock.send(text_data, ABNF.OPCODE_TEXT) == 0:
194 raise WebSocketConnectionClosedException("Connection is already closed.")
196 def send_bytes(self, data: Union[bytes, bytearray]) -> None:
197 """
198 Sends a sequence of bytes.
199 """
200 if not self.sock or self.sock.send(data, ABNF.OPCODE_BINARY) == 0:
201 raise WebSocketConnectionClosedException("Connection is already closed.")
203 def close(self, **kwargs: Any) -> None:
204 """
205 Close websocket connection.
206 """
207 self.keep_running = False
208 sock = self.sock
209 if sock:
210 sock.close(**kwargs)
211 # Capture the peer's close frame before clearing socket reference
212 if sock.close_frame is not None:
213 self.last_close_frame = sock.close_frame
214 self.sock = None
216 def _start_ping_thread(self) -> None:
217 self.last_ping_tm = self.last_pong_tm = float(0)
218 self.stop_ping = threading.Event()
219 self.ping_thread = threading.Thread(target=self._send_ping)
220 self.ping_thread.daemon = True
221 self.ping_thread.start()
223 def _stop_ping_thread(self) -> None:
224 if self.stop_ping:
225 self.stop_ping.set()
226 if self.ping_thread and self.ping_thread.is_alive():
227 self.ping_thread.join(3)
228 # Handle thread leak - if thread doesn't terminate within timeout,
229 # force cleanup and log warning instead of abandoning the thread
230 if self.ping_thread.is_alive():
231 warning(
232 "Ping thread failed to terminate within 3 seconds, "
233 "forcing cleanup. Thread may be blocked."
234 )
235 # Force cleanup by clearing references even if thread is still alive
236 # The daemon thread will eventually be cleaned up by Python's GC
237 # but we prevent resource leaks by not holding references
239 # Always clean up references regardless of thread state
240 self.ping_thread = None
241 self.stop_ping = None
242 self.last_ping_tm = self.last_pong_tm = float(0)
244 def _send_ping(self) -> None:
245 if self.stop_ping is None:
246 return
247 if self.keep_running is False:
248 return
249 while not self.stop_ping.wait(self.ping_interval) and self.keep_running is True:
250 if self.sock:
251 self.last_ping_tm = time.time()
252 try:
253 debug("Sending ping")
254 self.sock.ping(self.ping_payload)
255 except Exception as e:
256 debug(f"Failed to send ping: {e}")
258 def ready(self):
259 return self.sock and self.sock.connected
261 def run_forever(
262 self,
263 sockopt: Optional[list] = None,
264 sslopt: Optional[dict] = None,
265 ping_interval: Union[float, int] = 0,
266 ping_timeout: Optional[Union[float, int]] = None,
267 ping_payload: str = "",
268 http_proxy_host: Optional[str] = None,
269 http_proxy_port: Optional[Union[int, str]] = None,
270 http_no_proxy: Optional[list] = None,
271 http_proxy_auth: Optional[tuple] = None,
272 http_proxy_timeout: Optional[float] = None,
273 skip_utf8_validation: bool = False,
274 host: Optional[str] = None,
275 origin: Optional[str] = None,
276 dispatcher: Any = None,
277 suppress_origin: bool = False,
278 suppress_host: bool = False,
279 proxy_type: Optional[str] = None,
280 reconnect: Optional[int] = None,
281 ) -> bool:
282 """
283 Run event loop for WebSocket framework.
285 This loop is an infinite loop and is alive while websocket is available.
287 Parameters
288 ----------
289 sockopt: tuple
290 Values for socket.setsockopt.
291 sockopt must be tuple
292 and each element is argument of sock.setsockopt.
293 sslopt: dict
294 Optional dict object for ssl socket option.
295 ping_interval: int or float
296 Automatically send "ping" command
297 every specified period (in seconds).
298 If set to 0, no ping is sent periodically.
299 ping_timeout: int or float
300 Timeout (in seconds) if the pong message is not received.
301 ping_payload: str
302 Payload message to send with each ping.
303 http_proxy_host: str
304 HTTP proxy host name.
305 http_proxy_port: int or str
306 HTTP proxy port. Required when http_proxy_host is set. Proxies
307 from environment variables default to port 80.
308 http_no_proxy: list
309 Whitelisted host names that don't use the proxy.
310 http_proxy_timeout: int or float
311 HTTP proxy timeout, default is 60 sec as per python-socks.
312 http_proxy_auth: tuple
313 HTTP proxy auth information. tuple of username and password. Default is None.
314 skip_utf8_validation: bool
315 skip utf8 validation.
316 host: str
317 update host header.
318 origin: str
319 update origin header.
320 dispatcher: Dispatcher object
321 customize reading data from socket.
322 suppress_origin: bool
323 suppress outputting origin header.
324 suppress_host: bool
325 suppress outputting host header.
326 proxy_type: str
327 type of proxy from: http, socks4, socks4a, socks5, socks5h
328 reconnect: int
329 delay interval when reconnecting
331 Returns
332 -------
333 teardown: bool
334 False if the `WebSocketApp` is closed or caught KeyboardInterrupt,
335 True if any other exception was raised during a loop.
336 """
338 if reconnect is None:
339 reconnect = RECONNECT
341 if ping_timeout is not None and ping_timeout <= 0:
342 raise WebSocketException("Ensure ping_timeout > 0")
343 if ping_interval is not None and ping_interval < 0:
344 raise WebSocketException("Ensure ping_interval >= 0")
345 if ping_timeout and ping_interval and ping_interval <= ping_timeout:
346 raise WebSocketException("Ensure ping_interval > ping_timeout")
347 if not sockopt:
348 sockopt = []
349 if not sslopt:
350 sslopt = {}
351 if self.sock:
352 raise WebSocketException("socket is already opened")
354 self.ping_interval = ping_interval
355 self.ping_timeout = ping_timeout
356 self.ping_payload = ping_payload
357 self.has_done_teardown = False
358 self.has_errored = False
359 self.keep_running = True
361 def teardown(close_frame: Optional[ABNF] = None) -> None:
362 """
363 Tears down the connection.
365 Parameters
366 ----------
367 close_frame: ABNF frame
368 If close_frame is set, the on_close handler is invoked
369 with the statusCode and reason from the provided frame.
370 """
372 # teardown() is called in many code paths to ensure resources are cleaned up and on_close is fired.
373 # To ensure the work is only done once, we use this bool and lock.
374 with self.has_done_teardown_lock:
375 if self.has_done_teardown:
376 return
377 self.has_done_teardown = True
379 self._stop_ping_thread()
380 self.keep_running = False
382 if self.sock:
383 # in cases like handleDisconnect, the "on_error" callback is called first. If the WebSocketApp
384 # is being used in a multithreaded application, we nee to make sure that "self.sock" is cleared
385 # before calling close, otherwise logic built around the sock being set can cause issues -
386 # specifically calling "run_forever" again, since is checks if "self.sock" is set.
387 current_sock = self.sock
388 self.sock = None
389 current_sock.close()
391 # Use stored close frame as fallback if none provided (e.g., client-initiated close)
392 effective_close_frame = (
393 close_frame if close_frame else self.last_close_frame
394 )
395 close_status_code, close_reason = self._get_close_args(
396 effective_close_frame
397 )
398 # Finally call the callback AFTER all teardown is complete
399 self._callback(self.on_close, close_status_code, close_reason)
401 def initialize_socket(reconnecting: bool = False) -> None:
402 if reconnecting and self.sock:
403 self.sock.shutdown()
405 # Reset close frame to avoid stale data from previous connections
406 self.last_close_frame = None
408 self.sock = WebSocket(
409 self.get_mask_key,
410 sockopt=sockopt,
411 sslopt=sslopt,
412 fire_cont_frame=self.on_cont_message is not None,
413 skip_utf8_validation=skip_utf8_validation,
414 enable_multithread=True,
415 dispatcher=dispatcher,
416 )
418 self.sock.settimeout(getdefaulttimeout())
419 try:
420 header = self.header() if callable(self.header) else self.header
422 self.sock.connect(
423 self.url,
424 header=header,
425 cookie=self.cookie,
426 http_proxy_host=http_proxy_host,
427 http_proxy_port=http_proxy_port,
428 http_no_proxy=http_no_proxy,
429 http_proxy_auth=http_proxy_auth,
430 http_proxy_timeout=http_proxy_timeout,
431 subprotocols=self.subprotocols,
432 host=host,
433 origin=origin,
434 suppress_origin=suppress_origin,
435 suppress_host=suppress_host,
436 proxy_type=proxy_type,
437 socket=self.prepared_socket,
438 )
440 info("Websocket connected")
442 if self.ping_interval:
443 self._start_ping_thread()
445 if reconnecting and self.on_reconnect:
446 self._callback(self.on_reconnect)
447 else:
448 self._callback(self.on_open)
450 assert dispatcher is not None
451 dispatcher.read(self.sock.sock, read, check)
452 except (
453 WebSocketConnectionClosedException,
454 ConnectionRefusedError,
455 KeyboardInterrupt,
456 SystemExit,
457 Exception,
458 ) as e:
459 handleDisconnect(e, reconnecting)
461 def read() -> bool:
462 if not self.keep_running:
463 teardown()
464 return False
466 if self.sock is None:
467 return False
469 try:
470 op_code, frame = self.sock.recv_data_frame(True)
471 except (
472 WebSocketConnectionClosedException,
473 KeyboardInterrupt,
474 SSLError,
475 ConnectionResetError,
476 WebSocketTimeoutException,
477 ) as e:
478 if custom_dispatcher:
479 return closed(e)
480 else:
481 raise e
483 if op_code == ABNF.OPCODE_CLOSE:
484 return closed(frame)
485 elif op_code == ABNF.OPCODE_PING:
486 self._callback(self.on_ping, frame.data)
487 elif op_code == ABNF.OPCODE_PONG:
488 self.last_pong_tm = time.time()
489 self._callback(self.on_pong, frame.data)
490 elif op_code == ABNF.OPCODE_CONT and self.on_cont_message:
491 self._callback(self.on_data, frame.data, frame.opcode, frame.fin)
492 self._callback(self.on_cont_message, frame.data, frame.fin)
493 else:
494 data = frame.data
495 if op_code == ABNF.OPCODE_TEXT and not skip_utf8_validation:
496 data = data.decode("utf-8")
497 self._callback(self.on_data, data, frame.opcode, True)
498 self._callback(self.on_message, data)
500 return True
502 def check() -> bool:
503 if self.ping_timeout:
504 has_timeout_expired = (
505 time.time() - self.last_ping_tm > self.ping_timeout
506 )
507 has_pong_not_arrived_after_last_ping = (
508 self.last_pong_tm - self.last_ping_tm < 0
509 )
510 has_pong_arrived_too_late = (
511 self.last_pong_tm - self.last_ping_tm > self.ping_timeout
512 )
514 if (
515 self.last_ping_tm
516 and has_timeout_expired
517 and (
518 has_pong_not_arrived_after_last_ping
519 or has_pong_arrived_too_late
520 )
521 ):
522 raise WebSocketTimeoutException("ping/pong timed out")
523 return True
525 def closed(
526 e: Union[
527 WebSocketConnectionClosedException,
528 ConnectionRefusedError,
529 KeyboardInterrupt,
530 SystemExit,
531 Exception,
532 str,
533 "ABNF", # Now explicitly handle ABNF frame objects
534 ] = "closed unexpectedly",
535 ) -> bool:
536 close_frame: Optional[ABNF] = None
537 if type(e) is str:
538 e = WebSocketConnectionClosedException(e)
539 elif isinstance(e, ABNF) and e.opcode == ABNF.OPCODE_CLOSE:
540 close_frame = e
541 # Convert close frames to a descriptive exception for on_error callback
542 close_status_code, close_reason = self._parse_close_frame(e)
543 reason_parts: List[str] = []
544 if close_status_code is None:
545 message = "Connection closed"
546 elif close_status_code == 1000:
547 message = "Connection closed normally (code 1000)"
548 else:
549 message = f"Connection closed (code {close_status_code})"
550 if close_reason:
551 reason_parts.append(close_reason)
552 if reason_parts:
553 message = f"{message}: {'; '.join(reason_parts)}"
554 converted = WebSocketConnectionClosedException(message)
555 setattr(converted, "status_code", close_status_code)
556 setattr(converted, "reason", close_reason)
557 e = converted
558 return handleDisconnect(e, bool(reconnect), close_frame=close_frame) # type: ignore[arg-type]
560 def handleDisconnect(
561 e: Union[
562 WebSocketConnectionClosedException,
563 ConnectionRefusedError,
564 KeyboardInterrupt,
565 SystemExit,
566 Exception,
567 ],
568 reconnecting: bool = False,
569 close_frame: Optional[ABNF] = None,
570 ) -> bool:
571 if close_frame is None:
572 self.has_errored = True
573 self._stop_ping_thread()
574 if not reconnecting:
575 self._callback(self.on_error, e)
577 if isinstance(e, (KeyboardInterrupt, SystemExit)):
578 teardown(close_frame)
579 # Propagate further
580 raise
582 if reconnect:
583 info(f"{e} - reconnect")
584 if custom_dispatcher:
585 debug(
586 f"Calling custom dispatcher reconnect [{len(inspect.stack())} frames in stack]"
587 )
588 assert dispatcher is not None
589 dispatcher.reconnect(reconnect, initialize_socket)
590 else:
591 error(f"{e} - goodbye")
592 teardown(close_frame)
593 return self.has_errored
595 custom_dispatcher = bool(dispatcher)
596 dispatcher = self.create_dispatcher(
597 ping_timeout, dispatcher, parse_url(self.url)[3], closed
598 )
600 try:
601 initialize_socket()
602 if not custom_dispatcher and reconnect:
603 while self.keep_running:
604 debug(
605 f"Calling dispatcher reconnect [{len(inspect.stack())} frames in stack]"
606 )
607 dispatcher.reconnect(reconnect, initialize_socket)
608 except (KeyboardInterrupt, Exception) as e:
609 info(f"tearing down on exception {e}")
610 teardown()
611 finally:
612 if not custom_dispatcher:
613 # Ensure teardown was called before returning from run_forever
614 teardown()
616 return self.has_errored
618 def create_dispatcher(
619 self,
620 ping_timeout: Optional[Union[float, int]],
621 dispatcher: Optional[DispatcherBase] = None,
622 is_ssl: bool = False,
623 handleDisconnect: Optional[Callable] = None,
624 ) -> Union[Dispatcher, SSLDispatcher, WrappedDispatcher]:
625 if dispatcher: # If custom dispatcher is set, use WrappedDispatcher
626 return WrappedDispatcher(self, ping_timeout, dispatcher, handleDisconnect)
627 timeout = ping_timeout or 10
628 if is_ssl:
629 return SSLDispatcher(self, timeout)
630 return Dispatcher(self, timeout)
632 def _get_close_args(
633 self, close_frame: Optional[ABNF]
634 ) -> List[Optional[Union[int, str]]]:
635 """
636 _get_close_args extracts the close code and reason from the close body
637 if it exists (RFC6455 says WebSocket Connection Close Code is optional)
638 """
639 # Need to catch the case where close_frame is None
640 # Otherwise the following if statement causes an error
641 if not close_frame:
642 return [None, None]
643 close_status_code, reason = self._parse_close_frame(close_frame)
644 if not self.on_close:
645 return [None, None]
646 return [close_status_code, reason]
648 def _parse_close_frame(
649 self, close_frame: Optional[ABNF]
650 ) -> Tuple[Optional[int], Optional[str]]:
651 """
652 Parse a close frame into status code and UTF-8 reason text.
653 """
654 if not close_frame or not getattr(close_frame, "data", None):
655 return (None, None)
657 data = close_frame.data
658 if isinstance(data, bytes):
659 data_bytes = data
660 elif isinstance(data, str):
661 data_bytes = data.encode("utf-8")
662 else:
663 data_bytes = bytes(data)
665 if len(data_bytes) < 2:
666 return (None, None)
668 close_status_code = 256 * int(data_bytes[0]) + int(data_bytes[1])
669 reason_bytes = data_bytes[2:]
671 reason: Optional[str]
672 if not reason_bytes:
673 reason = None
674 else:
675 try:
676 reason = reason_bytes.decode("utf-8")
677 except UnicodeDecodeError:
678 reason = reason_bytes.decode("utf-8", errors="replace")
680 return (close_status_code, reason)
682 def _callback(self, callback: Optional[Callable], *args: Any) -> None:
683 if callback:
684 try:
685 callback(self, *args)
687 except Exception as e:
688 error(f"error from callback {callback}: {e}")
689 # Bug fix: Prevent infinite recursion by not calling on_error
690 # when the failing callback IS on_error itself
691 if self.on_error and callback is not self.on_error:
692 self.on_error(self, e)