Python gevent.socket 模块，socketpair() 实例源码

我们从Python开源项目中，提取了以下6个代码示例，用于说明如何使用gevent.socket.socketpair()。

项目：Intranet-Penetration 作者：yuxiaokui | 项目源码 | 文件源码

def patch_socket(dns=True, aggressive=True):
    """Replace the standard socket object with gevent's cooperative sockets.

    If *dns* is true, also patch dns functions in :mod:`socket`.
    """
    from gevent import socket
    _socket = __import__('socket')
    _socket.socket = socket.socket
    _socket.SocketType = socket.SocketType
    _socket.create_connection = socket.create_connection
    if hasattr(socket, 'socketpair'):
        _socket.socketpair = socket.socketpair
    if hasattr(socket, 'fromfd'):
        _socket.fromfd = socket.fromfd
    try:
        from gevent.socket import ssl, sslerror
        _socket.ssl = ssl
        _socket.sslerror = sslerror
    except ImportError:
        if aggressive:
            try:
                del _socket.ssl
            except AttributeError:
                pass
    if dns:
        patch_dns()

项目：MKFQ 作者：maojingios | 项目源码 | 文件源码

def patch_socket(dns=True, aggressive=True):
    """Replace the standard socket object with gevent's cooperative sockets.

    If *dns* is true, also patch dns functions in :mod:`socket`.
    """
    from gevent import socket
    _socket = __import__('socket')
    _socket.socket = socket.socket
    _socket.SocketType = socket.SocketType
    _socket.create_connection = socket.create_connection
    if hasattr(socket, 'socketpair'):
        _socket.socketpair = socket.socketpair
    if hasattr(socket, 'fromfd'):
        _socket.fromfd = socket.fromfd
    try:
        from gevent.socket import ssl, sslerror
        _socket.ssl = ssl
        _socket.sslerror = sslerror
    except ImportError:
        if aggressive:
            try:
                del _socket.ssl
            except AttributeError:
                pass
    if dns:
        patch_dns()

项目：py-bson-rpc 作者：seprich | 项目源码 | 文件源码

def _socketpair(tmodel):
    if tmodel == ThreadingModel.THREADS:
        return tsocket.socketpair()
    elif tmodel == ThreadingModel.GEVENT:
        return gsocket.socketpair()

项目：py-bson-rpc 作者：seprich | 项目源码 | 文件源码

def _socketpair(tmodel):
    if tmodel == ThreadingModel.THREADS:
        return tsocket.socketpair()
    elif tmodel == ThreadingModel.GEVENT:
        return gsocket.socketpair()

项目：xxNet 作者：drzorm | 项目源码 | 文件源码

def patch_socket(dns=True, aggressive=True):
    """Replace the standard socket object with gevent's cooperative sockets.

    If *dns* is true, also patch dns functions in :mod:`socket`.
    """
    from gevent import socket
    _socket = __import__('socket')
    _socket.socket = socket.socket
    _socket.SocketType = socket.SocketType
    _socket.create_connection = socket.create_connection
    if hasattr(socket, 'socketpair'):
        _socket.socketpair = socket.socketpair
    if hasattr(socket, 'fromfd'):
        _socket.fromfd = socket.fromfd
    try:
        from gevent.socket import ssl, sslerror
        _socket.ssl = ssl
        _socket.sslerror = sslerror
    except ImportError:
        if aggressive:
            try:
                del _socket.ssl
            except AttributeError:
                pass
    if dns:
        patch_dns()

项目：Docker-XX-Net 作者：kuanghy | 项目源码 | 文件源码

def patch_socket(dns=True, aggressive=True):
    """Replace the standard socket object with gevent's cooperative sockets.

    If *dns* is true, also patch dns functions in :mod:`socket`.
    """
    from gevent import socket
    _socket = __import__('socket')
    _socket.socket = socket.socket
    _socket.SocketType = socket.SocketType
    _socket.create_connection = socket.create_connection
    if hasattr(socket, 'socketpair'):
        _socket.socketpair = socket.socketpair
    if hasattr(socket, 'fromfd'):
        _socket.fromfd = socket.fromfd
    try:
        from gevent.socket import ssl, sslerror
        _socket.ssl = ssl
        _socket.sslerror = sslerror
    except ImportError:
        if aggressive:
            try:
                del _socket.ssl
            except AttributeError:
                pass
    if dns:
        patch_dns()