src/tools/telemetry/third_party/websocket-client/websocket.py

   1 """
   2 websocket - WebSocket client library for Python
   3
   4 Copyright (C) 2010 Hiroki Ohtani(liris)
   5
   6     This library is free software; you can redistribute it and/or
   7     modify it under the terms of the GNU Lesser General Public
   8     License as published by the Free Software Foundation; either
   9     version 2.1 of the License, or (at your option) any later version.
  10
  11     This library is distributed in the hope that it will be useful,
  12     but WITHOUT ANY WARRANTY; without even the implied warranty of
  13     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14     Lesser General Public License for more details.
  15
  16     You should have received a copy of the GNU Lesser General Public
  17     License along with this library; if not, write to the Free Software
  18     Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  19
  20 """
  21
  22
  23 import socket
  24
  25 try:
  26     import ssl
  27     HAVE_SSL = True
  28 except ImportError:
  29     HAVE_SSL = False
  30
  31 from urlparse import urlparse
  32 import os
  33 import array
  34 import struct
  35 import uuid
  36 import hashlib
  37 import base64
  38 import threading
  39 import time
  40 import logging
  41 import traceback
  42 import sys
  43
  44 """
  45 websocket python client.
  46 =========================
  47
  48 This version support only hybi-13.
  49 Please see http://tools.ietf.org/html/rfc6455 for protocol.
  50 """
  51
  52
  53 # websocket supported version.
  54 VERSION = 13
  55
  56 # closing frame status codes.
  57 STATUS_NORMAL = 1000
  58 STATUS_GOING_AWAY = 1001
  59 STATUS_PROTOCOL_ERROR = 1002
  60 STATUS_UNSUPPORTED_DATA_TYPE = 1003
  61 STATUS_STATUS_NOT_AVAILABLE = 1005
  62 STATUS_ABNORMAL_CLOSED = 1006
  63 STATUS_INVALID_PAYLOAD = 1007
  64 STATUS_POLICY_VIOLATION = 1008
  65 STATUS_MESSAGE_TOO_BIG = 1009
  66 STATUS_INVALID_EXTENSION = 1010
  67 STATUS_UNEXPECTED_CONDITION = 1011
  68 STATUS_TLS_HANDSHAKE_ERROR = 1015
  69
  70 logger = logging.getLogger()
  71
  72
  73 class WebSocketException(Exception):
  74     """
  75     websocket exeception class.
  76     """
  77     pass
  78
  79
  80 class WebSocketConnectionClosedException(WebSocketException):
  81     """
  82     If remote host closed the connection or some network error happened,
  83     this exception will be raised.
  84     """
  85     pass
  86
  87 class WebSocketTimeoutException(WebSocketException):
  88     """
  89     WebSocketTimeoutException will be raised at socket timeout during read/write data.
  90     """
  91     pass
  92
  93 default_timeout = None
  94 traceEnabled = False
  95
  96
  97 def enableTrace(tracable):
  98     """
  99     turn on/off the tracability.
 100
 101     tracable: boolean value. if set True, tracability is enabled.
 102     """
 103     global traceEnabled
 104     traceEnabled = tracable
 105     if tracable:
 106         if not logger.handlers:
 107             logger.addHandler(logging.StreamHandler())
 108         logger.setLevel(logging.DEBUG)
 109
 110
 111 def setdefaulttimeout(timeout):
 112     """
 113     Set the global timeout setting to connect.
 114
 115     timeout: default socket timeout time. This value is second.
 116     """
 117     global default_timeout
 118     default_timeout = timeout
 119
 120
 121 def getdefaulttimeout():
 122     """
 123     Return the global timeout setting(second) to connect.
 124     """
 125     return default_timeout
 126
 127
 128 def _parse_url(url):
 129     """
 130     parse url and the result is tuple of
 131     (hostname, port, resource path and the flag of secure mode)
 132
 133     url: url string.
 134     """
 135     if ":" not in url:
 136         raise ValueError("url is invalid")
 137
 138     scheme, url = url.split(":", 1)
 139
 140     parsed = urlparse(url, scheme="http")
 141     if parsed.hostname:
 142         hostname = parsed.hostname
 143     else:
 144         raise ValueError("hostname is invalid")
 145     port = 0
 146     if parsed.port:
 147         port = parsed.port
 148
 149     is_secure = False
 150     if scheme == "ws":
 151         if not port:
 152             port = 80
 153     elif scheme == "wss":
 154         is_secure = True
 155         if not port:
 156             port = 443
 157     else:
 158         raise ValueError("scheme %s is invalid" % scheme)
 159
 160     if parsed.path:
 161         resource = parsed.path
 162     else:
 163         resource = "/"
 164
 165     if parsed.query:
 166         resource += "?" + parsed.query
 167
 168     return (hostname, port, resource, is_secure)
 169
 170
 171 def create_connection(url, timeout=None, **options):
 172     """
 173     connect to url and return websocket object.
 174
 175     Connect to url and return the WebSocket object.
 176     Passing optional timeout parameter will set the timeout on the socket.
 177     If no timeout is supplied, the global default timeout setting returned by getdefauttimeout() is used.
 178     You can customize using 'options'.
 179     If you set "header" list object, you can set your own custom header.
 180
 181     >>> conn = create_connection("ws://echo.websocket.org/",
 182          ...     header=["User-Agent: MyProgram",
 183          ...             "x-custom: header"])
 184
 185
 186     timeout: socket timeout time. This value is integer.
 187              if you set None for this value, it means "use default_timeout value"
 188
 189     options: current support option is only "header".
 190              if you set header as dict value, the custom HTTP headers are added.
 191     """
 192     sockopt = options.get("sockopt", [])
 193     sslopt = options.get("sslopt", {})
 194     websock = WebSocket(sockopt=sockopt, sslopt=sslopt)
 195     websock.settimeout(timeout if timeout is not None else default_timeout)
 196     websock.connect(url, **options)
 197     return websock
 198
 199 _MAX_INTEGER = (1 << 32) -1
 200 _AVAILABLE_KEY_CHARS = range(0x21, 0x2f + 1) + range(0x3a, 0x7e + 1)
 201 _MAX_CHAR_BYTE = (1<<8) -1
 202
 203 # ref. Websocket gets an update, and it breaks stuff.
 204 # http://axod.blogspot.com/2010/06/websocket-gets-update-and-it-breaks.html
 205
 206
 207 def _create_sec_websocket_key():
 208     uid = uuid.uuid4()
 209     return base64.encodestring(uid.bytes).strip()
 210
 211
 212 _HEADERS_TO_CHECK = {
 213     "upgrade": "websocket",
 214     "connection": "upgrade",
 215     }
 216
 217
 218 class ABNF(object):
 219     """
 220     ABNF frame class.
 221     see http://tools.ietf.org/html/rfc5234
 222     and http://tools.ietf.org/html/rfc6455#section-5.2
 223     """
 224
 225     # operation code values.
 226     OPCODE_TEXT   = 0x1
 227     OPCODE_BINARY = 0x2
 228     OPCODE_CLOSE  = 0x8
 229     OPCODE_PING   = 0x9
 230     OPCODE_PONG   = 0xa
 231
 232     # available operation code value tuple
 233     OPCODES = (OPCODE_TEXT, OPCODE_BINARY, OPCODE_CLOSE,
 234                 OPCODE_PING, OPCODE_PONG)
 235
 236     # opcode human readable string
 237     OPCODE_MAP = {
 238         OPCODE_TEXT: "text",
 239         OPCODE_BINARY: "binary",
 240         OPCODE_CLOSE: "close",
 241         OPCODE_PING: "ping",
 242         OPCODE_PONG: "pong"
 243         }
 244
 245     # data length threashold.
 246     LENGTH_7  = 0x7d
 247     LENGTH_16 = 1 << 16
 248     LENGTH_63 = 1 << 63
 249
 250     def __init__(self, fin=0, rsv1=0, rsv2=0, rsv3=0,
 251                  opcode=OPCODE_TEXT, mask=1, data=""):
 252         """
 253         Constructor for ABNF.
 254         please check RFC for arguments.
 255         """
 256         self.fin = fin
 257         self.rsv1 = rsv1
 258         self.rsv2 = rsv2
 259         self.rsv3 = rsv3
 260         self.opcode = opcode
 261         self.mask = mask
 262         self.data = data
 263         self.get_mask_key = os.urandom
 264
 265     @staticmethod
 266     def create_frame(data, opcode):
 267         """
 268         create frame to send text, binary and other data.
 269
 270         data: data to send. This is string value(byte array).
 271             if opcode is OPCODE_TEXT and this value is uniocde,
 272             data value is conveted into unicode string, automatically.
 273
 274         opcode: operation code. please see OPCODE_XXX.
 275         """
 276         if opcode == ABNF.OPCODE_TEXT and isinstance(data, unicode):
 277             data = data.encode("utf-8")
 278         # mask must be set if send data from client
 279         return ABNF(1, 0, 0, 0, opcode, 1, data)
 280
 281     def format(self):
 282         """
 283         format this object to string(byte array) to send data to server.
 284         """
 285         if any(x not in (0, 1) for x in [self.fin, self.rsv1, self.rsv2, self.rsv3]):
 286             raise ValueError("not 0 or 1")
 287         if self.opcode not in ABNF.OPCODES:
 288             raise ValueError("Invalid OPCODE")
 289         length = len(self.data)
 290         if length >= ABNF.LENGTH_63:
 291             raise ValueError("data is too long")
 292
 293         frame_header = chr(self.fin << 7
 294                            | self.rsv1 << 6 | self.rsv2 << 5 | self.rsv3 << 4
 295                            | self.opcode)
 296         if length < ABNF.LENGTH_7:
 297             frame_header += chr(self.mask << 7 | length)
 298         elif length < ABNF.LENGTH_16:
 299             frame_header += chr(self.mask << 7 | 0x7e)
 300             frame_header += struct.pack("!H", length)
 301         else:
 302             frame_header += chr(self.mask << 7 | 0x7f)
 303             frame_header += struct.pack("!Q", length)
 304
 305         if not self.mask:
 306             return frame_header + self.data
 307         else:
 308             mask_key = self.get_mask_key(4)
 309             return frame_header + self._get_masked(mask_key)
 310
 311     def _get_masked(self, mask_key):
 312         s = ABNF.mask(mask_key, self.data)
 313         return mask_key + "".join(s)
 314
 315     @staticmethod
 316     def mask(mask_key, data):
 317         """
 318         mask or unmask data. Just do xor for each byte
 319
 320         mask_key: 4 byte string(byte).
 321
 322         data: data to mask/unmask.
 323         """
 324         _m = array.array("B", mask_key)
 325         _d = array.array("B", data)
 326         for i in xrange(len(_d)):
 327             _d[i] ^= _m[i % 4]
 328         return _d.tostring()
 329
 330
 331 class WebSocket(object):
 332     """
 333     Low level WebSocket interface.
 334     This class is based on
 335       The WebSocket protocol draft-hixie-thewebsocketprotocol-76
 336       http://tools.ietf.org/html/draft-hixie-thewebsocketprotocol-76
 337
 338     We can connect to the websocket server and send/recieve data.
 339     The following example is a echo client.
 340
 341     >>> import websocket
 342     >>> ws = websocket.WebSocket()
 343     >>> ws.connect("ws://echo.websocket.org")
 344     >>> ws.send("Hello, Server")
 345     >>> ws.recv()
 346     'Hello, Server'
 347     >>> ws.close()
 348
 349     get_mask_key: a callable to produce new mask keys, see the set_mask_key
 350       function's docstring for more details
 351     sockopt: values for socket.setsockopt.
 352         sockopt must be tuple and each element is argument of sock.setscokopt.
 353     sslopt: dict object for ssl socket option.
 354     """
 355
 356     def __init__(self, get_mask_key=None, sockopt=None, sslopt=None):
 357         """
 358         Initalize WebSocket object.
 359         """
 360         if sockopt is None:
 361             sockopt = []
 362         if sslopt is None:
 363             sslopt = {}
 364         self.connected = False
 365         self.sock = socket.socket()
 366         for opts in sockopt:
 367             self.sock.setsockopt(*opts)
 368         self.sslopt = sslopt
 369         self.get_mask_key = get_mask_key
 370
 371     def fileno(self):
 372         return self.sock.fileno()
 373
 374     def set_mask_key(self, func):
 375         """
 376         set function to create musk key. You can custumize mask key generator.
 377         Mainly, this is for testing purpose.
 378
 379         func: callable object. the fuct must 1 argument as integer.
 380               The argument means length of mask key.
 381               This func must be return string(byte array),
 382               which length is argument specified.
 383         """
 384         self.get_mask_key = func
 385
 386     def gettimeout(self):
 387         """
 388         Get the websocket timeout(second).
 389         """
 390         return self.sock.gettimeout()
 391
 392     def settimeout(self, timeout):
 393         """
 394         Set the timeout to the websocket.
 395
 396         timeout: timeout time(second).
 397         """
 398         self.sock.settimeout(timeout)
 399
 400     timeout = property(gettimeout, settimeout)
 401
 402     def connect(self, url, **options):
 403         """
 404         Connect to url. url is websocket url scheme. ie. ws://host:port/resource
 405         You can customize using 'options'.
 406         If you set "header" dict object, you can set your own custom header.
 407
 408         >>> ws = WebSocket()
 409         >>> ws.connect("ws://echo.websocket.org/",
 410                 ...     header={"User-Agent: MyProgram",
 411                 ...             "x-custom: header"})
 412
 413         timeout: socket timeout time. This value is integer.
 414                  if you set None for this value,
 415                  it means "use default_timeout value"
 416
 417         options: current support option is only "header".
 418                  if you set header as dict value,
 419                  the custom HTTP headers are added.
 420
 421         """
 422         hostname, port, resource, is_secure = _parse_url(url)
 423         # TODO: we need to support proxy
 424         self.sock.connect((hostname, port))
 425         if is_secure:
 426             if HAVE_SSL:
 427                 if self.sslopt is None:
 428                     sslopt = {}
 429                 else:
 430                     sslopt = self.sslopt
 431                 self.sock = ssl.wrap_socket(self.sock, **sslopt)
 432             else:
 433                 raise WebSocketException("SSL not available.")
 434
 435         self._handshake(hostname, port, resource, **options)
 436
 437     def _handshake(self, host, port, resource, **options):
 438         sock = self.sock
 439         headers = []
 440         headers.append("GET %s HTTP/1.1" % resource)
 441         headers.append("Upgrade: websocket")
 442         headers.append("Connection: Upgrade")
 443         if port == 80:
 444             hostport = host
 445         else:
 446             hostport = "%s:%d" % (host, port)
 447         headers.append("Host: %s" % hostport)
 448
 449         if "origin" in options:
 450             headers.append("Origin: %s" % options["origin"])
 451         else:
 452             headers.append("Origin: http://%s" % hostport)
 453
 454         key = _create_sec_websocket_key()
 455         headers.append("Sec-WebSocket-Key: %s" % key)
 456         headers.append("Sec-WebSocket-Version: %s" % VERSION)
 457         if "header" in options:
 458             headers.extend(options["header"])
 459
 460         headers.append("")
 461         headers.append("")
 462
 463         header_str = "\r\n".join(headers)
 464         self._send(header_str)
 465         if traceEnabled:
 466             logger.debug("--- request header ---")
 467             logger.debug(header_str)
 468             logger.debug("-----------------------")
 469
 470         status, resp_headers = self._read_headers()
 471         if status != 101:
 472             self.close()
 473             raise WebSocketException("Handshake Status %d" % status)
 474
 475         success = self._validate_header(resp_headers, key)
 476         if not success:
 477             self.close()
 478             raise WebSocketException("Invalid WebSocket Header")
 479
 480         self.connected = True
 481
 482     def _validate_header(self, headers, key):
 483         for k, v in _HEADERS_TO_CHECK.iteritems():
 484             r = headers.get(k, None)
 485             if not r:
 486                 return False
 487             r = r.lower()
 488             if v != r:
 489                 return False
 490
 491         result = headers.get("sec-websocket-accept", None)
 492         if not result:
 493             return False
 494         result = result.lower()
 495
 496         value = key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
 497         hashed = base64.encodestring(hashlib.sha1(value).digest()).strip().lower()
 498         return hashed == result
 499
 500     def _read_headers(self):
 501         status = None
 502         headers = {}
 503         if traceEnabled:
 504             logger.debug("--- response header ---")
 505
 506         while True:
 507             line = self._recv_line()
 508             if line == "\r\n":
 509                 break
 510             line = line.strip()
 511             if traceEnabled:
 512                 logger.debug(line)
 513             if not status:
 514                 status_info = line.split(" ", 2)
 515                 status = int(status_info[1])
 516             else:
 517                 kv = line.split(":", 1)
 518                 if len(kv) == 2:
 519                     key, value = kv
 520                     headers[key.lower()] = value.strip().lower()
 521                 else:
 522                     raise WebSocketException("Invalid header")
 523
 524         if traceEnabled:
 525             logger.debug("-----------------------")
 526
 527         return status, headers
 528
 529     def send(self, payload, opcode=ABNF.OPCODE_TEXT):
 530         """
 531         Send the data as string.
 532
 533         payload: Payload must be utf-8 string or unicoce,
 534                   if the opcode is OPCODE_TEXT.
 535                   Otherwise, it must be string(byte array)
 536
 537         opcode: operation code to send. Please see OPCODE_XXX.
 538         """
 539         frame = ABNF.create_frame(payload, opcode)
 540         if self.get_mask_key:
 541             frame.get_mask_key = self.get_mask_key
 542         data = frame.format()
 543         if traceEnabled:
 544             logger.debug("send: " + repr(data))
 545         while data:
 546             l = self._send(data)
 547             data = data[l:]
 548
 549     def send_binary(self, payload):
 550         return self.send(payload, ABNF.OPCODE_BINARY)
 551
 552     def ping(self, payload=""):
 553         """
 554         send ping data.
 555
 556         payload: data payload to send server.
 557         """
 558         self.send(payload, ABNF.OPCODE_PING)
 559
 560     def pong(self, payload):
 561         """
 562         send pong data.
 563
 564         payload: data payload to send server.
 565         """
 566         self.send(payload, ABNF.OPCODE_PONG)
 567
 568     def recv(self):
 569         """
 570         Receive string data(byte array) from the server.
 571
 572         return value: string(byte array) value.
 573         """
 574         opcode, data = self.recv_data()
 575         return data
 576
 577     def recv_data(self):
 578         """
 579         Recieve data with operation code.
 580
 581         return  value: tuple of operation code and string(byte array) value.
 582         """
 583         while True:
 584             frame = self.recv_frame()
 585             if not frame:
 586                 # handle error:
 587                 # 'NoneType' object has no attribute 'opcode'
 588                 raise WebSocketException("Not a valid frame %s" % frame)
 589             elif frame.opcode in (ABNF.OPCODE_TEXT, ABNF.OPCODE_BINARY):
 590                 return (frame.opcode, frame.data)
 591             elif frame.opcode == ABNF.OPCODE_CLOSE:
 592                 self.send_close()
 593                 return (frame.opcode, None)
 594             elif frame.opcode == ABNF.OPCODE_PING:
 595                 self.pong(frame.data)
 596
 597     def recv_frame(self):
 598         """
 599         recieve data as frame from server.
 600
 601         return value: ABNF frame object.
 602         """
 603         header_bytes = self._recv_strict(2)
 604         if not header_bytes:
 605             return
 606         b1 = ord(header_bytes[0])
 607         fin = b1 >> 7 & 1
 608         rsv1 = b1 >> 6 & 1
 609         rsv2 = b1 >> 5 & 1
 610         rsv3 = b1 >> 4 & 1
 611         opcode = b1 & 0xf
 612         b2 = ord(header_bytes[1])
 613         mask = b2 >> 7 & 1
 614         length = b2 & 0x7f
 615
 616         length_data = ""
 617         if length == 0x7e:
 618             length_data = self._recv_strict(2)
 619             length = struct.unpack("!H", length_data)[0]
 620         elif length == 0x7f:
 621             length_data = self._recv_strict(8)
 622             length = struct.unpack("!Q", length_data)[0]
 623
 624         mask_key = ""
 625         if mask:
 626             mask_key = self._recv_strict(4)
 627         data = self._recv_strict(length)
 628         if traceEnabled:
 629             recieved = header_bytes + length_data + mask_key + data
 630             logger.debug("recv: " + repr(recieved))
 631
 632         if mask:
 633             data = ABNF.mask(mask_key, data)
 634
 635         frame = ABNF(fin, rsv1, rsv2, rsv3, opcode, mask, data)
 636         return frame
 637
 638     def send_close(self, status=STATUS_NORMAL, reason=""):
 639         """
 640         send close data to the server.
 641
 642         status: status code to send. see STATUS_XXX.
 643
 644         reason: the reason to close. This must be string.
 645         """
 646         if status < 0 or status >= ABNF.LENGTH_16:
 647             raise ValueError("code is invalid range")
 648         self.send(struct.pack('!H', status) + reason, ABNF.OPCODE_CLOSE)
 649
 650     def close(self, status=STATUS_NORMAL, reason=""):
 651         """
 652         Close Websocket object
 653
 654         status: status code to send. see STATUS_XXX.
 655
 656         reason: the reason to close. This must be string.
 657         """
 658         if self.connected:
 659             if status < 0 or status >= ABNF.LENGTH_16:
 660                 raise ValueError("code is invalid range")
 661
 662             try:
 663                 self.send(struct.pack('!H', status) + reason, ABNF.OPCODE_CLOSE)
 664                 timeout = self.sock.gettimeout()
 665                 self.sock.settimeout(3)
 666                 try:
 667                     frame = self.recv_frame()
 668                     if logger.isEnabledFor(logging.ERROR):
 669                         recv_status = struct.unpack("!H", frame.data)[0]
 670                         if recv_status != STATUS_NORMAL:
 671                             logger.error("close status: " + repr(recv_status))
 672                 except:
 673                     pass
 674                 self.sock.settimeout(timeout)
 675                 self.sock.shutdown(socket.SHUT_RDWR)
 676             except:
 677                 pass
 678         self._closeInternal()
 679
 680     def _closeInternal(self):
 681         self.connected = False
 682         self.sock.close()
 683
 684     def _send(self, data):
 685         try:
 686             return self.sock.send(data)
 687         except socket.timeout as e:
 688             raise WebSocketTimeoutException(e.message)
 689         except Exception as e:
 690             if "timed out" in e.message:
 691                 raise WebSocketTimeoutException(e.message)
 692             else:
 693                 raise e
 694
 695     def _recv(self, bufsize):
 696         try:
 697             bytes = self.sock.recv(bufsize)
 698             if not bytes:
 699                 raise WebSocketConnectionClosedException()
 700             return bytes
 701         except socket.timeout as e:
 702             raise WebSocketTimeoutException(e.message)
 703         except Exception as e:
 704             if "timed out" in e.message:
 705                 raise WebSocketTimeoutException(e.message)
 706             else:
 707                 raise e
 708
 709
 710     def _recv_strict(self, bufsize):
 711         remaining = bufsize
 712         bytes = ""
 713         while remaining:
 714             bytes += self._recv(remaining)
 715             remaining = bufsize - len(bytes)
 716
 717         return bytes
 718
 719     def _recv_line(self):
 720         line = []
 721         while True:
 722             c = self._recv(1)
 723             line.append(c)
 724             if c == "\n":
 725                 break
 726         return "".join(line)
 727
 728
 729 class WebSocketApp(object):
 730     """
 731     Higher level of APIs are provided.
 732     The interface is like JavaScript WebSocket object.
 733     """
 734     def __init__(self, url, header=[],
 735                  on_open=None, on_message=None, on_error=None,
 736                  on_close=None, keep_running=True, get_mask_key=None):
 737         """
 738         url: websocket url.
 739         header: custom header for websocket handshake.
 740         on_open: callable object which is called at opening websocket.
 741           this function has one argument. The arugment is this class object.
 742         on_message: callbale object which is called when recieved data.
 743          on_message has 2 arguments.
 744          The 1st arugment is this class object.
 745          The passing 2nd arugment is utf-8 string which we get from the server.
 746        on_error: callable object which is called when we get error.
 747          on_error has 2 arguments.
 748          The 1st arugment is this class object.
 749          The passing 2nd arugment is exception object.
 750        on_close: callable object which is called when closed the connection.
 751          this function has one argument. The arugment is this class object.
 752        keep_running: a boolean flag indicating whether the app's main loop should
 753          keep running, defaults to True
 754        get_mask_key: a callable to produce new mask keys, see the WebSocket.set_mask_key's
 755          docstring for more information
 756         """
 757         self.url = url
 758         self.header = header
 759         self.on_open = on_open
 760         self.on_message = on_message
 761         self.on_error = on_error
 762         self.on_close = on_close
 763         self.keep_running = keep_running
 764         self.get_mask_key = get_mask_key
 765         self.sock = None
 766
 767     def send(self, data, opcode=ABNF.OPCODE_TEXT):
 768         """
 769         send message.
 770         data: message to send. If you set opcode to OPCODE_TEXT, data must be utf-8 string or unicode.
 771         opcode: operation code of data. default is OPCODE_TEXT.
 772         """
 773         if self.sock.send(data, opcode) == 0:
 774             raise WebSocketConnectionClosedException()
 775
 776     def close(self):
 777         """
 778         close websocket connection.
 779         """
 780         self.keep_running = False
 781         self.sock.close()
 782
 783     def _send_ping(self, interval):
 784         while self.keep_running:
 785             time.sleep(interval)
 786             self.sock.ping()
 787
 788     def run_forever(self, sockopt=None, sslopt=None, ping_interval=0):
 789         """
 790         run event loop for WebSocket framework.
 791         This loop is infinite loop and is alive during websocket is available.
 792         sockopt: values for socket.setsockopt.
 793             sockopt must be tuple and each element is argument of sock.setscokopt.
 794         sslopt: ssl socket optional dict.
 795         ping_interval: automatically send "ping" command every specified period(second)
 796             if set to 0, not send automatically.
 797         """
 798         if sockopt is None:
 799             sockopt = []
 800         if sslopt is None:
 801             sslopt = {}
 802         if self.sock:
 803             raise WebSocketException("socket is already opened")
 804         thread = None
 805
 806         try:
 807             self.sock = WebSocket(self.get_mask_key, sockopt=sockopt, sslopt=sslopt)
 808             self.sock.connect(self.url, header=self.header)
 809             self._callback(self.on_open)
 810
 811             if ping_interval:
 812                 thread = threading.Thread(target=self._send_ping, args=(ping_interval,))
 813                 thread.setDaemon(True)
 814                 thread.start()
 815
 816             while self.keep_running:
 817                 data = self.sock.recv()
 818                 if data is None:
 819                     break
 820                 self._callback(self.on_message, data)
 821         except Exception, e:
 822             self._callback(self.on_error, e)
 823         finally:
 824             if thread:
 825                 thread.join()
 826             self.sock.close()
 827             self._callback(self.on_close)
 828             self.sock = None
 829
 830     def _callback(self, callback, *args):
 831         if callback:
 832             try:
 833                 callback(self, *args)
 834             except Exception, e:
 835                 logger.error(e)
 836                 if logger.isEnabledFor(logging.DEBUG):
 837                     _, _, tb = sys_exc_info()
 838                     traceback.print_tb(tb)
 839
 840
 841 if __name__ == "__main__":
 842     enableTrace(True)
 843     ws = create_connection("ws://echo.websocket.org/")
 844     print("Sending 'Hello, World'...")
 845     ws.send("Hello, World")
 846     print("Sent")
 847     print("Receiving...")
 848     result = ws.recv()
 849     print("Received '%s'" % result)
 850     ws.close()