我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用socket.SocketIO()。
def backport_makefile(self, mode="r", buffering=None, encoding=None, errors=None, newline=None): """ Backport of ``socket.makefile`` from Python 3.5. """ if not set(mode) <= set(["r", "w", "b"]): raise ValueError( "invalid mode %r (only r, w, b allowed)" % (mode,) ) writing = "w" in mode reading = "r" in mode or not writing assert reading or writing binary = "b" in mode rawmode = "" if reading: rawmode += "r" if writing: rawmode += "w" raw = SocketIO(self, rawmode) self._makefile_refs += 1 if buffering is None: buffering = -1 if buffering < 0: buffering = io.DEFAULT_BUFFER_SIZE if buffering == 0: if not binary: raise ValueError("unbuffered streams must be binary") return raw if reading and writing: buffer = io.BufferedRWPair(raw, raw, buffering) elif reading: buffer = io.BufferedReader(raw, buffering) else: assert writing buffer = io.BufferedWriter(raw, buffering) if binary: return buffer text = io.TextIOWrapper(buffer, encoding, errors, newline) text.mode = mode return text
def makefile(self, mode='r', bufsize=-1): """Create a file object for the TLS connection (socket emulation). @rtype: L{socket._fileobject} """ self._refCount += 1 # So, it is pretty fragile to be using Python internal objects # like this, but it is probably the best/easiest way to provide # matching behavior for socket emulation purposes. The 'close' # argument is nice, its apparently a recent addition to this # class, so that when fileobject.close() gets called, it will # close() us, causing the refcount to be decremented (decrefAsync). # # If this is the last close() on the outstanding fileobjects / # TLSConnection, then the "actual" close alerts will be sent, # socket closed, etc. if sys.version_info < (3,): return socket._fileobject(self, mode, bufsize, close=True) else: # XXX need to wrap this further if buffering is requested return socket.SocketIO(self, mode)
def _textiowrap_for_test(self, buffering=-1): raw = socket.SocketIO(self, "r") if buffering < 0: buffering = io.DEFAULT_BUFFER_SIZE if buffering == 0: return raw buffer = io.BufferedReader(raw, buffering) text = io.TextIOWrapper(buffer, None, None) text.mode = "rb" return text
def _textiowrap_mock_socket(self, mock, buffering=-1): raw = socket.SocketIO(mock, "r") if buffering < 0: buffering = io.DEFAULT_BUFFER_SIZE if buffering == 0: return raw buffer = io.BufferedReader(raw, buffering) text = io.TextIOWrapper(buffer, None, None) text.mode = "rb" return text
def CP_makefile(sock, mode='r', bufsize=DEFAULT_BUFFER_SIZE): if 'r' in mode: return io.BufferedReader(socket.SocketIO(sock, mode), bufsize) else: return CP_BufferedWriter(socket.SocketIO(sock, mode), bufsize)