

def test_produce_consume(self):
    channel = self.c.channel()
    producer = Producer(channel, self.e)
    consumer1 = Consumer(channel, self.q)
    consumer2 = Consumer(channel, self.q2)
    self.q2(channel).declare()
    for i in range(10):
        producer.publish({
            'foo': i,
        }, routing_key='test_transport_memory')
    for i in range(10):
        producer.publish({
            'foo': i,
        }, routing_key='test_transport_memory2')
    _received1 = []
    _received2 = []

    def callback1(message_data, message):
        _received1.append(message)
        message.ack()

    def callback2(message_data, message):
        _received2.append(message)
        message.ack()
    consumer1.register_callback(callback1)
    consumer2.register_callback(callback2)
    consumer1.consume()
    consumer2.consume()
    while 1:
        if ((len(_received1) + len(_received2)) == 20):
            break
        self.c.drain_events()
    self.assertEqual((len(consumer2) + len(_received2)), 20)
    producer.publish({
        'compressed': True,
    }, routing_key='test_transport_memory', compression='zlib')
    m = self.q(channel).get()
    self.assertDictEqual(m.payload, {
        'compressed': True,
    })
    for i in range(10):
        producer.publish({
            'foo': i,
        }, routing_key='test_transport_memory')
    self.assertTrue(self.q(channel).get())
    self.q(channel).delete()
    self.q(channel).declare()
    self.assertIsNone(self.q(channel).get())
    for i in range(10):
        producer.publish({
            'foo': i,
        }, routing_key='test_transport_memory2')
    self.assertTrue(self.q2(channel).get())
    self.q2(channel).purge()
    self.assertIsNone(self.q2(channel).get())

dataset/ETHPy150Open celery/kombu/kombu/tests/transport/test_memory.py test_MemoryTransport.test_produce_consume/VarMisuse@251/320 `_received1`->`consumer2`