You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
351 lines
13 KiB
351 lines
13 KiB
"""
|
|
websocket - WebSocket client library for Python
|
|
|
|
Copyright (C) 2010 Hiroki Ohtani(liris)
|
|
|
|
This library is free software; you can redistribute it and/or
|
|
modify it under the terms of the GNU Lesser General Public
|
|
License as published by the Free Software Foundation; either
|
|
version 2.1 of the License, or (at your option) any later version.
|
|
|
|
This library is distributed in the hope that it will be useful,
|
|
but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
|
|
Lesser General Public License for more details.
|
|
|
|
You should have received a copy of the GNU Lesser General Public
|
|
License along with this library; if not, write to the Free Software
|
|
Foundation, Inc., 51 Franklin Street, Fifth Floor,
|
|
Boston, MA 02110-1335 USA
|
|
|
|
"""
|
|
|
|
"""
|
|
WebSocketApp provides higher level APIs.
|
|
"""
|
|
import inspect
|
|
import select
|
|
import sys
|
|
import threading
|
|
import time
|
|
import traceback
|
|
|
|
import six
|
|
|
|
from ._abnf import ABNF
|
|
from ._core import WebSocket, getdefaulttimeout
|
|
from ._exceptions import *
|
|
from . import _logging
|
|
|
|
|
|
__all__ = ["WebSocketApp"]
|
|
|
|
class Dispatcher:
|
|
def __init__(self, app, ping_timeout):
|
|
self.app = app
|
|
self.ping_timeout = ping_timeout
|
|
|
|
def read(self, sock, read_callback, check_callback):
|
|
while self.app.sock.connected:
|
|
r, w, e = select.select(
|
|
(self.app.sock.sock, ), (), (), self.ping_timeout)
|
|
if r:
|
|
if not read_callback():
|
|
break
|
|
check_callback()
|
|
|
|
class SSLDispacther:
|
|
def __init__(self, app, ping_timeout):
|
|
self.app = app
|
|
self.ping_timeout = ping_timeout
|
|
|
|
def read(self, sock, read_callback, check_callback):
|
|
while self.app.sock.connected:
|
|
r = self.select()
|
|
if r:
|
|
if not read_callback():
|
|
break
|
|
check_callback()
|
|
|
|
def select(self):
|
|
sock = self.app.sock.sock
|
|
if sock.pending():
|
|
return [sock,]
|
|
|
|
r, w, e = select.select((sock, ), (), (), self.ping_timeout)
|
|
return r
|
|
|
|
class WebSocketApp(object):
|
|
"""
|
|
Higher level of APIs are provided.
|
|
The interface is like JavaScript WebSocket object.
|
|
"""
|
|
|
|
def __init__(self, url, header=None,
|
|
on_open=None, on_message=None, on_error=None,
|
|
on_close=None, on_ping=None, on_pong=None,
|
|
on_cont_message=None,
|
|
keep_running=True, get_mask_key=None, cookie=None,
|
|
subprotocols=None,
|
|
on_data=None):
|
|
"""
|
|
url: websocket url.
|
|
header: custom header for websocket handshake.
|
|
on_open: callable object which is called at opening websocket.
|
|
this function has one argument. The argument is this class object.
|
|
on_message: callable object which is called when received data.
|
|
on_message has 2 arguments.
|
|
The 1st argument is this class object.
|
|
The 2nd argument is utf-8 string which we get from the server.
|
|
on_error: callable object which is called when we get error.
|
|
on_error has 2 arguments.
|
|
The 1st argument is this class object.
|
|
The 2nd argument is exception object.
|
|
on_close: callable object which is called when closed the connection.
|
|
this function has one argument. The argument is this class object.
|
|
on_cont_message: callback object which is called when receive continued
|
|
frame data.
|
|
on_cont_message has 3 arguments.
|
|
The 1st argument is this class object.
|
|
The 2nd argument is utf-8 string which we get from the server.
|
|
The 3rd argument is continue flag. if 0, the data continue
|
|
to next frame data
|
|
on_data: callback object which is called when a message received.
|
|
This is called before on_message or on_cont_message,
|
|
and then on_message or on_cont_message is called.
|
|
on_data has 4 argument.
|
|
The 1st argument is this class object.
|
|
The 2nd argument is utf-8 string which we get from the server.
|
|
The 3rd argument is data type. ABNF.OPCODE_TEXT or ABNF.OPCODE_BINARY will be came.
|
|
The 4th argument is continue flag. if 0, the data continue
|
|
keep_running: this parameter is obsolete and ignored.
|
|
get_mask_key: a callable to produce new mask keys,
|
|
see the WebSocket.set_mask_key's docstring for more information
|
|
subprotocols: array of available sub protocols. default is None.
|
|
"""
|
|
self.url = url
|
|
self.header = header if header is not None else []
|
|
self.cookie = cookie
|
|
|
|
self.on_open = on_open
|
|
self.on_message = on_message
|
|
self.on_data = on_data
|
|
self.on_error = on_error
|
|
self.on_close = on_close
|
|
self.on_ping = on_ping
|
|
self.on_pong = on_pong
|
|
self.on_cont_message = on_cont_message
|
|
self.keep_running = False
|
|
self.get_mask_key = get_mask_key
|
|
self.sock = None
|
|
self.last_ping_tm = 0
|
|
self.last_pong_tm = 0
|
|
self.subprotocols = subprotocols
|
|
|
|
def send(self, data, opcode=ABNF.OPCODE_TEXT):
|
|
"""
|
|
send message.
|
|
data: message to send. If you set opcode to OPCODE_TEXT,
|
|
data must be utf-8 string or unicode.
|
|
opcode: operation code of data. default is OPCODE_TEXT.
|
|
"""
|
|
|
|
if not self.sock or self.sock.send(data, opcode) == 0:
|
|
raise WebSocketConnectionClosedException(
|
|
"Connection is already closed.")
|
|
|
|
def close(self, **kwargs):
|
|
"""
|
|
close websocket connection.
|
|
"""
|
|
self.keep_running = False
|
|
if self.sock:
|
|
self.sock.close(**kwargs)
|
|
self.sock = None
|
|
|
|
def _send_ping(self, interval, event):
|
|
while not event.wait(interval):
|
|
self.last_ping_tm = time.time()
|
|
if self.sock:
|
|
try:
|
|
self.sock.ping()
|
|
except Exception as ex:
|
|
_logging.warning("send_ping routine terminated: {}".format(ex))
|
|
break
|
|
|
|
def run_forever(self, sockopt=None, sslopt=None,
|
|
ping_interval=0, ping_timeout=None,
|
|
http_proxy_host=None, http_proxy_port=None,
|
|
http_no_proxy=None, http_proxy_auth=None,
|
|
skip_utf8_validation=False,
|
|
host=None, origin=None, dispatcher=None,
|
|
suppress_origin = False, proxy_type=None):
|
|
"""
|
|
run event loop for WebSocket framework.
|
|
This loop is infinite loop and is alive during websocket is available.
|
|
sockopt: values for socket.setsockopt.
|
|
sockopt must be tuple
|
|
and each element is argument of sock.setsockopt.
|
|
sslopt: ssl socket optional dict.
|
|
ping_interval: automatically send "ping" command
|
|
every specified period(second)
|
|
if set to 0, not send automatically.
|
|
ping_timeout: timeout(second) if the pong message is not received.
|
|
http_proxy_host: http proxy host name.
|
|
http_proxy_port: http proxy port. If not set, set to 80.
|
|
http_no_proxy: host names, which doesn't use proxy.
|
|
skip_utf8_validation: skip utf8 validation.
|
|
host: update host header.
|
|
origin: update origin header.
|
|
dispatcher: customize reading data from socket.
|
|
suppress_origin: suppress outputting origin header.
|
|
|
|
Returns
|
|
-------
|
|
False if caught KeyboardInterrupt
|
|
True if other exception was raised during a loop
|
|
"""
|
|
|
|
if ping_timeout is not None and ping_timeout <= 0:
|
|
ping_timeout = None
|
|
if ping_timeout and ping_interval and ping_interval <= ping_timeout:
|
|
raise WebSocketException("Ensure ping_interval > ping_timeout")
|
|
if not sockopt:
|
|
sockopt = []
|
|
if not sslopt:
|
|
sslopt = {}
|
|
if self.sock:
|
|
raise WebSocketException("socket is already opened")
|
|
thread = None
|
|
self.keep_running = True
|
|
self.last_ping_tm = 0
|
|
self.last_pong_tm = 0
|
|
|
|
def teardown(close_frame=None):
|
|
"""
|
|
Tears down the connection.
|
|
If close_frame is set, we will invoke the on_close handler with the
|
|
statusCode and reason from there.
|
|
"""
|
|
if thread and thread.isAlive():
|
|
event.set()
|
|
thread.join()
|
|
self.keep_running = False
|
|
if self.sock:
|
|
self.sock.close()
|
|
close_args = self._get_close_args(
|
|
close_frame.data if close_frame else None)
|
|
self._callback(self.on_close, *close_args)
|
|
self.sock = None
|
|
|
|
try:
|
|
self.sock = WebSocket(
|
|
self.get_mask_key, sockopt=sockopt, sslopt=sslopt,
|
|
fire_cont_frame=self.on_cont_message is not None,
|
|
skip_utf8_validation=skip_utf8_validation,
|
|
enable_multithread=True if ping_interval else False)
|
|
self.sock.settimeout(getdefaulttimeout())
|
|
self.sock.connect(
|
|
self.url, header=self.header, cookie=self.cookie,
|
|
http_proxy_host=http_proxy_host,
|
|
http_proxy_port=http_proxy_port, http_no_proxy=http_no_proxy,
|
|
http_proxy_auth=http_proxy_auth, subprotocols=self.subprotocols,
|
|
host=host, origin=origin, suppress_origin=suppress_origin,
|
|
proxy_type=proxy_type)
|
|
if not dispatcher:
|
|
dispatcher = self.create_dispatcher(ping_timeout)
|
|
|
|
self._callback(self.on_open)
|
|
|
|
if ping_interval:
|
|
event = threading.Event()
|
|
thread = threading.Thread(
|
|
target=self._send_ping, args=(ping_interval, event))
|
|
thread.setDaemon(True)
|
|
thread.start()
|
|
|
|
def read():
|
|
if not self.keep_running:
|
|
return teardown()
|
|
|
|
op_code, frame = self.sock.recv_data_frame(True)
|
|
if op_code == ABNF.OPCODE_CLOSE:
|
|
return teardown(frame)
|
|
elif op_code == ABNF.OPCODE_PING:
|
|
self._callback(self.on_ping, frame.data)
|
|
elif op_code == ABNF.OPCODE_PONG:
|
|
self.last_pong_tm = time.time()
|
|
self._callback(self.on_pong, frame.data)
|
|
elif op_code == ABNF.OPCODE_CONT and self.on_cont_message:
|
|
self._callback(self.on_data, frame.data,
|
|
frame.opcode, frame.fin)
|
|
self._callback(self.on_cont_message,
|
|
frame.data, frame.fin)
|
|
else:
|
|
data = frame.data
|
|
if six.PY3 and op_code == ABNF.OPCODE_TEXT:
|
|
data = data.decode("utf-8")
|
|
self._callback(self.on_data, data, frame.opcode, True)
|
|
self._callback(self.on_message, data)
|
|
|
|
return True
|
|
|
|
def check():
|
|
if (ping_timeout):
|
|
has_timeout_expired = time.time() - self.last_ping_tm > ping_timeout
|
|
has_pong_not_arrived_after_last_ping = self.last_pong_tm - self.last_ping_tm < 0
|
|
has_pong_arrived_too_late = self.last_pong_tm - self.last_ping_tm > ping_timeout
|
|
|
|
if (self.last_ping_tm
|
|
and has_timeout_expired
|
|
and (has_pong_not_arrived_after_last_ping or has_pong_arrived_too_late)):
|
|
raise WebSocketTimeoutException("ping/pong timed out")
|
|
return True
|
|
|
|
dispatcher.read(self.sock.sock, read, check)
|
|
except (Exception, KeyboardInterrupt, SystemExit) as e:
|
|
self._callback(self.on_error, e)
|
|
if isinstance(e, SystemExit):
|
|
# propagate SystemExit further
|
|
raise
|
|
teardown()
|
|
return not isinstance(e, KeyboardInterrupt)
|
|
|
|
def create_dispatcher(self, ping_timeout):
|
|
timeout = ping_timeout or 10
|
|
if self.sock.is_ssl():
|
|
return SSLDispacther(self, timeout)
|
|
|
|
return Dispatcher(self, timeout)
|
|
|
|
def _get_close_args(self, data):
|
|
""" this functions extracts the code, reason from the close body
|
|
if they exists, and if the self.on_close except three arguments """
|
|
# if the on_close callback is "old", just return empty list
|
|
if sys.version_info < (3, 0):
|
|
if not self.on_close or len(inspect.getargspec(self.on_close).args) != 3:
|
|
return []
|
|
else:
|
|
if not self.on_close or len(inspect.getfullargspec(self.on_close).args) != 3:
|
|
return []
|
|
|
|
if data and len(data) >= 2:
|
|
code = 256 * six.byte2int(data[0:1]) + six.byte2int(data[1:2])
|
|
reason = data[2:].decode('utf-8')
|
|
return [code, reason]
|
|
|
|
return [None, None]
|
|
|
|
def _callback(self, callback, *args):
|
|
if callback:
|
|
try:
|
|
if inspect.ismethod(callback):
|
|
callback(*args)
|
|
else:
|
|
callback(self, *args)
|
|
|
|
except Exception as e:
|
|
_logging.error("error from callback {}: {}".format(callback, e))
|
|
if _logging.isEnabledForDebug():
|
|
_, _, tb = sys.exc_info()
|
|
traceback.print_tb(tb)
|
|
|