def start_tls(self, server_side, ssl_options, server_hostname=None):
        if not isinstance(ssl_options, SSL.Context):
            raise ValueError("ssl_options is not SSL.Context")

        _socket = self.detach()
        _socket = SSL.Connection(ssl_options, _socket)
        if server_side:
            _socket.set_accept_state()
        else:
            _socket.set_connect_state()
            if server_hostname:
                _socket.set_tlsext_host_name(server_hostname.encode("idna"))

        orig_close_callback = self._close_callback
        self._close_callback = None

        future = TracebackFuture()
        ssl_stream = MicroProxySSLIOStream(_socket,
                                           server_hostname=server_hostname,
                                           ssl_options=ssl_options,
                                           io_loop=self.io_loop)

        def close_callback():
            if not future.done():
                future.set_exception(ssl_stream.error or StreamClosedError())
            if orig_close_callback is not None:
                orig_close_callback()

        ssl_stream.set_close_callback(close_callback)
        ssl_stream._ssl_connect_callback = lambda: future.set_result(ssl_stream)
        ssl_stream.max_buffer_size = self.max_buffer_size
        ssl_stream.read_chunk_size = self.read_chunk_size
        return future

 def _defer_to_future(defer):
     future = TracebackFuture()
     defer.addCallbacks(
         future.set_result,
         lambda failure: future.set_exc_info(
             (failure.type, failure.value, failure.tb)))
     return future

class StubWebSocketClientConnection(object):
    """A stub version of tornado.websocket.WebSocketClientConnection for use in
    unit tests.

    Attributes:
        TIMEOUT: class constant, which should be overridden in subclasses, in
            order to vary the time from when the class is instantiated until the
            returned future is resolved.
        connect_future: the Future resolved when the client "connection" is
            made.
        written_messages: a list of messages passed to the write_message
            method to see they have been received.
    """
    TIMEOUT = datetime.timedelta(milliseconds=0)

    def __init__(self, io_loop, request, on_message_callback=None,
                 compression_options=None):
        del request, on_message_callback, compression_options
        self.connect_future = TracebackFuture()
        self.written_messages = []

        io_loop.add_timeout(self.TIMEOUT, self._complete)

    def _complete(self):
        self.connect_future.set_result(self)

    def write_message(self, message):
        """Adds writen message to the written_messages buffer."""
        self.written_messages.append(message)

 def open(self):
     future = TracebackFuture()
     try:
         self._stream.open(lambda :future.set_result(None))
     except Exception:
         future.set_exc_info(sys.exc_info())
     return future

    def fetch(self, request, callback=None, raise_error=True, **kwargs):
        """Executes a request, asynchronously returning an `HTTPResponse`.

        The request may be either a string URL or an `HTTPRequest` object.
        If it is a string, we construct an `HTTPRequest` using any additional
        kwargs: ``HTTPRequest(request, **kwargs)``

        This method returns a `.Future` whose result is an
        `HTTPResponse`. By default, the ``Future`` will raise an
        `HTTPError` if the request returned a non-200 response code
        (other errors may also be raised if the server could not be
        contacted). Instead, if ``raise_error`` is set to False, the
        response will always be returned regardless of the response
        code.

        If a ``callback`` is given, it will be invoked with the `HTTPResponse`.
        In the callback interface, `HTTPError` is not automatically raised.
        Instead, you must check the response's ``error`` attribute or
        call its `~HTTPResponse.rethrow` method.
        """
        if self._closed:
            raise RuntimeError("fetch() called on closed AsyncHTTPClient")
        if not isinstance(request, HTTPRequest):
            request = HTTPRequest(url=request, **kwargs)
        else:
            if kwargs:
                raise ValueError("kwargs can't be used if request is an HTTPRequest object")
        # We may modify this (to add Host, Accept-Encoding, etc),
        # so make sure we don't modify the caller's object.  This is also
        # where normal dicts get converted to HTTPHeaders objects.
        request.headers = httputil.HTTPHeaders(request.headers)
        request = _RequestProxy(request, self.defaults)
        future = TracebackFuture()
        if callback is not None:
            callback = stack_context.wrap(callback)

            def handle_future(future):
                exc = future.exception()
                if isinstance(exc, HTTPError) and exc.response is not None:
                    response = exc.response
                elif exc is not None:
                    response = HTTPResponse(
                        request, 599, error=exc,
                        request_time=time.time() - request.start_time)
                else:
                    response = future.result()
                self.io_loop.add_callback(callback, response)
            future.add_done_callback(handle_future)

        def handle_response(response):
            if raise_error and response.error:
                future.set_exception(response.error)
            else:
                future.set_result(response)
        self.fetch_impl(request, handle_response)
        return future

 def close(self):
     if self._cursor is None:
         future = TracebackFuture()
         future.set_result(None)
         return future
     future = async_call_method(self._cursor.close)
     def do_close(future):
         self._cursor = None
     future.add_done_callback(do_close)
     return future

 def register(self, name, callback=None):
     msg_id = next(self._generator)
     request = packer.dumps(('register', name))
     buff = struct.pack('!ibi', len(request), RPC_REGISTER, msg_id) + request
     future = TracebackFuture()
     if callback:
         future.add_done_callback(callback)
     self.add_request_table(msg_id, future)
     self.write(buff)
     return future

    def fetch(self, url, headers=None, body=None, method="GET", callback=None, raise_error=True, cache=None, **kwargs):
        headers = headers or {}
        body = body or "{}"
        """very simlar with AsyncHTTPClient.fetch
        """
        if self._closed:
            raise RuntimeError("fetch() called on closed AsyncHTTPClient")
        future = TracebackFuture()
        if isinstance(body, dict):
            for k,v in body.items():
                if v is None:
                    del body[k]
            body = urllib.urlencode(body)
        for k,v in headers.items(): #headers 只能接收str
            if v:
                headers[k] = str(headers[k])
            else:
                del headers[k]
        request = HTTPRequest(url=url,method=method,headers=headers,body=body, allow_nonstandard_methods=True, request_timeout=600 ,**kwargs)
        # We may modify this (to add Host, Accept-Encoding, etc),
        # so make sure we don't modify the caller's object.  This is also
        # where normal dicts get converted to HTTPHeaders objects.
        request.headers = httputil.HTTPHeaders(request.headers)
        request = _RequestProxy(request, self.defaults)
        if callback is not None:
            callback = stack_context.wrap(callback)

            def handle_future(future):
                exc = future.exception()
                if isinstance(exc, HTTPError) and exc.response is not None:
                    response = exc.response
                elif exc is not None:
                    response = HTTPResponse(
                        request, 599, error=exc,
                        request_time=time.time() - request.start_time)
                else:
                    response = future.result()
                self.io_loop.add_callback(callback, response)
            future.add_done_callback(handle_future)

        def handle_response(response):
            if raise_error and response.error:
                future.set_exception(response.error)
            else:
                try:
                    resp = json.loads(str(response.body))
                    if resp.get("statusCode") and resp.get("statusCode")==800:
                        future.set_result(resp)
                        log.info(json.dumps({"response":resp,"body":body,"headers":headers,"url":url}))
                    else:
                        future.set_result({"error_type":"statusCode is not 800", "response":resp,"body":body,"headers":headers,"url":url})
                        log.error(json.dumps({"error_type":"statusCode is not 800", "response":resp,"body":body,"headers":headers,"url":url}))
                except Exception,e:
                    future.set_result({"error_type":"json.loads failed!","error":str(e),"response.body":response.body,"body":body,"headers":headers,"url":url})
                    log.error(json.dumps({"error_type":"json.loads failed!","error":str(e),"response.body":response.body,"body":body,"headers":headers,"url":url}))

    def handle_yield(self, yielded):
        # Lists containing YieldPoints require stack contexts;
        # other lists are handled via multi_future in convert_yielded.
        if (isinstance(yielded, list) and
                any(isinstance(f, YieldPoint) for f in yielded)):
            yielded = Multi(yielded)
        elif (isinstance(yielded, dict) and
              any(isinstance(f, YieldPoint) for f in yielded.values())):
            yielded = Multi(yielded)

        if isinstance(yielded, YieldPoint):
            # YieldPoints are too closely coupled to the Runner to go
            # through the generic convert_yielded mechanism.
            self.future = TracebackFuture()

            def start_yield_point():
                try:
                    yielded.start(self)
                    if yielded.is_ready():
                        self.future.set_result(
                            yielded.get_result())
                    else:
                        self.yield_point = yielded
                except Exception:
                    self.future = TracebackFuture()
                    self.future.set_exc_info(sys.exc_info())

            if self.stack_context_deactivate is None:
                # Start a stack context if this is the first
                # YieldPoint we've seen.
                with stack_context.ExceptionStackContext(
                        self.handle_exception) as deactivate:
                    self.stack_context_deactivate = deactivate

                    def cb():
                        start_yield_point()
                        self.run()
                    self.io_loop.add_callback(cb)
                    return False
            else:
                start_yield_point()
        else:
            try:
                self.future = convert_yielded(yielded)
            except BadYieldError:
                self.future = TracebackFuture()
                self.future.set_exc_info(sys.exc_info())

        if not self.future.done() or self.future is moment:
            self.io_loop.add_future(
                self.future, lambda f: self.run())
            return False
        return True

    def handle_yield(self, yielded):
        # Lists containing YieldPoints require stack contexts;
        # other lists are handled in convert_yielded.
        if _contains_yieldpoint(yielded):
            yielded = multi(yielded)

        if isinstance(yielded, YieldPoint):
            # YieldPoints are too closely coupled to the Runner to go
            # through the generic convert_yielded mechanism.
            self.future = TracebackFuture()

            def start_yield_point():
                try:
                    yielded.start(self)
                    if yielded.is_ready():
                        self.future.set_result(
                            yielded.get_result())
                    else:
                        self.yield_point = yielded
                except Exception:
                    self.future = TracebackFuture()
                    self.future.set_exc_info(sys.exc_info())

            if self.stack_context_deactivate is None:
                # Start a stack context if this is the first
                # YieldPoint we've seen.
                with stack_context.ExceptionStackContext(
                        self.handle_exception) as deactivate:
                    self.stack_context_deactivate = deactivate

                    def cb():
                        start_yield_point()
                        self.run()
                    self.io_loop.add_callback(cb)
                    return False
            else:
                start_yield_point()
        else:
            try:
                self.future = convert_yielded(yielded)
            except BadYieldError:
                self.future = TracebackFuture()
                self.future.set_exc_info(sys.exc_info())

        if not self.future.done() or self.future is moment:
            def inner(f):
                # Break a reference cycle to speed GC.
                f = None # noqa
                self.run()
            self.io_loop.add_future(
                self.future, inner)
            return False
        return True

    def call(self, method_name, *arg, **kwargs):
        _callback = kwargs.get('callback')
        msg_id = next(self._generator)
        request = packer.dumps((method_name, arg))
        buff = struct.pack('!ibi', len(request), RPC_REQUEST, msg_id) + request

        future = TracebackFuture()
        self.add_request_table(msg_id, future)

        if _callback:
            future.add_done_callback(_callback)
        self.write(buff)
        return future

    def fetch(self, request, callback=None, **kwargs):
        """Executes a request, asynchronously returning an `HTTPResponse`.

        The request may be either a string URL or an `HTTPRequest` object.
        If it is a string, we construct an `HTTPRequest` using any additional
        kwargs: ``HTTPRequest(request, **kwargs)``

        This method returns a `.Future` whose result is an
        `HTTPResponse`.  The ``Future`` wil raise an `HTTPError` if
        the request returned a non-200 response code.

        If a ``callback`` is given, it will be invoked with the `HTTPResponse`.
        In the callback interface, `HTTPError` is not automatically raised.
        Instead, you must check the response's ``error`` attribute or
        call its `~HTTPResponse.rethrow` method.
        """
        if not isinstance(request, HTTPRequest):
            request = HTTPRequest(url=request, **kwargs)
        # We may modify this (to add Host, Accept-Encoding, etc),
        # so make sure we don't modify the caller's object.  This is also
        # where normal dicts get converted to HTTPHeaders objects.
        request.headers = httputil.HTTPHeaders(request.headers)
        if options.client_trace:
            request.trace.record(Annotation.client_send())
        request = _RequestProxy(request, self.defaults)
        future = TracebackFuture()
        if callback is not None:
            callback = stack_context.wrap(callback)

            def handle_future(future):
                exc = future.exception()
                if isinstance(exc, HTTPError) and exc.response is not None:
                    response = exc.response
                elif exc is not None:
                    response = HTTPResponse(
                        request, 599, error=exc,
                        request_time=time.time() - request.start_time)
                else:
                    response = future.result()
                self.io_loop.add_callback(callback, response)
            future.add_done_callback(handle_future)

        def handle_response(response):
            if response.error:
                future.set_exception(response.error)
            else:
                future.set_result(response)
            if options.client_trace:
                request.trace.record(Annotation.client_recv())
        self.fetch_impl(request, handle_response)
        return future

    def mock_fetch(self, request, callback=None, **kwargs):
        future = TracebackFuture()
        if callback is not None:
            callback = stack_context.wrap(callback)

            def handle_future(future):
                response = future.result()
                self.io_loop.add_callback(callback, response)
            future.add_done_callback(handle_future)

        res = MagicMock()
        future.set_result(res)

        return future

 def fetch(self, request, callback=None, **kwargs):
     if not isinstance(request, HTTPRequest):
         request = HTTPRequest(url=request, **kwargs)
     response_partial = RequestCollection.find(request)
     if response_partial:
         resp = response_partial(request)
     else:
         resp = HTTPResponse(request, 404)
     if callback is not None:
         callback(resp)
     else:
         future = TracebackFuture()
         future.set_result(resp)
         return future

    def handle_yield(self, yielded):
        if isinstance(yielded, list):
            if all(is_future(f) for f in yielded):
                yielded = multi_future(yielded)
            else:
                yielded = Multi(yielded)
        elif isinstance(yielded, dict):
            if all(is_future(f) for f in yielded.values()):
                yielded = multi_future(yielded)
            else:
                yielded = Multi(yielded)

        if isinstance(yielded, YieldPoint):
            self.future = TracebackFuture()
            def start_yield_point():
                try:
                    yielded.start(self)
                    if yielded.is_ready():
                        self.future.set_result(
                            yielded.get_result())
                    else:
                        self.yield_point = yielded
                except Exception:
                    self.future = TracebackFuture()
                    self.future.set_exc_info(sys.exc_info())
            if self.stack_context_deactivate is None:
                # Start a stack context if this is the first
                # YieldPoint we've seen.
                with stack_context.ExceptionStackContext(
                        self.handle_exception) as deactivate:
                    self.stack_context_deactivate = deactivate
                    def cb():
                        start_yield_point()
                        self.run()
                    self.io_loop.add_callback(cb)
                    return False
            else:
                start_yield_point()
        elif is_future(yielded):
            self.future = yielded
            if not self.future.done() or self.future is moment:
                self.io_loop.add_future(
                    self.future, lambda f: self.run())
                return False
        else:
            self.future = TracebackFuture()
            self.future.set_exception(BadYieldError(
                "yielded unknown object %r" % (yielded,)))
        return True

 def wrapper(*args, **kwargs):
     future = TracebackFuture()
     callback, args, kwargs = replacer.replace(future, args, kwargs)
     if callback is not None:
         future.add_done_callback(
             functools.partial(_auth_future_to_callback, callback))
     def handle_exception(typ, value, tb):
         if future.done():
             return False
         else:
             future.set_exc_info((typ, value, tb))
             return True
     with ExceptionStackContext(handle_exception):
         f(*args, **kwargs)
     return future

    def __init__(self, io_loop, request, on_message_callback=None,
                 compression_options=None):
        del request, on_message_callback, compression_options
        self.connect_future = TracebackFuture()
        self.written_messages = []

        io_loop.add_timeout(self.TIMEOUT, self._complete)

    def __init__(self, io_loop, request, compression_options=None):
        self.compression_options = compression_options
        self.connect_future = TracebackFuture()
        self.read_future = None
        self.read_queue = collections.deque()
        self.key = base64.b64encode(os.urandom(16))

        scheme, sep, rest = request.url.partition(":")
        scheme = {"ws": "http", "wss": "https"}[scheme]
        request.url = scheme + sep + rest
        request.headers.update(
            {
                "Upgrade": "websocket",
                "Connection": "Upgrade",
                "Sec-WebSocket-Key": self.key,
                "Sec-WebSocket-Version": "13",
            }
        )
        if self.compression_options is not None:
            # Always offer to let the server set our max_wbits (and even though
            # we don't offer it, we will accept a client_no_context_takeover
            # from the server).
            # TODO: set server parameters for deflate extension
            # if requested in self.compression_options.
            request.headers["Sec-WebSocket-Extensions"] = "permessage-deflate; client_max_window_bits"

        self.tcp_client = TCPClient(io_loop=io_loop)
        super(WebSocketClientConnection, self).__init__(
            io_loop, None, request, lambda: None, self._on_http_response, 104857600, self.tcp_client, 65536
        )

 def __init__(self, future=None):
     if future is not None:
         self._future = future
     else:
         self._future = TracebackFuture()
         self._future.add_done_callback(functools.partial(async_result_complete, self))
     self._condition = threading.Condition()

    def __init__(self, io_loop, request, on_message_callback=None,
                 compression_options=None):
        self.compression_options = compression_options
        self.connect_future = TracebackFuture()
        self.protocol = None
        self.read_future = None
        self.read_queue = collections.deque()
        self.key = base64.b64encode(os.urandom(16))
        self._on_message_callback = on_message_callback
        self.close_code = self.close_reason = None

        scheme, sep, rest = request.url.partition(':')
        scheme = {'ws': 'http', 'wss': 'https'}[scheme]
        request.url = scheme + sep + rest
        request.headers.update({
            'Upgrade': 'websocket',
            'Connection': 'Upgrade',
            'Sec-WebSocket-Key': self.key,
            'Sec-WebSocket-Version': '13',
        })
        if self.compression_options is not None:
            # Always offer to let the server set our max_wbits (and even though
            # we don't offer it, we will accept a client_no_context_takeover
            # from the server).
            # TODO: set server parameters for deflate extension
            # if requested in self.compression_options.
            request.headers['Sec-WebSocket-Extensions'] = (
                'permessage-deflate; client_max_window_bits')

        self.tcp_client = TCPClient(io_loop=io_loop)
        super(WebSocketClientConnection, self).__init__(
            io_loop, None, request, lambda: None, self._on_http_response,
            104857600, self.tcp_client, 65536, 104857600)