Python socket.bind方法代码示例

本文整理汇总了Python中socket.bind方法的典型用法代码示例。如果您正苦于以下问题:Python socket.bind方法的具体用法?Python socket.bind怎么用?Python socket.bind使用的例子?那么恭喜您, 这里精选的方法代码示例或许可以为您提供帮助。您也可以进一步了解该方法所在模块socket的用法示例。

在下文中一共展示了socket.bind方法的21个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: _resolveIPv6

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def _resolveIPv6(ip, port):
    """
    Resolve an IPv6 literal into an IPv6 address.
    This is necessary to resolve any embedded scope identifiers to the relevant
    C{sin6_scope_id} for use with C{socket.connect()}, C{socket.listen()}, or
    C{socket.bind()}; see U{RFC 3493 <https://tools.ietf.org/html/rfc3493>} for
    more information.
    @param ip: An IPv6 address literal.
    @type ip: C{str}
    @param port: A port number.
    @type port: C{int}
    @return: a 4-tuple of C{(host, port, flow, scope)}, suitable for use as an
        IPv6 address.
    @raise socket.gaierror: if either the IP or port is not numeric as it
        should be.
    """
    return socket.getaddrinfo(ip, port, 0, 0, 0, _NUMERIC_ONLY)[0][4] 
开发者ID:proxysh,项目名称:Safejumper-for-Desktop,代码行数:24,代码来源:tcp.py


示例2: _testTCPServerOption

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def _testTCPServerOption(self, level, option, values):
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            self._testSetAndGetOption(sock, level, option, values)
            # now bind and listen on the socket i.e. cause the implementation socket to be created
            sock.bind( (HOST, PORT) )
            sock.listen(50)
            retrieved_option_value = sock.getsockopt(level, option)
            msg = "TCP server option value '(%s,%s)'='%s' did not propagate to implementation socket. Got %s" % (level, option, values[-1], retrieved_option_value)
            if option == socket.SO_RCVBUF:
                # NOTE: see similar bsd/solaris workaround above
                self.assert_(retrieved_option_value >= values[-1], msg)
            else:
                self.failUnlessEqual(retrieved_option_value, values[-1], msg)
            self._testSetAndGetOption(sock, level, option, values)
        finally:
            sock.close() 
开发者ID:Acmesec,项目名称:CTFCrackTools-V2,代码行数:20,代码来源:test_socket.py


示例3: testSO_ERROR

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testSO_ERROR(self):
        good = bad = None
        try:
            good = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            good.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            good.bind((HOST, PORT))
            good.listen(1)
            bad = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            bad.bind((HOST, PORT))
            bad.listen(1)
            self.fail("Listen operation against same port did not generate an expected error")
        except socket.error, se:
            self.failUnlessEqual(bad.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR), se[0])
            # try again, should now be reset
            self.failUnlessEqual(bad.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR), 0) 
开发者ID:Acmesec,项目名称:CTFCrackTools-V2,代码行数:18,代码来源:test_socket.py


示例4: startVideoGeneratorServer

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def startVideoGeneratorServer():
    server_address = (settings.server_location, int(settings.server_port_vid_gen))
    print('Starting video generator server on %s port %s' % server_address)
    socket.bind(server_address)
    socket.listen(5)
    thread = Thread(target=waitConnect)
    thread.start()
    servertick = Thread(target=serverTick)
    servertick.start() 
开发者ID:HA6Bots,项目名称:Automatic-Youtube-Reddit-Text-To-Speech-Video-Generator-and-Uploader,代码行数:11,代码来源:socketservervideogenerator.py


示例5: get_sockaddr

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def get_sockaddr(host, port, family):
    """Return a fully qualified socket address that can be passed to
    :func:`socket.bind`."""
    if family == af_unix:
        return host.split("://", 1)[1]
    try:
        res = socket.getaddrinfo(
            host, port, family, socket.SOCK_STREAM, socket.IPPROTO_TCP
        )
    except socket.gaierror:
        return host, port
    return res[0][4] 
开发者ID:Frank-qlu,项目名称:recruit,代码行数:14,代码来源:serving.py


示例6: __init__

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def __init__(self, host, port, bindAddress, connector, reactor=None):
        # BaseClient.__init__ is invoked later
        self.connector = connector
        self.addr = (host, port)
        whenDone = self.resolveAddress
        err = None
        skt = None
        if abstract.isIPAddress(host):
            self._requiresResolution = False
        elif abstract.isIPv6Address(host):
            self._requiresResolution = False
            self.addr = _resolveIPv6(host, port)
            self.addressFamily = socket.AF_INET6
            self._addressType = address.IPv6Address
        else:
            self._requiresResolution = True
        try:
            skt = self.createInternetSocket()
        except socket.error as se:
            err = error.ConnectBindError(se.args[0], se.args[1])
            whenDone = None
        if whenDone and bindAddress is not None:
            try:
                if abstract.isIPv6Address(bindAddress[0]):
                    bindinfo = _resolveIPv6(*bindAddress)
                else:
                    bindinfo = bindAddress
                skt.bind(bindinfo)
            except socket.error as se:
                err = error.ConnectBindError(se.args[0], se.args[1])
                whenDone = None
        self._finishInit(whenDone, skt, err, reactor) 
开发者ID:proxysh,项目名称:Safejumper-for-Desktop,代码行数:36,代码来源:tcp.py


示例7: startListening

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def startListening(self):
        """Create and bind my socket, and begin listening on it.
        This is called on unserialization, and must be called after creating a
        server to begin listening on the specified port.
        """
        if self._preexistingSocket is None:
            # Create a new socket and make it listen
            try:
                skt = self.createInternetSocket()
                if self.addressFamily == socket.AF_INET6:
                    addr = _resolveIPv6(self.interface, self.port)
                else:
                    addr = (self.interface, self.port)
                skt.bind(addr)
            except socket.error as le:
                raise CannotListenError(self.interface, self.port, le)
            skt.listen(self.backlog)
        else:
            # Re-use the externally specified socket
            skt = self._preexistingSocket
            self._preexistingSocket = None
            # Avoid shutting it down at the end.
            self._shouldShutdown = False
        # Make sure that if we listened on port 0, we update that to
        # reflect what the OS actually assigned us.
        self._realPortNumber = skt.getsockname()[1]
        log.msg("%s starting on %s" % (
                self._getLogPrefix(self.factory), self._realPortNumber))
        # The order of the next 5 lines is kind of bizarre.  If no one
        # can explain it, perhaps we should re-arrange them.
        self.factory.doStart()
        self.connected = True
        self.socket = skt
        self.fileno = self.socket.fileno
        self.numberAccepts = 100
        self.startReading() 
开发者ID:proxysh,项目名称:Safejumper-for-Desktop,代码行数:43,代码来源:tcp.py


示例8: setUp

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def setUp(self):
        self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.serv.bind((self.HOST, self.PORT))
        self.serv.listen(1) 
开发者ID:ofermend,项目名称:medicare-demo,代码行数:7,代码来源:test_socket.py


示例9: testSockName

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testSockName(self):
        # Testing getsockname()
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.bind(("0.0.0.0", PORT+1))
        name = sock.getsockname()
        self.assertEqual(name, ("0.0.0.0", PORT+1)) 
开发者ID:ofermend,项目名称:medicare-demo,代码行数:8,代码来源:test_socket.py


示例10: _testUDPOption

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def _testUDPOption(self, level, option, values):
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            self._testSetAndGetOption(sock, level, option, values)
            # now bind the socket i.e. cause the implementation socket to be created
            sock.bind( (HOST, PORT) )
            self.failUnlessEqual(sock.getsockopt(level, option), values[-1], \
                 "Option value '(%s, %s)'='%s' did not propagate to implementation socket" % (level, option, values[-1]) )
            self._testSetAndGetOption(sock, level, option, values)
        finally:
            sock.close() 
开发者ID:ofermend,项目名称:medicare-demo,代码行数:13,代码来源:test_socket.py


示例11: _testTCPClientOption

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def _testTCPClientOption(self, level, option, values):
        sock = None
        try:
            # First listen on a server socket, so that the connection won't be refused.
            server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            server_sock.bind( (HOST, PORT) )
            server_sock.listen(50)
            # Now do the tests
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self._testSetAndGetOption(sock, level, option, values)
            # now connect the socket i.e. cause the implementation socket to be created
            # First bind, so that the SO_REUSEADDR setting propagates
            sock.bind( (HOST, PORT+1) )
            sock.connect( (HOST, PORT) )
            msg = "Option value '%s'='%s' did not propagate to implementation socket" % (option, values[-1])
            if option in (socket.SO_RCVBUF, socket.SO_SNDBUF):
                # NOTE: there's no guarantee that bufsize will be the
                # exact setsockopt value, particularly after
                # establishing a connection. seems it will be *at least*
                # the values we test (which are rather small) on
                # BSDs.
                self.assert_(sock.getsockopt(level, option) >= values[-1], msg)
            else:
                self.failUnlessEqual(sock.getsockopt(level, option), values[-1], msg)
            self._testSetAndGetOption(sock, level, option, values)
        finally:
            server_sock.close()
            if sock:
                sock.close() 
开发者ID:ofermend,项目名称:medicare-demo,代码行数:31,代码来源:test_socket.py


示例12: testBindSpecific

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testBindSpecific(self):
        self.sock.bind( (self.HOST, self.PORT) ) # Use a specific port
        actual_port = self.sock.getsockname()[1]
        self.failUnless(actual_port == self.PORT,
            "Binding to specific port number should have returned same number: %d != %d" % (actual_port, self.PORT)) 
开发者ID:ofermend,项目名称:medicare-demo,代码行数:7,代码来源:test_socket.py


示例13: testBindEphemeral

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testBindEphemeral(self):
        self.sock.bind( (self.HOST, 0) ) # let system choose a free port
        self.failUnless(self.sock.getsockname()[1] != 0, "Binding to port zero should have allocated an ephemeral port number") 
开发者ID:ofermend,项目名称:medicare-demo,代码行数:5,代码来源:test_socket.py


示例14: testShutdown

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testShutdown(self):
        self.sock.bind( (self.HOST, self.PORT) )
        self.sock.shutdown(socket.SHUT_RDWR) 
开发者ID:ofermend,项目名称:medicare-demo,代码行数:5,代码来源:test_socket.py


示例15: testBroadcast

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testBroadcast(self):
        self.serv.bind( ("", self.PORT) )
        msg = self.serv.recv(len(EIGHT_BIT_MSG))
        self.assertEqual(msg, EIGHT_BIT_MSG) 
开发者ID:ofermend,项目名称:medicare-demo,代码行数:6,代码来源:test_socket.py


示例16: _testNonBlockingConnect

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def _testNonBlockingConnect(self):
        # Testing non-blocking connect
        self.cli.setblocking(0)
        result = self.cli.connect_ex((self.HOST, self.PORT))
        rfds, wfds, xfds = select.select([], [self.cli], [])
        self.failUnless(self.cli in wfds)
        try:
            self.cli.send(MSG)
        except socket.error:
            self.fail("Sending on connected socket should not have raised socket.error")
    #
    # AMAK: 20070518
    # Introduced a new test for connect with bind to specific local address
    # 
开发者ID:ofermend,项目名称:medicare-demo,代码行数:17,代码来源:test_socket.py


示例17: testBindException

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testBindException(self):
        # First bind to the target port
        self.s.bind( (HOST, PORT) )
        self.s.listen(50)
        try:
            # And then try to bind again
            t = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            t.bind( (HOST, PORT) )
            t.listen(50)
        except socket.error, se:
            self.failUnlessEqual(se[0], errno.EADDRINUSE) 
开发者ID:ofermend,项目名称:medicare-demo,代码行数:13,代码来源:test_socket.py


示例18: testBindNonTupleEndpointRaisesTypeError

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testBindNonTupleEndpointRaisesTypeError(self):
        try:
            self.socket.bind(HOST, PORT)
        except TypeError:
            pass
        else:
            self.fail("Illegal non-tuple bind address did not raise TypeError") 
开发者ID:ofermend,项目名称:medicare-demo,代码行数:9,代码来源:test_socket.py


示例19: testAddrTupleIDNAHostname

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testAddrTupleIDNAHostname(self):
        idna_domain = u"al\u00e1n.com"
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        if socket.supports('idna'):
            try:
                s.bind( (idna_domain, 80) )
            except socket.error:
                # We're not worried about socket errors, i.e. bind problems, etc.
                pass
            except Exception, x:
                self.fail("Unexpected exception raised for socket.bind(%s)" % repr(idna_domain)) 
开发者ID:ofermend,项目名称:medicare-demo,代码行数:13,代码来源:test_socket.py


示例20: testGetsocknameBoundNoImpl

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def testGetsocknameBoundNoImpl(self):
        self.s.bind( ("localhost", 0) )
        try:
            self.s.getsockname()
        except socket.error, se:
            self.fail("getsockname() on bound socket should have not raised socket.error") 
开发者ID:ofermend,项目名称:medicare-demo,代码行数:8,代码来源:test_socket.py


示例21: _create_impl_socket

# 需要导入模块: import socket [as 别名]
# 或者: from socket import bind [as 别名]
def _create_impl_socket(self):
        self.s.bind(("localhost", 0))
        self.s.listen(5) 
开发者ID:ofermend,项目名称:medicare-demo,代码行数:5,代码来源:test_socket.py



注:本文中的socket.bind方法示例整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。