Python socket.CMSG_LEN属性代码示例

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

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

示例1: ready_to_read

# 需要导入模块: import socket [as 别名]
# 或者: from socket import CMSG_LEN [as 别名]
def ready_to_read(self):
        while True:
            ancillary_size = socket.CMSG_LEN(self.MAX_SIZE)
            try:
                message, ancillary_messages, _msg_flags, from_info = \
                    self.sock.recvmsg(self.MAX_SIZE, ancillary_size)
            except (IOError, OSError) as err:
                if err.args[0] != errno.EWOULDBLOCK:
                    self.warning("Socket receive failed: %s", err)
                return
            if not MACOS:
                rx_interface_index = None
                for anc in ancillary_messages:
                    # pylint:disable=no-member
                    if anc[0] == socket.SOL_IP and anc[1] == socket.IP_PKTINFO:
                        packet_info = in_pktinfo.from_buffer_copy(anc[2])
                        rx_interface_index = packet_info.ipi_ifindex
                    elif anc[0] == socket.SOL_IPV6 and anc[1] == socket.IPV6_PKTINFO:
                        packet_info = in6_pktinfo.from_buffer_copy(anc[2])
                        rx_interface_index = packet_info.ipi6_ifindex
                if rx_interface_index and (rx_interface_index != self._interface_index):
                    # Message received on "wrong" interface; ignore
                    return
            self._receive_function(message, from_info, self.sock) 
开发者ID:brunorijsman,项目名称:rift-python,代码行数:26,代码来源:udp_rx_handler.py


示例2: testCMSG_LEN

# 需要导入模块: import socket [as 别名]
# 或者: from socket import CMSG_LEN [as 别名]
def testCMSG_LEN(self):
        # Test CMSG_LEN() with various valid and invalid values,
        # checking the assumptions used by recvmsg() and sendmsg().
        toobig = self.socklen_t_limit - socket.CMSG_LEN(0) + 1
        values = list(range(257)) + list(range(toobig - 257, toobig))
        # struct cmsghdr has at least three members, two of which are ints
        self.assertGreater(socket.CMSG_LEN(0), array.array("i").itemsize * 2)
        for n in values:
            ret = socket.CMSG_LEN(n)
            # This is how recvmsg() calculates the data size
            self.assertEqual(ret - socket.CMSG_LEN(0), n)
            self.assertLessEqual(ret, self.socklen_t_limit)
        self.assertRaises(OverflowError, socket.CMSG_LEN, -1)
        # sendmsg() shares code with these functions, and requires
        # that it reject values over the limit.
        self.assertRaises(OverflowError, socket.CMSG_LEN, toobig)
        self.assertRaises(OverflowError, socket.CMSG_LEN, sys.maxsize) 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:21,代码来源:test_socket.py


示例3: testCMSG_SPACE

# 需要导入模块: import socket [as 别名]
# 或者: from socket import CMSG_LEN [as 别名]
def testCMSG_SPACE(self):
        # Test CMSG_SPACE() with various valid and invalid values,
        # checking the assumptions used by sendmsg().
        toobig = self.socklen_t_limit - socket.CMSG_SPACE(1) + 1
        values = list(range(257)) + list(range(toobig - 257, toobig))
        last = socket.CMSG_SPACE(0)
        # struct cmsghdr has at least three members, two of which are ints
        self.assertGreater(last, array.array("i").itemsize * 2)
        for n in values:
            ret = socket.CMSG_SPACE(n)
            self.assertGreaterEqual(ret, last)
            self.assertGreaterEqual(ret, socket.CMSG_LEN(n))
            self.assertGreaterEqual(ret, n + socket.CMSG_LEN(0))
            self.assertLessEqual(ret, self.socklen_t_limit)
            last = ret
        self.assertRaises(OverflowError, socket.CMSG_SPACE, -1)
        # sendmsg() shares code with these functions, and requires
        # that it reject values over the limit.
        self.assertRaises(OverflowError, socket.CMSG_SPACE, toobig)
        self.assertRaises(OverflowError, socket.CMSG_SPACE, sys.maxsize) 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:24,代码来源:test_socket.py


示例4: recvfds

# 需要导入模块: import socket [as 别名]
# 或者: from socket import CMSG_LEN [as 别名]
def recvfds(sock, size):
        '''Receive an array of fds over an AF_UNIX socket.'''
        a = array.array('i')
        bytes_size = a.itemsize * size
        msg, ancdata, flags, addr = sock.recvmsg(1, socket.CMSG_LEN(bytes_size))
        if not msg and not ancdata:
            raise EOFError
        try:
            if ACKNOWLEDGE:
                sock.send(b'A')
            if len(ancdata) != 1:
                raise RuntimeError('received %d items of ancdata' %
                                   len(ancdata))
            cmsg_level, cmsg_type, cmsg_data = ancdata[0]
            if (cmsg_level == socket.SOL_SOCKET and
                cmsg_type == socket.SCM_RIGHTS):
                if len(cmsg_data) % a.itemsize != 0:
                    raise ValueError
                a.frombytes(cmsg_data)
                assert len(a) % 256 == msg[0]
                return list(a)
        except (ValueError, IndexError):
            pass
        raise RuntimeError('Invalid data received') 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:26,代码来源:reduction.py


示例5: testFDPassCMSG_LEN

# 需要导入模块: import socket [as 别名]
# 或者: from socket import CMSG_LEN [as 别名]
def testFDPassCMSG_LEN(self):
        # Test using CMSG_LEN() to calculate ancillary buffer size.
        self.checkRecvmsgFDs(1,
                             self.doRecvmsg(self.serv_sock, len(MSG),
                                            socket.CMSG_LEN(4 * SIZEOF_INT)),
                             # RFC 3542 says implementations may set
                             # MSG_CTRUNC if there isn't enough space
                             # for trailing padding.
                             ignoreflags=socket.MSG_CTRUNC) 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:11,代码来源:test_socket.py


示例6: testFDPassSeparateMinSpace

# 需要导入模块: import socket [as 别名]
# 或者: from socket import CMSG_LEN [as 别名]
def testFDPassSeparateMinSpace(self):
        # Pass two FDs in two separate arrays, receiving them into the
        # minimum space for two arrays.
        self.checkRecvmsgFDs(2,
                             self.doRecvmsg(self.serv_sock, len(MSG),
                                            socket.CMSG_SPACE(SIZEOF_INT) +
                                            socket.CMSG_LEN(SIZEOF_INT)),
                             maxcmsgs=2, ignoreflags=socket.MSG_CTRUNC) 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:10,代码来源:test_socket.py


示例7: testCmsgTruncLen0Minus1

# 需要导入模块: import socket [as 别名]
# 或者: from socket import CMSG_LEN [as 别名]
def testCmsgTruncLen0Minus1(self):
        self.checkTruncatedHeader(self.doRecvmsg(self.serv_sock, len(MSG),
                                                 socket.CMSG_LEN(0) - 1)) 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:5,代码来源:test_socket.py


示例8: testCmsgTruncLen0Plus1

# 需要导入模块: import socket [as 别名]
# 或者: from socket import CMSG_LEN [as 别名]
def testCmsgTruncLen0Plus1(self):
        self.checkTruncatedArray(ancbuf=socket.CMSG_LEN(0) + 1, maxdata=1) 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:4,代码来源:test_socket.py


示例9: testCmsgTruncLen1

# 需要导入模块: import socket [as 别名]
# 或者: from socket import CMSG_LEN [as 别名]
def testCmsgTruncLen1(self):
        self.checkTruncatedArray(ancbuf=socket.CMSG_LEN(SIZEOF_INT),
                                 maxdata=SIZEOF_INT) 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:5,代码来源:test_socket.py


示例10: testCmsgTruncLen2Minus1

# 需要导入模块: import socket [as 别名]
# 或者: from socket import CMSG_LEN [as 别名]
def testCmsgTruncLen2Minus1(self):
        self.checkTruncatedArray(ancbuf=socket.CMSG_LEN(2 * SIZEOF_INT) - 1,
                                 maxdata=(2 * SIZEOF_INT) - 1) 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:5,代码来源:test_socket.py


示例11: _testRecvHopLimitCMSG_SPACE

# 需要导入模块: import socket [as 别名]
# 或者: from socket import CMSG_LEN [as 别名]
def _testRecvHopLimitCMSG_SPACE(self):
        self.assertTrue(self.misc_event.wait(timeout=self.fail_timeout))
        self.sendToServer(MSG)
    # Could test receiving into buffer sized using CMSG_LEN, but RFC
    # 3542 says portable applications must provide space for trailing
    # padding.  Implementations may set MSG_CTRUNC if there isn't
    # enough space for the padding. 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:10,代码来源:test_socket.py


示例12: testSingleCmsgTruncLen0Minus1

# 需要导入模块: import socket [as 别名]
# 或者: from socket import CMSG_LEN [as 别名]
def testSingleCmsgTruncLen0Minus1(self):
        self.checkHopLimitTruncatedHeader(ancbufsize=socket.CMSG_LEN(0) - 1) 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:4,代码来源:test_socket.py


示例13: testSecondCmsgTruncLen0Minus1

# 需要导入模块: import socket [as 别名]
# 或者: from socket import CMSG_LEN [as 别名]
def testSecondCmsgTruncLen0Minus1(self):
        self.checkTruncatedSecondHeader(socket.CMSG_SPACE(SIZEOF_INT) +
                                        socket.CMSG_LEN(0) - 1) 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:5,代码来源:test_socket.py


示例14: testSecomdCmsgTruncInData

# 需要导入模块: import socket [as 别名]
# 或者: from socket import CMSG_LEN [as 别名]
def testSecomdCmsgTruncInData(self):
        # Test truncation of the second of two control messages inside
        # its associated data.
        self.serv_sock.setsockopt(socket.IPPROTO_IPV6,
                                  socket.IPV6_RECVHOPLIMIT, 1)
        self.serv_sock.setsockopt(socket.IPPROTO_IPV6,
                                  socket.IPV6_RECVTCLASS, 1)
        self.misc_event.set()
        msg, ancdata, flags, addr = self.doRecvmsg(
            self.serv_sock, len(MSG),
            socket.CMSG_SPACE(SIZEOF_INT) + socket.CMSG_LEN(SIZEOF_INT) - 1)
        self.assertEqual(msg, MSG)
        self.checkRecvmsgAddress(addr, self.cli_addr)
        self.checkFlags(flags, eor=True, checkset=socket.MSG_CTRUNC)
        cmsg_types = {socket.IPV6_TCLASS, socket.IPV6_HOPLIMIT}
        cmsg_level, cmsg_type, cmsg_data = ancdata.pop(0)
        self.assertEqual(cmsg_level, socket.IPPROTO_IPV6)
        cmsg_types.remove(cmsg_type)
        self.assertEqual(len(cmsg_data), SIZEOF_INT)
        a = array.array("i")
        a.frombytes(cmsg_data)
        self.assertGreaterEqual(a[0], 0)
        self.assertLessEqual(a[0], 255)
        if ancdata:
            cmsg_level, cmsg_type, cmsg_data = ancdata.pop(0)
            self.assertEqual(cmsg_level, socket.IPPROTO_IPV6)
            cmsg_types.remove(cmsg_type)
            self.assertLess(len(cmsg_data), SIZEOF_INT)
        self.assertEqual(ancdata, []) 
开发者ID:Microvellum,项目名称:Fluid-Designer,代码行数:36,代码来源:test_socket.py



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