def testInternalRecvStrict(self):
        sock = ws.WebSocket()
        s = sock.sock = SockMock()
        s.add_packet(six.b("foo"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("bar"))
        # s.add_packet(SSLError("The read operation timed out"))
        s.add_packet(six.b("baz"))
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.frame_buffer.recv_strict(9)
        # if six.PY2:
        #     with self.assertRaises(ws.WebSocketTimeoutException):
        #         data = sock._recv_strict(9)
        # else:
        #     with self.assertRaises(SSLError):
        #         data = sock._recv_strict(9)
        data = sock.frame_buffer.recv_strict(9)
        self.assertEqual(data, six.b("foobarbaz"))
        with self.assertRaises(ws.WebSocketConnectionClosedException):
            sock.frame_buffer.recv_strict(1) 

def testRecvTimeout(self):
        sock = ws.WebSocket()
        s = sock.sock = SockMock()
        s.add_packet(six.b("\x81"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("\x8dabcd\x29\x07\x0f\x08\x0e"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("\x4e\x43\x33\x0e\x10\x0f\x00\x40"))
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.recv()
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.recv()
        data = sock.recv()
        self.assertEqual(data, "Hello, World!")
        with self.assertRaises(ws.WebSocketConnectionClosedException):
            sock.recv() 

def testInternalRecvStrict(self):
        sock = ws.WebSocket()
        s = sock.sock = SockMock()
        s.add_packet(six.b("foo"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("bar"))
        # s.add_packet(SSLError("The read operation timed out"))
        s.add_packet(six.b("baz"))
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.frame_buffer.recv_strict(9)
        # if six.PY2:
        #     with self.assertRaises(ws.WebSocketTimeoutException):
        #         data = sock._recv_strict(9)
        # else:
        #     with self.assertRaises(SSLError):
        #         data = sock._recv_strict(9)
        data = sock.frame_buffer.recv_strict(9)
        self.assertEqual(data, six.b("foobarbaz"))
        with self.assertRaises(ws.WebSocketConnectionClosedException):
            sock.frame_buffer.recv_strict(1) 

def testRecvTimeout(self):
        sock = ws.WebSocket()
        s = sock.sock = SockMock()
        s.add_packet(six.b("\x81"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("\x8dabcd\x29\x07\x0f\x08\x0e"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("\x4e\x43\x33\x0e\x10\x0f\x00\x40"))
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.recv()
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.recv()
        data = sock.recv()
        self.assertEqual(data, "Hello, World!")
        with self.assertRaises(ws.WebSocketConnectionClosedException):
            sock.recv() 

def _data_thread(self):
        try:
            conn = create_connection(self.addr)
        except Exception:
            self._controller_q.put('restart_data')
            return

        while self.running:
            try:
                data = conn.recv()
                data = json.loads(data)
            except WebSocketTimeoutException:
                self._controller_q.put('restart_data')
                return
            try:
                pair = data['MarketDataIncrementalRefresh']['symbol']
                endpoint = 'MarketDataIncrementalRefresh'
            except KeyError:
                pair = data['MarketDataSnapshotFullRefresh']['symbol']
                endpoint = 'MarketDataSnapshotFullRefresh'
            self.data_q.put((endpoint, pair, data[endpoint], time.time())) 

def _trade_thread(self):
        try:
            conn = create_connection(self.trader_addr)
        except Exception:
            log.exception('Trader Thread Error!')
            self._controller_q.put('restart_trader')
            return

        while self.running:
            try:
                data = conn.recv()
            except WebSocketTimeoutException:
                self._controller_q.put('restart_data')
                return
            self.data_q.put(json.loads(data))

            try:
                payload = self.trade_command_q.get()
            except Empty:
                continue

            try:
                conn.send(self.sign(payload))
            except (WebSocketTimeoutException, ConnectionResetError):
                continue 

def __connect(self, wsURL, symbol):
        '''Connect to the websocket in a thread.'''
        self.logger.debug("Starting thread")

        self.ws = websocket.WebSocketApp(wsURL,
                                         on_message=self.__on_message,
                                         on_close=self.__on_close,
                                         on_open=self.__on_open,
                                         on_error=self.__on_error,
                                         header=self.__get_auth())

        self.wst = threading.Thread(target=lambda: self.ws.run_forever())
        self.wst.daemon = True
        self.wst.start()
        self.logger.debug("Started thread")

        # Wait for connect before continuing
        conn_timeout = 5
        while not self.ws.sock or not self.ws.sock.connected and conn_timeout:
            sleep(1)
            conn_timeout -= 1
        if not conn_timeout:
            self.logger.error("Couldn't connect to WS! Exiting.")
            self.exit()
            raise websocket.WebSocketTimeoutException('Couldn\'t connect to WS! Exiting.') 

def testInternalRecvStrict(self):
        sock = ws.WebSocket()
        s = sock.sock = SockMock()
        s.add_packet(six.b("foo"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("bar"))
        # s.add_packet(SSLError("The read operation timed out"))
        s.add_packet(six.b("baz"))
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.frame_buffer.recv_strict(9)
        # if six.PY2:
        #     with self.assertRaises(ws.WebSocketTimeoutException):
        #         data = sock._recv_strict(9)
        # else:
        #     with self.assertRaises(SSLError):
        #         data = sock._recv_strict(9)
        data = sock.frame_buffer.recv_strict(9)
        self.assertEqual(data, six.b("foobarbaz"))
        with self.assertRaises(ws.WebSocketConnectionClosedException):
            sock.frame_buffer.recv_strict(1) 

def testRecvTimeout(self):
        sock = ws.WebSocket()
        s = sock.sock = SockMock()
        s.add_packet(six.b("\x81"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("\x8dabcd\x29\x07\x0f\x08\x0e"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("\x4e\x43\x33\x0e\x10\x0f\x00\x40"))
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.recv()
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.recv()
        data = sock.recv()
        self.assertEqual(data, "Hello, World!")
        with self.assertRaises(ws.WebSocketConnectionClosedException):
            sock.recv() 

def recv(self):
        if not self.client.connected:
            self._try_to_establish_connection()
        while self._is_active():
            try:
                data = self.client.recv()
                self._on_successful_connect()
                return data
            except WebSocketTimeoutException:
                raise
            except WebSocketConnectionClosedException:
                if self._is_active():
                    self._handle_lost_connection()
                else:
                    raise
            except WebsocketNotConnectedException:
                if self._is_active():
                    self._handle_lost_connection()
            except Exception:
                if self._is_active():
                    self._handle_lost_connection() 

def __connect_to_tab(self, tab_key):
		assert tab_key not in self.soclist

		cr_tab_meta = self._get_cr_tab_meta_for_key(tab_key)

		if not 'webSocketDebuggerUrl' in cr_tab_meta:
			raise cr_exceptions.ChromeConnectFailure("Tab %s has no 'webSocketDebuggerUrl' (%s)" % (tab_key, self.tablist))

		wsurl = cr_tab_meta['webSocketDebuggerUrl']

		try:
			self.log.info("Setting up websocket connection for key '%s'", tab_key)
			self.soclist[tab_key] = websocket.create_connection(wsurl)
			self.soclist[tab_key].settimeout(self.websocket_timeout)

		except (socket.timeout, websocket.WebSocketTimeoutException):
			raise cr_exceptions.ChromeCommunicationsError("Could not connect to remote chromium.") 

def ___recv(self, tab_key, timeout=None):

		try:
			if timeout:
				self.soclist[tab_key].settimeout(timeout)

			tmp = self.soclist[tab_key].recv()
			self.log.debug("		Received: '%s'", tmp)

			decoded = json.loads(tmp)
			return decoded
		except (socket.timeout, websocket.WebSocketTimeoutException):
			return None
		except websocket.WebSocketConnectionClosedException:
			raise cr_exceptions.ChromeCommunicationsError("Websocket appears to have been closed. Is the"
				" remote chromium instance dead?")
		finally:
			self.soclist[tab_key].settimeout(self.websocket_timeout) 

def testInternalRecvStrict(self):
        sock = ws.WebSocket()
        s = sock.sock = SockMock()
        s.add_packet(six.b("foo"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("bar"))
        # s.add_packet(SSLError("The read operation timed out"))
        s.add_packet(six.b("baz"))
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.frame_buffer.recv_strict(9)
        # if six.PY2:
        #     with self.assertRaises(ws.WebSocketTimeoutException):
        #         data = sock._recv_strict(9)
        # else:
        #     with self.assertRaises(SSLError):
        #         data = sock._recv_strict(9)
        data = sock.frame_buffer.recv_strict(9)
        self.assertEqual(data, six.b("foobarbaz"))
        with self.assertRaises(ws.WebSocketConnectionClosedException):
            sock.frame_buffer.recv_strict(1) 

def testRecvTimeout(self):
        sock = ws.WebSocket()
        s = sock.sock = SockMock()
        s.add_packet(six.b("\x81"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("\x8dabcd\x29\x07\x0f\x08\x0e"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("\x4e\x43\x33\x0e\x10\x0f\x00\x40"))
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.recv()
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.recv()
        data = sock.recv()
        self.assertEqual(data, "Hello, World!")
        with self.assertRaises(ws.WebSocketConnectionClosedException):
            sock.recv() 

def testInternalRecvStrict(self):
        sock = ws.WebSocket()
        s = sock.sock = SockMock()
        s.add_packet(six.b("foo"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("bar"))
        # s.add_packet(SSLError("The read operation timed out"))
        s.add_packet(six.b("baz"))
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.frame_buffer.recv_strict(9)
        # if six.PY2:
        #     with self.assertRaises(ws.WebSocketTimeoutException):
        #         data = sock._recv_strict(9)
        # else:
        #     with self.assertRaises(SSLError):
        #         data = sock._recv_strict(9)
        data = sock.frame_buffer.recv_strict(9)
        self.assertEqual(data, six.b("foobarbaz"))
        with self.assertRaises(ws.WebSocketConnectionClosedException):
            sock.frame_buffer.recv_strict(1) 

def testRecvTimeout(self):
        sock = ws.WebSocket()
        s = sock.sock = SockMock()
        s.add_packet(six.b("\x81"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("\x8dabcd\x29\x07\x0f\x08\x0e"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("\x4e\x43\x33\x0e\x10\x0f\x00\x40"))
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.recv()
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.recv()
        data = sock.recv()
        self.assertEqual(data, "Hello, World!")
        with self.assertRaises(ws.WebSocketConnectionClosedException):
            sock.recv() 

def testInternalRecvStrict(self):
        sock = ws.WebSocket()
        s = sock.sock = SockMock()
        s.add_packet(six.b("foo"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("bar"))
        # s.add_packet(SSLError("The read operation timed out"))
        s.add_packet(six.b("baz"))
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.frame_buffer.recv_strict(9)
        # if six.PY2:
        #     with self.assertRaises(ws.WebSocketTimeoutException):
        #         data = sock._recv_strict(9)
        # else:
        #     with self.assertRaises(SSLError):
        #         data = sock._recv_strict(9)
        data = sock.frame_buffer.recv_strict(9)
        self.assertEqual(data, six.b("foobarbaz"))
        with self.assertRaises(ws.WebSocketConnectionClosedException):
            sock.frame_buffer.recv_strict(1) 

def testRecvTimeout(self):
        sock = ws.WebSocket()
        s = sock.sock = SockMock()
        s.add_packet(six.b("\x81"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("\x8dabcd\x29\x07\x0f\x08\x0e"))
        s.add_packet(socket.timeout())
        s.add_packet(six.b("\x4e\x43\x33\x0e\x10\x0f\x00\x40"))
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.recv()
        with self.assertRaises(ws.WebSocketTimeoutException):
            sock.recv()
        data = sock.recv()
        self.assertEqual(data, "Hello, World!")
        with self.assertRaises(ws.WebSocketConnectionClosedException):
            sock.recv() 

def __connect(self, wsURL, symbol):
        '''Connect to the websocket in a thread.'''
        self.logger.debug("Starting thread")

        self.ws = websocket.WebSocketApp(wsURL,
                                         on_message=self.__on_message,
                                         on_close=self.__on_close,
                                         on_open=self.__on_open,
                                         on_error=self.__on_error,
                                         header=None)

        self.wst = threading.Thread(target=lambda: self.ws.run_forever())
        self.wst.daemon = True
        self.wst.start()
        self.logger.debug("Started thread")

        # Wait for connect before continuing
        conn_timeout = 5
        while not self.ws.sock or not self.ws.sock.connected and conn_timeout:
            sleep(1)
            conn_timeout -= 1
        if not conn_timeout:
            self.logger.error("Couldn't connect to WS! Exiting.")
            self.exit()
            raise websocket.WebSocketTimeoutException('Couldn\'t connect to WS! Exiting.') 

def __connect(self, wsURL, symbol):
        '''Connect to the websocket in a thread.'''
        self.logger.debug("Starting thread")

        self.ws = websocket.WebSocketApp(wsURL,
                                         on_message=self.__on_message,
                                         on_close=self.__on_close,
                                         on_open=self.__on_open,
                                         on_error=self.__on_error,
                                         header=self.__get_auth())

        self.wst = threading.Thread(target=lambda: self.ws.run_forever())
        self.wst.setDaemon(True)
        self.wst.start()
        self.logger.debug("Started thread")

        # Wait for connect before continuing
        conn_timeout = 5
        while not self.ws.sock or not self.ws.sock.connected and conn_timeout:
            sleep(1)
            conn_timeout -= 1
        if not conn_timeout:
            self.logger.error("Couldn't connect to WS! Exiting.")
            self.exit()
            raise websocket.WebSocketTimeoutException('Couldn\'t connect to WS! Exiting.') 

def _websocket_exc_handler(method):
        def wrapper(self, *args, **kwds):
            try:
                return method(self, *args, **kwds)
            except (TimeoutError, websocket.WebSocketTimeoutException):
                self.logger.warning("Websocket timeout")
            except (ConnectionError, websocket.WebSocketException) as e:
                # If we lose the connection we must reinitialize our state
                self.logger.error(str(e))
                self._reset_state()
            except Exception as e:
                self.logger.exception("Websocket callback unhandled exception")
                self._reset_state()

        return wrapper 

def catch_websocket_connection_errors():
  """A context manager that translates websocket errors into ConnectionError."""
  try:
    yield
  except websocket.WebSocketConnectionClosedException:
    raise ConnectionError("Connection already closed. SC2 probably crashed. "
                          "Check the error log.")
  except websocket.WebSocketTimeoutException:
    raise ConnectionError("Websocket timed out.")
  except socket.error as e:
    raise ConnectionError("Socket error: %s" % e) 

def start(self):
        """
        Start the websocket client threads
        :return:
        """
        super(BitfinexWSS, self).start()

        log.info("BitfinexWSS.start(): Initializing Websocket connection..")
        while self.conn is None:
            try:
                self.conn = create_connection(self.addr, timeout=10)
            except WebSocketTimeoutException:
                self.conn = None
                print("Couldn't create websocket connection - retrying!")

        log.info("BitfinexWSS.start(): Initializing receiver thread..")
        if not self.receiver_thread:
            self.receiver_thread = Thread(target=self.receive, name='Receiver Thread')
            self.receiver_thread.start()
        else:
            log.info("BitfinexWSS.start(): Thread not started! "
                     "self.receiver_thread is populated!")

        log.info("BitfinexWSS.start(): Initializing processing thread..")
        if not self.processing_thread:
            self.processing_thread = Thread(target=self.process, name='Processing Thread')
            self.processing_thread.start()
        else:
            log.info("BitfinexWSS.start(): Thread not started! "
                     "self.processing_thread is populated!")

        self.setup_subscriptions() 

def receive(self):
        """
        Receives incoming websocket messages, and puts them on the Client queue
        for processing.
        :return:
        """
        while self.running:
            if self._receiver_lock.acquire(blocking=False):
                try:
                    raw = self.conn.recv()
                except WebSocketTimeoutException:
                    self._receiver_lock.release()
                    continue
                except WebSocketConnectionClosedException:
                    # this needs to restart the client, while keeping track
                    # of the currently subscribed channels!
                    self.conn = None
                    self._controller_q.put('restart')
                except AttributeError:
                    # self.conn is None, idle loop until shutdown of thread
                    self._receiver_lock.release()
                    continue
                msg = time.time(), json.loads(raw)
                log.debug("receiver Thread: Data Received: %s", msg)
                self.receiver_q.put(msg)
                self._receiver_lock.release()
            else:
                # The receiver_lock was locked, idling until available
                time.sleep(0.5) 

def _process_data(self):
        self.conn = create_connection(self.addr, timeout=4)
        payload = json.dumps({'type': 'subscribe', 'product_ids': self.pairs})
        self.conn.send(payload)
        while self.running:
            try:
                data = json.loads(self.conn.recv())
            except (WebSocketTimeoutException, ConnectionResetError):
                self._controller_q.put('restart')

            if 'product_id' in data:
                self.data_q.put(('order_book', data['product_id'],
                                 data, time.time()))
        self.conn = None 

def _subscription_thread(self, endpoint):
        """
        Thread Method, running the connection for each endpoint.
        :param endpoint:
        :return:
        """
        try:
            conn = create_connection(self.addr + endpoint, timeout=5)
        except WebSocketTimeoutException:
            self.restart_q.put(endpoint)
            return

        while self.threads_running[endpoint]:
            try:
                msg = conn.recv()
            except WebSocketTimeoutException:
                self._controller_q.put(endpoint)

            log.debug("%s, %s", endpoint, msg)
            ep, pair = endpoint.split('/')
            log.debug("_subscription_thread(): Putting data on q..")
            try:
                self.data_q.put((ep, pair, msg, time.time()), timeout=1)
            except TimeoutError:
                continue
            finally:
                log.debug("_subscription_thread(): Data Processed, looping back..")
        conn.close()
        log.debug("_subscription_thread(): Thread Loop Ended.") 

def run1(self, payload):
        data = None
        browser = None
        begin_time = datetime.datetime.now()
        retry = payload.get('retried', False)
        try:
            socket_timeout = payload.get('sockettimeout') or self._socket_timeout
            browser = websocket.create_connection(self._browser_url, timeout=socket_timeout)
            data = self.run1_core(payload, browser, begin_time)
            return data
        except websocket.WebSocketTimeoutException as e:
            if retry:
                error_data = {
                    'state': 'critical',
                    'error_code': -6,
                    'error_desc': str(type(e)) + ': ' + str(e)
                }
                ret = self.crawl_info(error_data, payload, begin_time)
                return ret
            else:
                sleep(payload.get('retry_sleep', 3))
                payload['sockettimeout'] = int(payload.get('sockettimeout') or self._socket_timeout) + payload.get(
                    'retry_extra', 10)
                payload['loadtimeout'] = int(payload.get('loadtimeout') or self._socket_timeout) + payload.get('retry_extra',
                                                                                                               10)
                payload['retried'] = True
                return self.run1_core(payload, browser=browser, begin_time=begin_time)
        except Exception as e:
            error_data = {
                'state': 'critical',
                'error_code': -7,
                'error_desc': str(type(e)) + ': ' + str(e)
            }
            ret = self.crawl_info(error_data, payload, begin_time)
            return ret

        finally:
            if browser is not None:
                browser.close() 

def __connect(self, wsURL, symbol):
        '''Connect to the websocket in a thread.'''
        self.logger.debug(".....Starting thread")

        self.ws = websocket.WebSocketApp(wsURL,
                                         on_message=self.__on_message,
                                         on_close=self.__on_close,
                                         on_open=self.__on_open,
                                         on_error=self.__on_error,
                                         header=self.__get_auth())

        self.wst = threading.Thread(target=lambda: self.ws.run_forever())
        self.wst.daemon = True
        self.wst.start()
        self.logger.debug("Started thread")

        # Wait for connect before continuing
        conn_timeout = 5
        while not self.ws.sock or not self.ws.sock.connected and conn_timeout:
            sleep(1)            
            conn_timeout -= 1

        if not conn_timeout:
            self.logger.error("Couldn't connect to WS! Exiting.")
            self.exit()
            raise websocket.WebSocketTimeoutException('Couldn\'t connect to WS! Exiting.')

        self.logger.debug("exit connect") 

def __connect(self, wsURL, symbol):
        '''Connect to the websocket in a thread.'''
        self.logger.debug(".....Starting thread")

        self.ws = websocket.WebSocketApp(wsURL,
                                         on_message=self.__on_message,
                                         on_close=self.__on_close,
                                         on_open=self.__on_open,
                                         on_error=self.__on_error,
                                         header=self.__get_auth())

        self.wst = threading.Thread(target=lambda: self.ws.run_forever())
        self.wst.daemon = True
        self.wst.start()
        self.logger.debug("Started thread")

        # Wait for connect before continuing
        conn_timeout = 5
        while not self.ws.sock or not self.ws.sock.connected and conn_timeout:
            sleep(1)            
            conn_timeout -= 1

        if not conn_timeout:
            self.logger.error("Couldn't connect to WS! Exiting.")
            self.exit()
            raise websocket.WebSocketTimeoutException('Couldn\'t connect to WS! Exiting.')

        self.logger.debug("exit connect") 

def recv_until_string(self, expected_string):

        ws_result = []
        ws_instance = self._tab.get("ws_instance")

        while True:
            try:
                result = ws_instance.recv()
            except websocket.WebSocketTimeoutException:
                return ws_result

            if self._debug:
                print("[DEBUG] result: {0}".format(result))
            if not result:
                return ws_result

            # 提取结果
            ws_result.append(result)

            # 调用用户定义的事件
            formatted_result = json.loads(result)
            event_name = formatted_result.get("method")
            params = formatted_result.get("params")
            self._call_event_listener(event_name, params)

            # 查找结束标志
            if expected_string in result:
                break
        return ws_result