test_udp.py 文件源码

def test_multiListen(self):
        """
        Test that multiple sockets can listen on the same multicast port and
        that they both receive multicast messages directed to that address.
        """
        firstClient = Server()
        firstPort = reactor.listenMulticast(
            0, firstClient, listenMultiple=True)

        portno = firstPort.getHost().port

        secondClient = Server()
        secondPort = reactor.listenMulticast(
            portno, secondClient, listenMultiple=True)

        joined = self.server.transport.joinGroup("225.0.0.250")

        def serverJoined(ignored):
            d1 = firstClient.packetReceived = Deferred()
            d2 = secondClient.packetReceived = Deferred()
            firstClient.transport.write("hello world", ("225.0.0.250", portno))
            return gatherResults([d1, d2])
        joined.addCallback(serverJoined)

        def gotPackets(ignored):
            self.assertEquals(firstClient.packets[0][0], "hello world")
            self.assertEquals(secondClient.packets[0][0], "hello world")
        joined.addCallback(gotPackets)

        def cleanup(passthrough):
            result = gatherResults([
                maybeDeferred(firstPort.stopListening),
                maybeDeferred(secondPort.stopListening)])
            result.addCallback(lambda ign: passthrough)
            return result
        joined.addBoth(cleanup)
        return joined