You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
defconnect(self, address, callback=None):
"""Connects the socket to a remote address without blocking. May only be called if the socket passed to the constructor was not previously connected. The address parameter is in the same format as for socket.connect, i.e. a (host, port) tuple. If callback is specified, it will be called when the connection is completed. Note that it is safe to call IOStream.write while the connection is pending, in which case the data will be written as soon as the connection is ready. Calling IOStream read methods before the socket is connected works on some platforms but is non-portable. """self._connecting=Truetry:
self.socket.connect(address)
exceptsocket.error, e:
# In non-blocking mode connect() always raises an exceptionife.args[0] notin (errno.EINPROGRESS, errno.EWOULDBLOCK):
raiseself._connect_callback=stack_context.wrap(callback)
self._add_io_state(self.io_loop.WRITE)
defread_until(self, delimiter, callback):
"""Call callback when we read the given delimiter."""assertnotself._read_callback, "Already reading"self._read_delimiter=delimiterself._read_callback=stack_context.wrap(callback)
whileTrue:
# See if we've already got the data from a previous readifself._read_from_buffer():
returnself._check_closed()
ifself._read_to_buffer() ==0:
breakself._add_io_state(self.io_loop.READ)
IOStream.read_bytes
参考 read_until,读限定字节
defread_bytes(self, num_bytes, callback):
"""Call callback when we read the given number of bytes."""assertnotself._read_callback, "Already reading"ifnum_bytes==0:
callback("")
returnself._read_bytes=num_bytesself._read_callback=stack_context.wrap(callback)
whileTrue:
ifself._read_from_buffer():
returnself._check_closed()
ifself._read_to_buffer() ==0:
breakself._add_io_state(self.io_loop.READ)
IOStream.write
defwrite(self, data, callback=None):
"""Write the given data to this stream. If callback is given, we call it when all of the buffered write data has been successfully written to the stream. If there was previously buffered write data and an old write callback, that callback is simply overwritten with this new callback. """self._check_closed()
self._write_buffer.append(data)
self._add_io_state(self.io_loop.WRITE)
self._write_callback=stack_context.wrap(callback)
defset_close_callback(self, callback):
"""Call the given callback when the stream is closed."""self._close_callback=stack_context.wrap(callback)
IOStream.close
从 ioloop 移除 socket 事件
关闭 socket
调用关闭回调
defclose(self):
"""Close this stream."""ifself.socketisnotNone:
self.io_loop.remove_handler(self.socket.fileno())
self.socket.close()
self.socket=Noneifself._close_callback:
self._run_callback(self._close_callback)
defreading(self):
"""Returns true if we are currently reading from the stream."""returnself._read_callbackisnotNonedefwriting(self):
"""Returns true if we are currently writing to the stream."""returnbool(self._write_buffer)
defclosed(self):
returnself.socketisNone
def_run_callback(self, callback, *args, **kwargs):
try:
# Use a NullContext to ensure that all StackContexts are run# inside our blanket exception handler rather than outside.withstack_context.NullContext():
callback(*args, **kwargs)
except:
logging.error("Uncaught exception, closing connection.",
exc_info=True)
# Close the socket on an uncaught exception from a user callback# (It would eventually get closed when the socket object is# gc'd, but we don't want to rely on gc happening before we# run out of file descriptors)self.close()
# Re-raise the exception so that IOLoop.handle_callback_exception# can see it and log the errorraise
IOStream._run_callback
读回调
从 socket 读取数据到缓存
无数据,socket 关闭
检测是否满足 read_until read_bytes
满足则执行对应回调
def_handle_read(self):
whileTrue:
try:
# Read from the socket until we get EWOULDBLOCK or equivalent.# SSL sockets do some internal buffering, and if the data is# sitting in the SSL object's buffer select() and friends# can't see it; the only way to find out if it's there is to# try to read it.result=self._read_to_buffer()
exceptException:
self.close()
returnifresult==0:
breakelse:
ifself._read_from_buffer():
return
IOStream._read_from_socket
从 socket 读取数据
def_read_from_socket(self):
"""Attempts to read from the socket. Returns the data read or None if there is nothing to read. May be overridden in subclasses. """try:
chunk=self.socket.recv(self.read_chunk_size)
exceptsocket.error, e:
ife.args[0] in (errno.EWOULDBLOCK, errno.EAGAIN):
returnNoneelse:
raiseifnotchunk:
self.close()
returnNonereturnchunk
IOStream._read_to_buffer
从 socket 读取数据存入缓存
def_read_to_buffer(self):
"""Reads from the socket and appends the result to the read buffer. Returns the number of bytes read. Returns 0 if there is nothing to read (i.e. the read returns EWOULDBLOCK or equivalent). On error closes the socket and raises an exception. """try:
chunk=self._read_from_socket()
exceptsocket.error, e:
# ssl.SSLError is a subclass of socket.errorlogging.warning("Read error on %d: %s",
self.socket.fileno(), e)
self.close()
raiseifchunkisNone:
return0self._read_buffer.append(chunk)
ifself._read_buffer_size() >=self.max_buffer_size:
logging.error("Reached maximum read buffer size")
self.close()
raiseIOError("Reached maximum read buffer size")
returnlen(chunk)
def_read_from_buffer(self):
"""Attempts to complete the currently-pending read from the buffer. Returns True if the read was completed. """ifself._read_bytes:
ifself._read_buffer_size() >=self._read_bytes:
num_bytes=self._read_bytescallback=self._read_callbackself._read_callback=Noneself._read_bytes=Noneself._run_callback(callback, self._consume(num_bytes))
returnTrueelifself._read_delimiter:
_merge_prefix(self._read_buffer, sys.maxint)
loc=self._read_buffer[0].find(self._read_delimiter)
ifloc!=-1:
callback=self._read_callbackdelimiter_len=len(self._read_delimiter)
self._read_callback=Noneself._read_delimiter=Noneself._run_callback(callback,
self._consume(loc+delimiter_len))
returnTruereturnFalse
def_handle_write(self):
whileself._write_buffer:
try:
ifnotself._write_buffer_frozen:
# On windows, socket.send blows up if given a# write buffer that's too large, instead of just# returning the number of bytes it was able to# process. Therefore we must not call socket.send# with more than 128KB at a time._merge_prefix(self._write_buffer, 128*1024)
num_bytes=self.socket.send(self._write_buffer[0])
self._write_buffer_frozen=False_merge_prefix(self._write_buffer, num_bytes)
self._write_buffer.popleft()
exceptsocket.error, e:
ife.args[0] in (errno.EWOULDBLOCK, errno.EAGAIN):
# With OpenSSL, after send returns EWOULDBLOCK,# the very same string object must be used on the# next call to send. Therefore we suppress# merging the write buffer after an EWOULDBLOCK.# A cleaner solution would be to set# SSL_MODE_ACCEPT_MOVING_WRITE_BUFFER, but this is# not yet accessible from python# (http://bugs.python.org/issue8240)self._write_buffer_frozen=Truebreakelse:
logging.warning("Write error on %d: %s",
self.socket.fileno(), e)
self.close()
returnifnotself._write_bufferandself._write_callback:
callback=self._write_callbackself._write_callback=Noneself._run_callback(callback)
IOStream._consume
从读缓存消费 loc 长度的数据
def_consume(self, loc):
_merge_prefix(self._read_buffer, loc)
returnself._read_buffer.popleft()
def_check_closed(self):
ifnotself.socket:
raiseIOError("Stream is closed")
IOStream._add_io_state
增加 socket 事件状态
def_add_io_state(self, state):
ifself.socketisNone:
# connection has been closed, so there can be no future eventsreturnifnotself._state&state:
self._state=self._state|stateself.io_loop.update_handler(self.socket.fileno(), self._state)
IOStream._read_buffer_size
获取读缓存中已有数据长度
def_read_buffer_size(self):
returnsum(len(chunk) forchunkinself._read_buffer)
classSSLIOStream(IOStream):
"""A utility class to write to and read from a non-blocking socket. If the socket passed to the constructor is already connected, it should be wrapped with ssl.wrap_socket(sock, do_handshake_on_connect=False, **kwargs) before constructing the SSLIOStream. Unconnected sockets will be wrapped when IOStream.connect is finished. """def__init__(self, *args, **kwargs):
"""Creates an SSLIOStream. If a dictionary is provided as keyword argument ssl_options, it will be used as additional keyword arguments to ssl.wrap_socket. """self._ssl_options=kwargs.pop('ssl_options', {})
super(SSLIOStream, self).__init__(*args, **kwargs)
self._ssl_accepting=Trueself._handshake_reading=Falseself._handshake_writing=Falsedefreading(self):
returnself._handshake_readingorsuper(SSLIOStream, self).reading()
defwriting(self):
returnself._handshake_writingorsuper(SSLIOStream, self).writing()
def_do_ssl_handshake(self):
# Based on code from test_ssl.py in the python stdlibtry:
self._handshake_reading=Falseself._handshake_writing=Falseself.socket.do_handshake()
exceptssl.SSLError, err:
iferr.args[0] ==ssl.SSL_ERROR_WANT_READ:
self._handshake_reading=Truereturneliferr.args[0] ==ssl.SSL_ERROR_WANT_WRITE:
self._handshake_writing=Truereturneliferr.args[0] in (ssl.SSL_ERROR_EOF,
ssl.SSL_ERROR_ZERO_RETURN):
returnself.close()
eliferr.args[0] ==ssl.SSL_ERROR_SSL:
logging.warning("SSL Error on %d: %s", self.socket.fileno(), err)
returnself.close()
raiseexceptsocket.error, err:
iferr.args[0] ==errno.ECONNABORTED:
returnself.close()
else:
self._ssl_accepting=Falsesuper(SSLIOStream, self)._handle_connect()
def_handle_read(self):
ifself._ssl_accepting:
self._do_ssl_handshake()
returnsuper(SSLIOStream, self)._handle_read()
def_handle_write(self):
ifself._ssl_accepting:
self._do_ssl_handshake()
returnsuper(SSLIOStream, self)._handle_write()
def_handle_connect(self):
self.socket=ssl.wrap_socket(self.socket,
do_handshake_on_connect=False,
**self._ssl_options)
# Don't call the superclass's _handle_connect (which is responsible# for telling the application that the connection is complete)# until we've completed the SSL handshake (so certificates are# available, etc).def_read_from_socket(self):
try:
# SSLSocket objects have both a read() and recv() method,# while regular sockets only have recv().# The recv() method blocks (at least in python 2.6) if it is# called when there is nothing to read, so we have to use# read() instead.chunk=self.socket.read(self.read_chunk_size)
exceptssl.SSLError, e:
# SSLError is a subclass of socket.error, so this except# block must come first.ife.args[0] ==ssl.SSL_ERROR_WANT_READ:
returnNoneelse:
raiseexceptsocket.error, e:
ife.args[0] in (errno.EWOULDBLOCK, errno.EAGAIN):
returnNoneelse:
raiseifnotchunk:
self.close()
returnNonereturnchunkdef_merge_prefix(deque, size):
"""Replace the first entries in a deque of strings with a single string of up to size bytes. >>> d = collections.deque(['abc', 'de', 'fghi', 'j']) >>> _merge_prefix(d, 5); print d deque(['abcde', 'fghi', 'j']) Strings will be split as necessary to reach the desired size. >>> _merge_prefix(d, 7); print d deque(['abcdefg', 'hi', 'j']) >>> _merge_prefix(d, 3); print d deque(['abc', 'defg', 'hi', 'j']) >>> _merge_prefix(d, 100); print d deque(['abcdefghij']) """prefix= []
remaining=sizewhiledequeandremaining>0:
chunk=deque.popleft()
iflen(chunk) >remaining:
deque.appendleft(chunk[remaining:])
chunk=chunk[:remaining]
prefix.append(chunk)
remaining-=len(chunk)
deque.appendleft(''.join(prefix))
defdoctests():
importdoctestreturndoctest.DocTestSuite()
iostream.py
A utility class to write to and read from a non-blocking socket.
IOStream 对 socket 进行包装,采用注册回调方式实现非阻塞。
通过接口注册各个事件回调
ioloop 中 socket 事件发生后,调用 IOStream._handle_events 方法,对事件进行分发。
对应的事件处理过程中,如果满足注册的回调条件,则调用回调函数
回调函数在 IOStream._handle_events 中被调用
contents
example
一个简单的 IOStream 客户端示例
由此可见, IOStream 是一个异步回调链
head
IOStream.__init__
包装 socket 类
关键语句
self.io_loop.add_handler( self.socket.fileno(), self._handle_events, self._state)
将自身的_handle_events 加入到全局 ioloop poll 事件回调此时只注册了 ERROR 类型事件
_read_buffer: 读缓冲
IOStream.connect
连接 socket 到远程地址,非阻塞模式
IOStream.read_until
如果缓存中数据满足条件,则直接执行 callback 并返回,
否则,保存 callback 函数下次 read 事件发生时,_handle_events 处理读事件时,再进行检测及调用
IOStream.read_bytes
参考 read_until,读限定字节
IOStream.write
IOStream.close
IOStream._handle_events
核心回调
任何类型的 socket 事件触发 ioloop 回调_handle_events,然后在_handle_events 再进行分发
值得注意的是,IOStream 不处理连接请求的 read 事件
注意
作为服务端,默认代理的是已经建立连接的 socket
作为客户端,需要手动调用 IOStream.connect,连接成功后,成功回调在 write 事件中处理
这个实现比较别扭
IOStream._run_callback
执行回调
IOStream._run_callback
读回调
IOStream._read_from_socket
从 socket 读取数据
IOStream._read_to_buffer
从 socket 读取数据存入缓存
IOStream._read_from_buffer
从缓冲中过滤数据
检测是否满足结束条件(read_until/read_bytes),满足则调用之前注册的回调
采用的是查询方式
IOStream._handle_connect
调用连接建立回调,并清除连接中标志
IOStream._handle_write
写事件
IOStream._consume
从读缓存消费 loc 长度的数据
IOStream._add_io_state
增加 socket 事件状态
IOStream._read_buffer_size
获取读缓存中已有数据长度
copyright
author:bigfish
copyright: 许可协议 知识共享署名-非商业性使用 4.0 国际许可协议
The text was updated successfully, but these errors were encountered: