def binance_test_uri_length(query):
    websocket = create_connection("wss://stream.binance.com:9443/stream?streams=" + query)
    while True:
        result = websocket.recv()
        websocket.close()
        print("Received '%s'\r\n" % result)
        break 

def ws(url, callback, json=False, wrap=False):
    '''Connect to websocket and pipe results through the callback

    Args:
        url (str): websocket url to connect to
        callback (callable): function to call on websocket data
        json (bool): load websocket data as json
        wrap (bool): wrap result in a list
    '''
    ws = create_connection(url)
    for x in run(ws.recv):
        if isinstance(x, StreamNone):
            continue
        elif not x or isinstance(x, StreamEnd):
            break

        if json:
            x = load_json(x)
        if wrap:
            x = [x]
        callback(x) 

def reconnect(self):
        """重连"""
        try:
            if self.DEBUG:
                print('DataApi:reconnect:{}'.format(self.url))
            if not self.proxyHost:
                self.ws = create_connection(self.url)
            else:
                self.ws = create_connection(self.url, 
                                            http_proxy_host=self.proxyHost, 
                                            http_proxy_port=self.proxyPort)
            return True
        except:
            msg = traceback.format_exc()
            self.onError(u'行情服务器重连失败：%s' %msg)            
            return False
        
    #---------------------------------------------------------------------- 

def _socket_create_connection(self, stream_url):
        ws = None
        try:
            ws = websocket.create_connection(stream_url,
                                             timeout=self.timeout)
        except ValueError:
            with excutils.save_and_reraise_exception():
                LOG.error("websocket create connection invalid URL")
        except Exception:
            # Although a number of exceptions can occur here
            # we handle them all the same way, return None.
            # As such, enough to just "except Exception."
            LOG.exception("websocket create connection failed",
                          exc_info=True)
            return None
        if ws is None or not ws.connected:
            LOG.error("websocket create connection unsuccessful")
            return None

        LOG.debug("websocket connection established")
        return ws 

def __init__(self, config):
        import websocket
        self.token_file = os.path.dirname(os.path.realpath(__file__)) + "/" + config["mac_address"]
        if not config["port"]:
            config["port"] = 8001

        if "timeout" not in config or ("timeout" in config and config["timeout"] == 0):
            config["timeout"] = None

        if config["port"] == 8002:
            url = SSL_URL_FORMAT.format(config["host"], config["port"],
                                        self._serialize_string(config["name"]))
            if os.path.isfile(self.token_file):
                with open(self.token_file, "r") as token_file:
                    url += "&token=" + token_file.readline()
            else:
                config["timeout"] = 10
            self.connection = websocket.create_connection(url, config["timeout"], sslopt={"cert_reqs": ssl.CERT_NONE})

        else:
            url = URL_FORMAT.format(config["host"], config["port"],
                                    self._serialize_string(config["name"]))
            self.connection = websocket.create_connection(url, config["timeout"])
        self._read_response() 

def find_sent_link_with_priority(cls, d_list, top_k=5, predict=False):
        ws_second = websocket.create_connection(cls.second_round_path)
        for i in spcl(range(len(d_list))):
            ws_second.send(json.dumps(d_list[i]))
            item = json.loads(ws_second.recv())
            pids = [it[0] for it in item['prioritized_docids']]
            item['prioritized_docids_aside'] = \
                [it for it in item['prioritized_docids_aside']\
                    if it[0] not in pids]
            if predict:
                porg = \
                    set([k for k, v \
                           in sorted(item['prioritized_docids'],
                                     key=lambda x: (-x[1], x[0]))][:top_k])
                paside = \
                    set([k for k, v \
                            in sorted(item['prioritized_docids_aside'],
                                      key=lambda x: (-x[1], x[0]))][:top_k])
                item['predicted_docids'] = list(porg | paside)
                item['predicted_docids_origin'] = list(porg)
                item['predicted_docids_aside'] = list(paside)
            d_list[i] = item
        ws_second.close() 

def test_client_auth_timeout(self):
        with QdbServer(client_host='localhost',
                       client_port=0,
                       auth_timeout=1,  # Timeout after 1 second.
                       tracer_server=QdbNopServer()) as server:
            ws = create_connection(
                'ws://localhost:%d%s' % (server.client_server.server_port,
                                         DEFAULT_ROUTE_FMT.format(uuid='test'))
            )

            auth_failed_dict = fmt_err_msg('auth', 'No start event received')
            disable_dict = fmt_msg('disable')

            auth_failed_msg = ''
            disable_msg = ''

            with gevent.Timeout(2, False):
                # The server should time us out in 1 second and send back these
                # two messages.
                auth_failed_msg = ws.recv()
                disable_msg = ws.recv()

            self.assertEquals(auth_failed_msg, json.dumps(auth_failed_dict))
            self.assertEquals(disable_msg, json.dumps(disable_dict))
            self.assertFalse('test' in server.session_store) 

def test_bad_auth_tracer(self):
        """
        Tests a non-valid auth message for a tracer.
        """
        with QdbServer(tracer_host='localhost',
                       tracer_port=0,
                       tracer_auth_fn=lambda _: False,
                       client_server=QdbNopServer()) as server:

            auth_failed_dict = fmt_err_msg('auth', 'Authentication failed')

            sck = gevent.socket.create_connection(
                ('localhost', server.tracer_server.server_port)
            )

            send_tracer_event(sck, 'start', {
                'uuid': 'test',
                'auth': 'friendzoned-again',
                'local': (0, 0),
            })
            # We failed auth so the socket should be closed.
            self.assertEqual(auth_failed_dict,
                             recv_tracer_event(sck))
            self.assertFalse('test' in server.session_store) 

def test_tracer_auth_timeout(self):
        """
        Tests the auth timeout for new connections from the client.
        """
        with QdbServer(tracer_host='localhost',
                       tracer_port=0,
                       client_server=QdbNopServer(),
                       auth_timeout=1) as server:

            auth_failed_dict = fmt_err_msg('auth', 'No start event received')
            sck = gevent.socket.create_connection(
                ('localhost', server.tracer_server.server_port)
            )

            self.assertEqual(auth_failed_dict, recv_tracer_event(sck))
            self.assertFalse('test' in server.session_store) 

def test_client_orphan_session(self):
        """
        Tests that a client makes it into the session store without a tracer
        attaching if attach_timeout is set to ALLOW_ORPHANS or 0.
        """
        with QdbServer(tracer_server=QdbNopServer(),
                       client_host='localhost',
                       client_port=0,
                       attach_timeout=ALLOW_ORPHANS) as server:
            client = create_connection(
                'ws://localhost:%d%s' % (server.client_server.server_port,
                                         DEFAULT_ROUTE_FMT.format(uuid='test'))
            )
            send_client_event(client, 'start', '')
            # yield to the session_store to let it get attached.
            gyield()
            self.assertIn('test', server.session_store) 

def test_tracer_orphan_session(self):
        """
        Tests that a tracer makes it into the session_store without a client
        attaching if attach_timeout is set to ALLOW_ORPHANS or 0.
        """
        with QdbServer(client_server=QdbNopServer(),
                       tracer_host='localhost',
                       tracer_port=0,
                       attach_timeout=ALLOW_ORPHANS) as server:
            tracer = gevent.socket.create_connection(
                ('localhost', server.tracer_server.server_port)
            )
            send_tracer_event(tracer, 'start', {
                'uuid': 'test',
                'auth': '',
                'local': (0, 0),
            })
            # yield to the session_store to let it get attached.
            gyield()
            self.assertIn('test', server.session_store) 

def _connect_and_join(self, userid, username, mode, gameid, force_start, public_server):
		logging.debug("Creating connection")
		self._ws = create_connection(ENDPOINT_BOT if not public_server else ENDPOINT_PUBLIC)
		self._lock = threading.RLock()
		_spawn(self._start_sending_heartbeat)
		self._send(["set_username", userid, username, BOT_KEY])

		logging.info("Joining game")
		self._gameid = None
		if mode == "private":
			self._gameid = gameid
			if gameid is None:
				raise ValueError("Gameid must be provided for private games")
			self._send(["join_private", gameid, userid, BOT_KEY])
		elif mode == "1v1":
			self._send(["join_1v1", userid, BOT_KEY])
		elif mode == "team":
			self._send(["join_team", userid, BOT_KEY])
		elif mode == "ffa":
			self._send(["play", userid, BOT_KEY])
		else:
			raise ValueError("Invalid mode")

		if force_start:
			_spawn(self.send_forcestart) 

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 __init__(self, url, connection_payload=None, **ws_options):
        '''
        :param url: ws:// or wss:// url to connect to
        :type url: str

        :param connection_payload: data to pass during initiation of the
          WebSocket connection. It's passed as ``"payload"`` key of the
          ``connection_init`` type message.
        :type connection_payload: dict

        :param ws_options: options to pass to websocket.create_connection
        :type ws_options: dict
        '''
        self.url = url
        self.connection_payload = connection_payload
        self.ws_options = ws_options
        self.keep_alives = ['ka'] 

def _communicate(self, msg):
        import websocket

        if isinstance(msg, str):
            msg = json.loads(msg)

        self._msg_id += 1
        msg['jsonrpc'] = '2.0'
        msg['id'] = self._msg_id
        msg = json.dumps(msg)

        ws = websocket.create_connection(self.url)
        ws.send(msg)
        response = json.loads(ws.recv()).get('result')
        ws.close()
        return response 

def __init__(self, server):
        super().__init__()
        self.setDaemon(True)
        self._stop = threading.Event()
        self._reading = thread_lock()
        self.dt = np.dtype(np.uint16)
        self.dt = self.dt.newbyteorder('<')
        self._data = [np.zeros(WS_FRAME_SIZE, self.dt),
                      np.zeros(WS_FRAME_SIZE, self.dt)]
        self._buf = False
        self.ws = create_connection("ws://{}/".format(server)) 

def parse_text(self,conllu):
        submitted=set()
        to_submit=[]
        submitted_tokdata=[]
        sents=list(prepare_data.read_conllu(conllu.split("\n")))
        ws = create_connection("ws://localhost:{}/translate".format(self.args.marian_port))
        for comments,sent in sents:
            for cols in sent:
                token_data=(cols[FORM],cols[UPOS],cols[XPOS],cols[FEATS])
                if token_data not in self.cache and token_data not in self.local_cache and token_data not in submitted:
                    lem_in=prepare_data.transform_token(cols)[0]
                    to_submit.append(lem_in)
                    submitted_tokdata.append(token_data)
                    submitted.add(token_data)
        
        ws.send("\n".join(to_submit))
        lemmatized=ws.recv()
        #print("LEMMATIZED",lemmatized,file=sys.stderr,flush=True)
        for token_data,lemma in zip(submitted_tokdata,lemmatized.strip().split("\n")):
            self.local_cache[token_data]=prepare_data.detransform_string(lemma)
        result=[]
        for comments,sent in sents:
            for c in comments:
                result.append(c)
            for cols in sent:
                token_data=(cols[FORM],cols[UPOS],cols[XPOS],cols[FEATS])
                if token_data in self.cache:
                    lemma=self.cache[token_data]
                elif token_data in self.local_cache:
                    lemma=self.local_cache[token_data]
                else:
                    assert False
                if not lemma.strip():
                    lemma="_"
                cols[LEMMA]=lemma
                result.append("\t".join(cols))
            result.append("")
        result.append("")
        return "\n".join(result) 

def _connect(self):
        if self.products is None:
            self.products = ["BTC-USD"]
        elif not isinstance(self.products, list):
            self.products = [self.products]

        if self.url[-1] == "/":
            self.url = self.url[:-1]

        if self.channels is None:
            sub_params = {'type': 'subscribe', 'product_ids': self.products}
        else:
            sub_params = {'type': 'subscribe', 'product_ids': self.products, 'channels': self.channels}

        if self.auth:
            timestamp = str(time.time())
            message = timestamp + 'GET' + '/users/self/verify'
            auth_headers = get_auth_headers(timestamp, message, self.api_key, self.api_secret, self.api_passphrase)
            sub_params['signature'] = auth_headers['CB-ACCESS-SIGN']
            sub_params['key'] = auth_headers['CB-ACCESS-KEY']
            sub_params['passphrase'] = auth_headers['CB-ACCESS-PASSPHRASE']
            sub_params['timestamp'] = auth_headers['CB-ACCESS-TIMESTAMP']

        self.ws = create_connection(self.url)

        self.ws.send(json.dumps(sub_params)) 

def connect(self):
        return create_connection(self.get_ws_url()) 

def connect(self):
        url = self.host_url
        try:
            self.ws = websocket.create_connection(url,
                                                  skip_utf8_validation=True)
        except socket.error as e:
            raise exception.ConnectionFailed(e)
        except websocket.WebSocketConnectionClosedException as e:
            raise exception.ConnectionFailed(e)
        except websocket.WebSocketBadStatusException as e:
            raise exception.ConnectionFailed(e) 

def _run_socket():
    global _websocket
    last_start = None
    fallback = 2
    while True:
        try:
            if last_start is not None and last_start > timezone.now() - timedelta(seconds=10):
                time.sleep(fallback)
                fallback = fallback * 2
                if fallback > 120:
                    fallback = 120
            else:
                fallback = 2
            last_start = timezone.now()

            _websocket = websocket.create_connection('wss://socket.lichess.org/api/socket')
            with _games_lock:
                for game_id in list(_games.keys()):
                    _start_watching(game_id)
            while True:
                msg = json.loads(_websocket.recv())
                if msg['t'] == 'fen':
                    with _games_lock:
                        game_id = msg['d']['id']
                        if game_id in _games:
                            _games[game_id] = msg
        except:
            continue 

def _listen(self):
        self.sock.listen(100)
        index = 0
        basic_auth_string = self.create_basic_auth()
        while True:
            self.client, _address = self.sock.accept()
            self.client.settimeout(1800)
            host = 'wss://{}{}'.format(self.remote_addr, '.scm.azurewebsites.net/AppServiceTunnel/Tunnel.ashx')
            basic_auth_header = 'Authorization: Basic {}'.format(basic_auth_string)
            cli_logger = get_logger()  # get CLI logger which has the level set through command lines
            is_verbose = any(handler.level <= logs.INFO for handler in cli_logger.handlers)
            if is_verbose:
                logger.info('Websocket tracing enabled')
                websocket.enableTrace(True)
            else:
                logger.warning('Websocket tracing disabled, use --verbose flag to enable')
                websocket.enableTrace(False)
            self.ws = create_connection(host,
                                        sockopt=((socket.IPPROTO_TCP, socket.TCP_NODELAY, 1),),
                                        class_=TunnelWebSocket,
                                        header=[basic_auth_header],
                                        sslopt={'cert_reqs': ssl.CERT_NONE},
                                        enable_multithread=True)
            logger.info('Websocket, connected status: %s', self.ws.connected)
            index = index + 1
            logger.info('Got debugger connection... index: %s', index)
            debugger_thread = Thread(target=self._listen_to_client, args=(self.client, self.ws, index))
            web_socket_thread = Thread(target=self._listen_to_web_socket, args=(self.client, self.ws, index))
            debugger_thread.start()
            web_socket_thread.start()
            logger.info('Both debugger and websocket threads started...')
            logger.warning('Successfully connected to local server..')
            debugger_thread.join()
            web_socket_thread.join()
            logger.info('Both debugger and websocket threads stopped...')
            logger.warning('Stopped local server..') 

def testIter(self):
        count = 2
        for _ in ws.create_connection('ws://stream.meetup.com/2/rsvps'):
            count -= 1
            if count == 0:
                break 

def testNext(self):
        sock = ws.create_connection('ws://stream.meetup.com/2/rsvps')
        self.assertEqual(str, type(next(sock))) 

def testWebSocket(self):
        s = ws.create_connection("ws://echo.websocket.org/")
        self.assertNotEqual(s, None)
        s.send("Hello, World")
        result = s.recv()
        self.assertEqual(result, "Hello, World")

        s.send(u"こにゃにゃちは、世界")
        result = s.recv()
        self.assertEqual(result, "こにゃにゃちは、世界")
        s.close() 

def testPingPong(self):
        s = ws.create_connection("ws://echo.websocket.org/")
        self.assertNotEqual(s, None)
        s.ping("Hello")
        s.pong("Hi")
        s.close() 

def testSecureWebSocket(self):
        if 1:
            import ssl
            s = ws.create_connection("wss://echo.websocket.org/")
            self.assertNotEqual(s, None)
            self.assertTrue(isinstance(s.sock, ssl.SSLSocket))
            s.send("Hello, World")
            result = s.recv()
            self.assertEqual(result, "Hello, World")
            s.send(u"こにゃにゃちは、世界")
            result = s.recv()
            self.assertEqual(result, "こにゃにゃちは、世界")
            s.close()
        #except:
        #    pass 

def testAfterClose(self):
        s = ws.create_connection("ws://echo.websocket.org/")
        self.assertNotEqual(s, None)
        s.close()
        self.assertRaises(ws.WebSocketConnectionClosedException, s.send, "Hello")
        self.assertRaises(ws.WebSocketConnectionClosedException, s.recv) 

def testSockOpt(self):
        sockopt = ((socket.IPPROTO_TCP, socket.TCP_NODELAY, 1),)
        s = ws.create_connection("ws://echo.websocket.org", sockopt=sockopt)
        self.assertNotEqual(s.sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY), 0)
        s.close()