def read_until_close(self, callback, streaming_callback=None):
        """Reads all data from the socket until it is closed.

        If a ``streaming_callback`` is given, it will be called with chunks
        of data as they become available, and the argument to the final
        ``callback`` will be empty.  Otherwise, the ``callback`` gets the
        data as an argument.

        Subject to ``max_buffer_size`` limit from `IOStream` constructor if
        a ``streaming_callback`` is not used.
        """
        self._set_read_callback(callback)
        self._streaming_callback = stack_context.wrap(streaming_callback)
        if self.closed():
            if self._streaming_callback is not None:
                self._run_callback(self._streaming_callback,
                                   self._consume(self._read_buffer_size))
            self._run_callback(self._read_callback,
                               self._consume(self._read_buffer_size))
            self._streaming_callback = None
            self._read_callback = None
            return
        self._read_until_close = True
        self._streaming_callback = stack_context.wrap(streaming_callback)
        self._try_inline_read()

 def add_callback(self, callback, *args, **kwargs):
     if thread.get_ident() != self._thread_ident:
         # If we're not on the IOLoop's thread, we need to synchronize
         # with other threads, or waking logic will induce a race.
         with self._callback_lock:
             if self._closing:
                 return
             list_empty = not self._callbacks
             self._callbacks.append(functools.partial(
                 stack_context.wrap(callback), *args, **kwargs))
             if list_empty:
                 # If we're not in the IOLoop's thread, and we added the
                 # first callback to an empty list, we may need to wake it
                 # up (it may wake up on its own, but an occasional extra
                 # wake is harmless).  Waking up a polling IOLoop is
                 # relatively expensive, so we try to avoid it when we can.
                 self._waker.wake()
     else:
         if self._closing:
             return
         # If we're on the IOLoop's thread, we don't need the lock,
         # since we don't need to wake anyone, just add the
         # callback. Blindly insert into self._callbacks. This is
         # safe even from signal handlers because the GIL makes
         # list.append atomic. One subtlety is that if the signal
         # is interrupting another thread holding the
         # _callback_lock block in IOLoop.start, we may modify
         # either the old or new version of self._callbacks, but
         # either way will work.
         self._callbacks.append(functools.partial(
             stack_context.wrap(callback), *args, **kwargs))

 def __init__(self, stream, address, server):
     self.stream = stream
     self.address = address
     self.server = server
     self.headers_callback = stack_context.wrap(self.on_headers)
     self.body_callback = stack_context.wrap(self.on_body)
     self.stream.read_until(',', self.headers_callback)

 def __init__(self, stream, address, request_callback, no_keep_alive=False,
              xheaders=False, headers_callback=None, close_callback=None):
     self.stream = stream
     if self.stream.socket.family not in (socket.AF_INET, socket.AF_INET6):
         # Unix (or other) socket; fake the remote address
         address = ('0.0.0.0', 0)
     self.address = address
     self.request_callback = request_callback
     self.no_keep_alive = no_keep_alive
     self.xheaders = xheaders
     self._request = None
     self._request_finished = False
     # Save stack context here, outside of any request.  This keeps
     # contexts from one request from leaking into the next.
     self._header_callback = stack_context.wrap(self._on_headers)
     if headers_callback:
         self.on_headers = stack_context.wrap(headers_callback)
     else:
         self.on_headers = lambda *args: None
     if close_callback:
         self.on_finish = stack_context.wrap(close_callback)
     else:
         self.on_finish = lambda *args: None
     self.stream.read_until(b("\r\n\r\n"), self._header_callback)
     self._write_callback = None

 def __init__(self, callback, barrier_type, on_exception=None):
   callback = stack_context.wrap(callback)
   on_exception = stack_context.wrap(on_exception)
   # Parent frame is derived class __init__, so get grandparent frame.
   frame = sys._getframe().f_back.f_back
   self._barrier = _Barrier(callback, on_exception, barrier_type, frame)
   self._stack_context = stack_context.ExceptionStackContext(self._barrier.ReportException)

    def read_by_delimiter_until_close(self, callback, streaming_callback=None, delimiter=None):
        """Reads all data from the socket until it is closed by delimiters.

        :param streaming_callback: function called on all chunk
        :param delimite: chunks delimiter
        :param callback:
        """
        self._set_read_callback(callback)
        self._read_delimiter = delimiter
        self._read_delimiter_len = len(delimiter)
        self._streaming_callback = stack_context.wrap(streaming_callback)

        if self.closed():

            if self._streaming_callback is not None:

                self._run_callback(self._streaming_callback,
                                   self._consume(self._read_buffer_size))


            self._run_callback(self._read_callback,
                               self._consume(self._read_buffer_size))
            self._streaming_callback = None
            self._read_callback = None
            return

        self._read_until_close = True
        self._streaming_callback = stack_context.wrap(streaming_callback)
        self._try_inline_read()

    def __init__(self,stream,clientid,address,receiveDataCallback,closeCallback,**kwargs):
        '''
        @param clientid,int16
        '''
        self.clientid = clientid

        self.stream = stream
        self.address = address
        self.MASK1 = kwargs.get('mask1',0x59) 
        self.MASK2 = kwargs.get('mask2',0x7a)

        self._closeCallback = closeCallback
        self._receive_callback = stack_context.wrap(receiveDataCallback)

        self._masklen = 2 
        self._read_buffer = []

        stream.set_close_callback(self.loseConnection)

        # Save stack context here, outside of any request.  This keeps
        # contexts from one request from leaking into the next.
        self._read_callback = stack_context.wrap(self._on_read_start_mask)
        self._read_length_callback = stack_context.wrap(self._on_read_length)
        self._read_body_callback = stack_context.wrap(self._on_read_body)

        self.stream.read_bytes(self._masklen,self._read_callback)

def get_lj_post_task_list(task_cfg, task_begin_handle=None, task_end_handle=None):
    task_begin_handle = stack_context.wrap(task_begin_handle)
    task_end_handle = stack_context.wrap(task_end_handle)

    raw_accs_iter = get_items.get_random_infinite_items(task_cfg.accs, is_csv=True)
    if task_cfg.tags is not None:
        tags_iter = get_items.get_random_infinite_items(task_cfg.tags)
    else:
        tags_iter = None
    title_and_content_iter = get_items.get_title_and_content(
        get_items.get_random_infinite_items, task_cfg.titles, task_cfg.content
    )

    def next_acc():
        if "lj:0" == task_cfg.acc_fmt:
            while True:
                acc_row = next(raw_accs_iter)

                if len(acc_row) != 4:
                    raise NotImplementedError("invalid or not implemented account format")

                email, email_password, username, password = acc_row

                return username, password, acc_row

        # if 'lj:...' == task_cfg.acc_fmt:
        #  ...
        #  return

        raise NotImplementedError("not implemented account format")

    for task_i in range(task_cfg.count):
        task = Task()

        task.i = task_i
        task.username, task.password, task._acc_row = next_acc()
        task.blog_id = "lj:{}".format(task.username)
        task.title, task.content = next(title_and_content_iter)
        task.ua_name = task_cfg.ua_name
        task.proxy_kwargs = task_cfg.proxy_kwargs

        if tags_iter is not None:
            tags_list = []
            for tag_i in range(max(round(random.gauss(TAGS_RANDOM_MU, TAGS_RANDOM_SIGMA)), 0)):
                tag = next(tags_iter)
                if tag in tags_list:
                    continue
                tags_list.append(tag)
            task.tags = ", ".join(tags_list)
        else:
            task.tags = None

        task.acc_save = lambda _task=task: lj_acc_save(task_cfg, _task)

        task.task_begin_handle = task_begin_handle
        task.task_end_handle = task_end_handle

        yield task

 def add_callback(self, callback, *args, **kwargs):
     if self.closing:
         raise RuntimeError("IOLoop is closing")
     if kwargs:
         self.asyncio_loop.call_soon_threadsafe(
             functools.partial(self._run_callback, stack_context.wrap(callback), *args, **kwargs)
         )
     else:
         self.asyncio_loop.call_soon_threadsafe(self._run_callback, stack_context.wrap(callback), *args)

    def get(self, callback, timeout=None):
        '''
        Wait for the RPC associated with this :class:`RPCResponseFuture`
        to return a result.  When the result is received, resolve the
        task by calling the passed in ``callback``.

        :param callback: The callback that will be called with the RPC
            response upon completion of the RPC.  It is recommended that
            this not be passed in directly, but rather that
            :meth:`~.get` be called as a function passed to
            :class:`tornado.gen.Task`.

        :param timeout: The amount of time to wait before raising an
            :exc:`RPCTimeoutError` to indicate that the RPC has timed
            out.  This can be a number or a :class:`timedelta`
            object.  If it is a number, it will be treated as
            seconds.
        '''

        self.get_time = datetime.now()

        if self.response_received:
            logger.info('Response has already been received, return '
                        'the value immediately.')
            callback(self.response)
        else:
            callback = stack_context.wrap(callback)
            if self.timeout and not timeout:
                timeout = self.timeout
            elif not self.timeout and not timeout:
                timeout = timedelta(seconds=6)

            key = uuid.uuid4()
            self.wait_callback = yield gen.Callback(key)

            logger.info('Response has not been received yet.  Adding '
                        'timeout to the io_loop in case the response '
                        'times out.')

            if isinstance(timeout, numbers.Real):
                timeout = timedelta(seconds=timeout)

            timeout_callback = stack_context.wrap(self.timeout_callback)
            self.io_loop.add_timeout(timeout, timeout_callback)

            logger.info('Waiting for the response.')
            yield gen.Wait(key)

            if self.timed_out:
                raise RPCTimeoutError('Future waiting for message with cid: '
                                      '"%s" timed out' % str(self.cid))
            elif self.response_received:
                logger.info('Response received successfully.')
                callback(self.response)
            else:
                raise Exception("Neither timed out nor response received")

 def add_job(self, func, cb, exception_cb, prio=10):
     try:
         ThreadPoolExecutor.count += 1
         self.events.put((
             (prio, ThreadPoolExecutor.count),
             (func, stack_context.wrap(cb), stack_context.wrap(exception_cb))
         ))
     except Exception as e:
         jobs_log.exception('cannot put job to queue')
         IOLoop.instance().add_callback(partial(exception_cb, e))

 def __init__(self, connections, object_callback=None,
              finished_callback=None):
     self._object_callback = stack_context.wrap(object_callback)
     self._finished_callback = stack_context.wrap(finished_callback)
     # Connections that have not finished searching
     self._connections = set(connections)
     # Connections without an outstanding blast request
     self._blocking = set(connections)
     self._started = False
     self._paused = False

    def __init__(self, client, request, release_callback,
                 final_callback, max_buffer_size, tcp_client,
                 max_header_size, max_body_size):
        self.io_loop = IOLoop.current()
        self.start_time = self.io_loop.time()
        self.client = client
        self.request = request
        self.release_callback = release_callback
        self.final_callback = final_callback
        self.max_buffer_size = max_buffer_size
        self.tcp_client = tcp_client
        self.max_header_size = max_header_size
        self.max_body_size = max_body_size
        self.code = None
        self.headers = None
        self.chunks = []
        self._decompressor = None
        # Timeout handle returned by IOLoop.add_timeout
        self._timeout = None
        self._sockaddr = None
        with stack_context.ExceptionStackContext(self._handle_exception):
            self.parsed = urlparse.urlsplit(_unicode(self.request.url))
            if self.parsed.scheme not in ("http", "https"):
                raise ValueError("Unsupported url scheme: %s" %
                                 self.request.url)
            # urlsplit results have hostname and port results, but they
            # didn't support ipv6 literals until python 2.7.
            netloc = self.parsed.netloc
            if "@" in netloc:
                userpass, _, netloc = netloc.rpartition("@")
            host, port = httputil.split_host_and_port(netloc)
            if port is None:
                port = 443 if self.parsed.scheme == "https" else 80
            if re.match(r'^\[.*\]$', host):
                # raw ipv6 addresses in urls are enclosed in brackets
                host = host[1:-1]
            self.parsed_hostname = host  # save final host for _on_connect

            if request.allow_ipv6 is False:
                af = socket.AF_INET
            else:
                af = socket.AF_UNSPEC

            ssl_options = self._get_ssl_options(self.parsed.scheme)

            timeout = min(self.request.connect_timeout, self.request.request_timeout)
            if timeout:
                self._timeout = self.io_loop.add_timeout(
                    self.start_time + timeout,
                    stack_context.wrap(functools.partial(self._on_timeout, "while connecting")))
            fut = self.tcp_client.connect(host, port, af=af,
                                          ssl_options=ssl_options,
                                          max_buffer_size=self.max_buffer_size)
            fut.add_done_callback(stack_context.wrap(self._on_connect))

 def write(self, chunk, callback=None):
     # ZMQWEB NOTE: This method is overriden from the base class.
     msg_list = self._build_reply()
     msg_list.extend([b'DATA', chunk])
     logging.debug('Sending write: %r', msg_list)
     self.stream.send_multipart(msg_list)
     # ZMQWEB NOTE: We don't want to permanently register an on_send callback
     # with the stream, so we just call the callback immediately.
     if callback is not None:
         try:
             stack_context.wrap(callback)()
         except:
             logging.error('Unexpected exception in write callback', exc_info=True)

    def authorize_redirect(self, callback_uri=None, extra_params=None,
                           http_client=None, callback=None):
        """Redirects the user to obtain OAuth authorization for this service.

        The ``callback_uri`` may be omitted if you have previously
        registered a callback URI with the third-party service. For
        some services, you must use a previously-registered callback
        URI and cannot specify a callback via this method.

        This method sets a cookie called ``_oauth_request_token`` which is
        subsequently used (and cleared) in `get_authenticated_user` for
        security purposes.

        This method is asynchronous and must be called with ``await``
        or ``yield`` (This is different from other ``auth*_redirect``
        methods defined in this module). It calls
        `.RequestHandler.finish` for you so you should not write any
        other response after it returns.

        .. versionchanged:: 3.1
           Now returns a `.Future` and takes an optional callback, for
           compatibility with `.gen.coroutine`.

        .. deprecated:: 5.1

           The ``callback`` argument is deprecated and will be removed in 6.0.
           Use the returned awaitable object instead.

        """
        if callback_uri and getattr(self, "_OAUTH_NO_CALLBACKS", False):
            raise Exception("This service does not support oauth_callback")
        if http_client is None:
            http_client = self.get_auth_http_client()
        if getattr(self, "_OAUTH_VERSION", "1.0a") == "1.0a":
            fut = http_client.fetch(
                self._oauth_request_token_url(callback_uri=callback_uri,
                                              extra_params=extra_params))
            fut.add_done_callback(wrap(functools.partial(
                self._on_request_token,
                self._OAUTH_AUTHORIZE_URL,
                callback_uri,
                callback)))
        else:
            fut = http_client.fetch(self._oauth_request_token_url())
            fut.add_done_callback(
                wrap(functools.partial(
                    self._on_request_token, self._OAUTH_AUTHORIZE_URL,
                    callback_uri,
                    callback)))

 def read_until_close(self, callback, streaming_callback=None):
     """ 读取直到关闭。
     如果streaming_callback不为空，则它将处理所有的数据，callback得到的参数将为空。 """
     self._set_read_callback(callback)
     self._streaming_callback = stack_context.wrap(streaming_callback)
     if self.closed(): # 如果已经关闭则一次性消费完整个_read_buffer然后返回
         if self._streaming_callback is not None:
             self._run_callback(self._streaming_callback, self._consume(self._read_buffer_size))
         self._run_callback(self._read_callback, self._consume(self._read_buffer_size))
         self._streaming_callback = None
         self._read_callback = None
         return
     self._read_until_close = True
     self._streaming_callback = stack_context.wrap(streaming_callback) # 设置好_streaming_callback后注册io_loop
     self._add_io_state(self.io_loop.READ)

 def read_bytes(self, num_bytes, callback, failure_callback=None):
     """Call callback when we read the given number of bytes."""
     assert not self._blocking
     assert not self._read_callback, "Already reading"
     assert isinstance(num_bytes, int)
     self._read_bytes = num_bytes
     self._read_callback = stack_context.wrap(callback)
     self._read_failure_callback = stack_context.wrap(failure_callback)
     while True:
         if self._read_from_buffer():
             return
         self._check_closed()
         if self._read_to_buffer() == 0:
             break
     self._add_io_state(self.io_loop.READ)

 def __init__(self, stream, address, request_callback, no_keep_alive=False,
              xheaders=False, close_callback=None, headers_callback=None):
     self.stream = stream
     self.address = address
     self.request_callback = request_callback
     self.no_keep_alive = no_keep_alive
     self.xheaders = xheaders
     self._request = None
     self._request_finished = False
     # Save stack context here, outside of any request.  This keeps
     # contexts from one request from leaking into the next.
     self._header_callback = stack_context.wrap(self._on_headers)
     self._headers_callback = stack_context.wrap(headers_callback)
     self._close_callback = stack_context.wrap(close_callback)
     self.stream.read_until("\r\n\r\n", self._header_callback)

 def read_until_regex(self, regex, callback, failure_callback=None):
     """Call callback when we read the given regex pattern."""
     assert not self._read_callback, "Already reading"
     self._read_regex = re.compile(regex)
     self._read_callback = stack_context.wrap(callback)
     if failure_callback:
         self._read_failure_callback = stack_context.wrap(failure_callback)
     while True:
         # See if we've already got the data from a previous read
         if self._read_from_buffer():
             return
         self._check_closed()
         if self._read_to_buffer() == 0:
             break
     self._add_io_state(self.io_loop.READ)

 def on_recv(self, callback, copy=True):
     """Register a callback for when a message is ready to recv.
     
     There can be only one callback registered at a time, so each
     call to `on_recv` replaces previously registered callbacks.
     
     on_recv(None) disables recv event polling.
     
     Use on_recv_stream(callback) instead, to register a callback that will receive
     both this ZMQStream and the message, instead of just the message.
     
     Parameters
     ----------
     
     callback : callable
         callback must take exactly one argument, which will be a
         list, as returned by socket.recv_multipart()
         if callback is None, recv callbacks are disabled.
     copy : bool
         copy is passed directly to recv, so if copy is False,
         callback will receive Message objects. If copy is True,
         then callback will receive bytes/str objects.
     
     Returns : None
     """
     
     self._check_closed()
     assert callback is None or callable(callback)
     self._recv_callback = stack_context.wrap(callback)
     self._recv_copy = copy
     if callback is None:
         self._drop_io_state(self.io_loop.READ)
     else:
         self._add_io_state(self.io_loop.READ)