#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#
from qpid.client import Client, Closed
from qpid.queue import Empty
from qpid.testlib import TestBase010
from qpid.datatypes import Message, RangedSet
from qpid.session import SessionException

from qpid.content import Content
from time import sleep

class MessageTests(TestBase010):
    """Tests for 'methods' on the amqp message 'class'"""

    def test_no_local(self):
        """
        NOTE: this is a test of a QPID specific feature
        
        Test that the qpid specific no_local arg is honoured.
        """
        session = self.session
        #setup, declare two queues one of which excludes delivery of locally sent messages
        session.queue_declare(queue="test-queue-1a", exclusive=True, auto_delete=True)
        session.queue_declare(queue="test-queue-1b", exclusive=True, auto_delete=True, arguments={'no-local':'true'})
        #establish two consumers 
        self.subscribe(destination="local_included", queue="test-queue-1a")
        self.subscribe(destination="local_excluded", queue="test-queue-1b")

        #send a message
        session.message_transfer(message=Message(session.delivery_properties(routing_key="test-queue-1a"), "deliver-me"))
        session.message_transfer(message=Message(session.delivery_properties(routing_key="test-queue-1b"), "dont-deliver-me"))

        #send a message from another session on the same connection to each queue
        session2 = self.conn.session("my-local-session")
        session2.message_transfer(message=Message(session2.delivery_properties(routing_key="test-queue-1a"), "deliver-me-as-well"))
        session2.message_transfer(message=Message(session2.delivery_properties(routing_key="test-queue-1b"), "dont-deliver-me-either"))

        #send a message from a session on another connection to each queue
        for q in ["test-queue-1a", "test-queue-1b"]:
            session.exchange_bind(queue=q, exchange="amq.fanout", binding_key="my-key")
        other = self.connect()
        session3 = other.session("my-other-session")
        session3.message_transfer(destination="amq.fanout", message=Message("i-am-not-local"))
        other.close()

        #check the queues of the two consumers
        excluded = session.incoming("local_excluded")
        included = session.incoming("local_included")
        for b in ["deliver-me", "deliver-me-as-well", "i-am-not-local"]:
            msg = included.get(timeout=1)
            self.assertEqual(b, msg.body)
        msg = excluded.get(timeout=1)
        self.assertEqual("i-am-not-local", msg.body)
        try:
            excluded.get(timeout=1)
            self.fail("Received locally published message though no_local=true")
        except Empty: None

    def test_no_local_awkward(self):

        """
        NOTE: this is a test of a QPID specific feature
        
        Check that messages which will be excluded through no-local
        processing will not block subsequent deliveries
        """

        session = self.session
        #setup:
        session.queue_declare(queue="test-queue", exclusive=True, auto_delete=True, arguments={'no-local':'true'})
        #establish consumer which excludes delivery of locally sent messages
        self.subscribe(destination="local_excluded", queue="test-queue")

        #send a 'local' message
        session.message_transfer(message=Message(session.delivery_properties(routing_key="test-queue"), "local"))

        #send a non local message
        other = self.connect()
        session2 = other.session("my-session", 1)
        session2.message_transfer(message=Message(session2.delivery_properties(routing_key="test-queue"), "foreign"))
        session2.close()
        other.close()

        #check that the second message only is delivered
        excluded = session.incoming("local_excluded")
        msg = excluded.get(timeout=1)
        self.assertEqual("foreign", msg.body)
        try:
            excluded.get(timeout=1)
            self.fail("Received extra message")
        except Empty: None
        #check queue is empty
        self.assertEqual(0, session.queue_query(queue="test-queue").message_count)

    def test_no_local_exclusive_subscribe(self):
        """
        NOTE: this is a test of a QPID specific feature

        Test that the no_local processing works on queues not declared
        as exclusive, but with an exclusive subscription
        """
        session = self.session

        #setup, declare two queues one of which excludes delivery of
        #locally sent messages but is not declared as exclusive
        session.queue_declare(queue="test-queue-1a", exclusive=True, auto_delete=True)
        session.queue_declare(queue="test-queue-1b", auto_delete=True, arguments={'no-local':'true'})
        #establish two consumers 
        self.subscribe(destination="local_included", queue="test-queue-1a")
        self.subscribe(destination="local_excluded", queue="test-queue-1b", exclusive=True)

        #send a message from the same session to each queue
        session.message_transfer(message=Message(session.delivery_properties(routing_key="test-queue-1a"), "deliver-me"))
        session.message_transfer(message=Message(session.delivery_properties(routing_key="test-queue-1b"), "dont-deliver-me"))

        #send a message from another session on the same connection to each queue
        session2 = self.conn.session("my-session")
        session2.message_transfer(message=Message(session2.delivery_properties(routing_key="test-queue-1a"), "deliver-me-as-well"))
        session2.message_transfer(message=Message(session2.delivery_properties(routing_key="test-queue-1b"), "dont-deliver-me-either"))

        #send a message from a session on another connection to each queue
        for q in ["test-queue-1a", "test-queue-1b"]:
            session.exchange_bind(queue=q, exchange="amq.fanout", binding_key="my-key")
        other = self.connect()
        session3 = other.session("my-other-session")
        session3.message_transfer(destination="amq.fanout", message=Message("i-am-not-local"))
        other.close()

        #check the queues of the two consumers
        excluded = session.incoming("local_excluded")
        included = session.incoming("local_included")
        for b in ["deliver-me", "deliver-me-as-well", "i-am-not-local"]:
            msg = included.get(timeout=1)
            self.assertEqual(b, msg.body)
        msg = excluded.get(timeout=1)
        self.assertEqual("i-am-not-local", msg.body)
        try:
            excluded.get(timeout=1)
            self.fail("Received locally published message though no_local=true")
        except Empty: None


    def test_consume_exclusive(self):
        """
        Test an exclusive consumer prevents other consumer being created
        """
        session = self.session
        session.queue_declare(queue="test-queue-2", exclusive=True, auto_delete=True)
        session.message_subscribe(destination="first", queue="test-queue-2", exclusive=True)
        try:
            session.message_subscribe(destination="second", queue="test-queue-2")
            self.fail("Expected consume request to fail due to previous exclusive consumer")
        except SessionException, e:
            self.assertEquals(405, e.args[0].error_code)

    def test_consume_exclusive2(self):
        """
        Check that an exclusive consumer cannot be created if a consumer already exists:
        """
        session = self.session
        session.queue_declare(queue="test-queue-2", exclusive=True, auto_delete=True)
        session.message_subscribe(destination="first", queue="test-queue-2")
        try:
            session.message_subscribe(destination="second", queue="test-queue-2", exclusive=True)
            self.fail("Expected exclusive consume request to fail due to previous consumer")
        except SessionException, e:
            self.assertEquals(405, e.args[0].error_code)

    def test_consume_queue_not_found(self):
        """
        Test error conditions associated with the queue field of the consume method:
        """
        session = self.session
        try:
            #queue specified but doesn't exist:
            session.message_subscribe(queue="invalid-queue", destination="a")
            self.fail("Expected failure when consuming from non-existent queue")
        except SessionException, e:
            self.assertEquals(404, e.args[0].error_code)

    def test_consume_queue_not_specified(self):
        session = self.session
        try:
            #queue not specified and none previously declared for channel:
            session.message_subscribe(destination="a")
            self.fail("Expected failure when consuming from unspecified queue")
        except SessionException, e:
            self.assertEquals(531, e.args[0].error_code)

    def test_consume_unique_consumers(self):
        """
        Ensure unique consumer tags are enforced
        """
        session = self.session
        #setup, declare a queue:
        session.queue_declare(queue="test-queue-3", exclusive=True, auto_delete=True)

        #check that attempts to use duplicate tags are detected and prevented:
        session.message_subscribe(destination="first", queue="test-queue-3")
        try:
            session.message_subscribe(destination="first", queue="test-queue-3")
            self.fail("Expected consume request to fail due to non-unique tag")
        except SessionException, e:
            self.assertEquals(530, e.args[0].error_code)

    def test_cancel(self):
        """
        Test compliance of the basic.cancel method
        """
        session = self.session
        #setup, declare a queue:
        session.queue_declare(queue="test-queue-4", exclusive=True, auto_delete=True)
        session.message_transfer(message=Message(session.delivery_properties(routing_key="test-queue-4"), "One"))

        session.message_subscribe(destination="my-consumer", queue="test-queue-4")
        myqueue = session.incoming("my-consumer")
        session.message_flow(destination="my-consumer", unit=session.credit_unit.message, value=0xFFFFFFFFL)
        session.message_flow(destination="my-consumer", unit=session.credit_unit.byte, value=0xFFFFFFFFL)

        #should flush here

        #cancel should stop messages being delivered
        session.message_cancel(destination="my-consumer")
        session.message_transfer(message=Message(session.delivery_properties(routing_key="test-queue-4"), "Two"))
        msg = myqueue.get(timeout=1)
        self.assertEqual("One", msg.body)
        try:
            msg = myqueue.get(timeout=1)
            self.fail("Got message after cancellation: " + msg)
        except Empty: None

        #cancellation of non-existant consumers should be result in 404s
        try:
            session.message_cancel(destination="my-consumer")
            self.fail("Expected 404 for recancellation of subscription.")
        except SessionException, e:
            self.assertEquals(404, e.args[0].error_code)

        session = self.conn.session("alternate-session", timeout=10)
        try:
            session.message_cancel(destination="this-never-existed")
            self.fail("Expected 404 for cancellation of unknown subscription.")
        except SessionException, e:
            self.assertEquals(404, e.args[0].error_code)


    def test_ack(self):
        """
        Test basic ack/recover behaviour using a combination of implicit and
        explicit accept subscriptions.
        """
        self.startQmf()
        session1 = self.conn.session("alternate-session", timeout=10)
        session1.queue_declare(queue="test-ack-queue", auto_delete=True)

        delivery_properties = session1.delivery_properties(routing_key="test-ack-queue")
        for i in ["One", "Two", "Three", "Four", "Five"]:
            session1.message_transfer(message=Message(delivery_properties, i))

        # verify enqueued message count, use both QMF and session query to verify consistency
        self.assertEqual(5, session1.queue_query(queue="test-ack-queue").message_count)
        queueObj = self.qmf.getObjects(_class="queue", name="test-ack-queue")[0]
        self.assertEquals(queueObj.msgDepth, 5)
        self.assertEquals(queueObj.msgTotalEnqueues, 5)
        self.assertEquals(queueObj.msgTotalDequeues, 0)

        # subscribe with implied acquire, explicit accept:
        session1.message_subscribe(queue = "test-ack-queue", destination = "consumer")
        session1.message_flow(destination="consumer", unit=session1.credit_unit.message, value=0xFFFFFFFFL)
        session1.message_flow(destination="consumer", unit=session1.credit_unit.byte, value=0xFFFFFFFFL)
        queue = session1.incoming("consumer")

        msg1 = queue.get(timeout=1)
        msg2 = queue.get(timeout=1)
        msg3 = queue.get(timeout=1)
        msg4 = queue.get(timeout=1)
        msg5 = queue.get(timeout=1)

        self.assertEqual("One", msg1.body)
        self.assertEqual("Two", msg2.body)
        self.assertEqual("Three", msg3.body)
        self.assertEqual("Four", msg4.body)
        self.assertEqual("Five", msg5.body)

        # messages should not be on the queue:
        self.assertEqual(0, session1.queue_query(queue="test-ack-queue").message_count)
        # QMF shows the dequeues as not having happened yet, since they are have
        # not been accepted
        queueObj.update()
        self.assertEquals(queueObj.msgDepth, 5)
        self.assertEquals(queueObj.msgTotalEnqueues, 5)
        self.assertEquals(queueObj.msgTotalDequeues, 0)

        session1.message_accept(RangedSet(msg1.id, msg2.id, msg4.id))#One, Two and Four

        # QMF should now reflect the accepted messages as being dequeued
        self.assertEqual(0, session1.queue_query(queue="test-ack-queue").message_count)
        queueObj.update()
        self.assertEquals(queueObj.msgDepth, 2)
        self.assertEquals(queueObj.msgTotalEnqueues, 5)
        self.assertEquals(queueObj.msgTotalDequeues, 3)

        #subscribe from second session here to ensure queue is not auto-deleted
        #when alternate session closes.  Use implicit accept mode to test that
        #we don't need to explicitly accept
        session2 = self.conn.session("alternate-session-2", timeout=10)
        session2.message_subscribe(queue = "test-ack-queue", destination = "checker", accept_mode=1)

        #now close the first session, and see that the unaccepted messages are
        #then redelivered to another subscriber:
        session1.close(timeout=10)

        # check the statistics - the queue_query will show the non-accepted
        # messages have been released. QMF never considered them dequeued, so
        # those counts won't change
        self.assertEqual(2, session2.queue_query(queue="test-ack-queue").message_count)
        queueObj.update()
        self.assertEquals(queueObj.msgDepth, 2)
        self.assertEquals(queueObj.msgTotalEnqueues, 5)
        self.assertEquals(queueObj.msgTotalDequeues, 3)

        session2.message_flow(destination="checker", unit=session2.credit_unit.message, value=0xFFFFFFFFL)
        session2.message_flow(destination="checker", unit=session2.credit_unit.byte, value=0xFFFFFFFFL)
        queue = session2.incoming("checker")

        msg3b = queue.get(timeout=1)
        msg5b = queue.get(timeout=1)

        self.assertEqual("Three", msg3b.body)
        self.assertEqual("Five", msg5b.body)

        try:
            extra = queue.get(timeout=1)
            self.fail("Got unexpected message: " + extra.body)
        except Empty: None

        self.assertEqual(0, session2.queue_query(queue="test-ack-queue").message_count)
        queueObj.update()
        self.assertEquals(queueObj.msgDepth, 0)
        self.assertEquals(queueObj.msgTotalEnqueues, 5)
        self.assertEquals(queueObj.msgTotalDequeues, 5)

        # Subscribe one last time to keep the queue available, and to verify
        # that the implied accept worked by verifying no messages have been
        # returned when session2 is closed.
        self.session.message_subscribe(queue = "test-ack-queue", destination = "final-checker")

        session2.close(timeout=10)

        # check the statistics - they should not have changed
        self.assertEqual(0, self.session.queue_query(queue="test-ack-queue").message_count)
        queueObj.update()
        self.assertEquals(queueObj.msgDepth, 0)
        self.assertEquals(queueObj.msgTotalEnqueues, 5)
        self.assertEquals(queueObj.msgTotalDequeues, 5)

        self.session.message_flow(destination="final-checker", unit=self.session.credit_unit.message, value=0xFFFFFFFFL)
        self.session.message_flow(destination="final-checker", unit=self.session.credit_unit.byte, value=0xFFFFFFFFL)
        try:
            extra = self.session.incoming("final-checker").get(timeout=1)
            self.fail("Got unexpected message: " + extra.body)
        except Empty: None

    def test_reject(self):
        session = self.session
        session.queue_declare(queue = "q", exclusive=True, auto_delete=True, alternate_exchange="amq.fanout")
        session.queue_declare(queue = "r", exclusive=True, auto_delete=True)
        session.exchange_bind(queue = "r", exchange = "amq.fanout")

        session.message_subscribe(queue = "q", destination = "consumer")
        session.message_flow(destination="consumer", unit=session.credit_unit.message, value=0xFFFFFFFFL)
        session.message_flow(destination="consumer", unit=session.credit_unit.byte, value=0xFFFFFFFFL)
        session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "blah, blah"))
        msg = session.incoming("consumer").get(timeout = 1)
        self.assertEquals(msg.body, "blah, blah")
        session.message_reject(RangedSet(msg.id))

        session.message_subscribe(queue = "r", destination = "checker")
        session.message_flow(destination="checker", unit=session.credit_unit.message, value=0xFFFFFFFFL)
        session.message_flow(destination="checker", unit=session.credit_unit.byte, value=0xFFFFFFFFL)
        msg = session.incoming("checker").get(timeout = 1)
        self.assertEquals(msg.body, "blah, blah")

    def test_credit_flow_messages(self):
        """
        Test basic credit based flow control with unit = message
        """
        #declare an exclusive queue
        session = self.session
        session.queue_declare(queue = "q", exclusive=True, auto_delete=True)
        #create consumer (for now that defaults to infinite credit)
        session.message_subscribe(queue = "q", destination = "c")
        session.message_set_flow_mode(flow_mode = 0, destination = "c")
        #send batch of messages to queue
        for i in range(1, 11):
            session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "Message %d" % i))

        #set message credit to finite amount (less than enough for all messages)
        session.message_flow(unit = session.credit_unit.message, value = 5, destination = "c")
        #set infinite byte credit
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "c")
        #check that expected number were received
        q = session.incoming("c")
        for i in range(1, 6):
            self.assertDataEquals(session, q.get(timeout = 1), "Message %d" % i)
        self.assertEmpty(q)

        #increase credit again and check more are received
        for i in range(6, 11):
            session.message_flow(unit = session.credit_unit.message, value = 1, destination = "c")
            self.assertDataEquals(session, q.get(timeout = 1), "Message %d" % i)
            self.assertEmpty(q)

    def test_credit_flow_bytes(self):
        """
        Test basic credit based flow control with unit = bytes
        """
        #declare an exclusive queue
        session = self.session
        session.queue_declare(queue = "q", exclusive=True, auto_delete=True)
        #create consumer (for now that defaults to infinite credit)
        session.message_subscribe(queue = "q", destination = "c")
        session.message_set_flow_mode(flow_mode = 0, destination = "c")
        #send batch of messages to queue
        for i in range(10):
            session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "abcdefgh"))

        #each message is currently interpreted as requiring msg_size bytes of credit
        msg_size = 19

        #set byte credit to finite amount (less than enough for all messages)
        session.message_flow(unit = session.credit_unit.byte, value = msg_size*5, destination = "c")
        #set infinite message credit
        session.message_flow(unit = session.credit_unit.message, value = 0xFFFFFFFFL, destination = "c")
        #check that expected number were received
        q = session.incoming("c")
        for i in range(5):
            self.assertDataEquals(session, q.get(timeout = 1), "abcdefgh")
        self.assertEmpty(q)

        #increase credit again and check more are received
        for i in range(5):
            session.message_flow(unit = session.credit_unit.byte, value = msg_size, destination = "c")
            self.assertDataEquals(session, q.get(timeout = 1), "abcdefgh")
            self.assertEmpty(q)


    def test_window_flow_messages(self):
        """
        Test basic window based flow control with unit = message
        """
        #declare an exclusive queue
        session = self.session
        session.queue_declare(queue = "q", exclusive=True, auto_delete=True)
        #create consumer (for now that defaults to infinite credit)
        session.message_subscribe(queue = "q", destination = "c")
        session.message_set_flow_mode(flow_mode = 1, destination = "c")
        #send batch of messages to queue
        for i in range(1, 11):
            session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "Message %d" % i))

        #set message credit to finite amount (less than enough for all messages)
        session.message_flow(unit = session.credit_unit.message, value = 5, destination = "c")
        #set infinite byte credit
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "c")
        #check that expected number were received
        q = session.incoming("c")
        ids = []
        for i in range(1, 6):            
            msg = q.get(timeout = 1)
            ids.append(msg.id)
            self.assertDataEquals(session, msg, "Message %d" % i)
        self.assertEmpty(q)

        #acknowledge messages and check more are received
        #TODO: there may be a nicer way of doing this
        for i in ids:
           session.receiver._completed.add(i)
        session.channel.session_completed(session.receiver._completed)

        for i in range(6, 11):
            self.assertDataEquals(session, q.get(timeout = 1), "Message %d" % i)
        self.assertEmpty(q)


    def test_window_flow_bytes(self):
        """
        Test basic window based flow control with unit = bytes
        """
        #declare an exclusive queue
        session = self.session
        session.queue_declare(queue = "q", exclusive=True, auto_delete=True)
        #create consumer (for now that defaults to infinite credit)
        session.message_subscribe(queue = "q", destination = "c")
        session.message_set_flow_mode(flow_mode = 1, destination = "c")
        #send batch of messages to queue
        for i in range(10):
            session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "abcdefgh"))

        #each message is currently interpreted as requiring msg_size bytes of credit
        msg_size = 19

        #set byte credit to finite amount (less than enough for all messages)
        session.message_flow(unit = session.credit_unit.byte, value = msg_size*5, destination = "c")
        #set infinite message credit
        session.message_flow(unit = session.credit_unit.message, value = 0xFFFFFFFFL, destination = "c")
        #check that expected number were received
        q = session.incoming("c")
        msgs = []
        for i in range(5):
            msg = q.get(timeout = 1)
            msgs.append(msg)
            self.assertDataEquals(session, msg, "abcdefgh")
        self.assertEmpty(q)

        #ack each message individually and check more are received
        for i in range(5):
            msg = msgs.pop()
            #TODO: there may be a nicer way of doing this
            session.receiver._completed.add(msg.id)
            session.channel.session_completed(session.receiver._completed)
            self.assertDataEquals(session, q.get(timeout = 1), "abcdefgh")
            self.assertEmpty(q)

    def test_window_flush_ack_flow(self):
        """
        Test basic window based flow control with unit = bytes
        """
        #declare an exclusive queue
        ssn = self.session
        ssn.queue_declare(queue = "q", exclusive=True, auto_delete=True)
        #create consumer
        ssn.message_subscribe(queue = "q", destination = "c",
                              accept_mode=ssn.accept_mode.explicit)
        ssn.message_set_flow_mode(flow_mode = ssn.flow_mode.window, destination = "c")

        #send message A
        ssn.message_transfer(message=Message(ssn.delivery_properties(routing_key="q"), "A"))

        for unit in ssn.credit_unit.VALUES:
            ssn.message_flow("c", unit, 0xFFFFFFFFL)

        q = ssn.incoming("c")
        msgA = q.get(timeout=10)

        ssn.message_flush(destination="c")

        # XXX
        ssn.receiver._completed.add(msgA.id)
        ssn.channel.session_completed(ssn.receiver._completed)
        ssn.message_accept(RangedSet(msgA.id))

        for unit in ssn.credit_unit.VALUES:
            ssn.message_flow("c", unit, 0xFFFFFFFFL)

        #send message B
        ssn.message_transfer(message=Message(ssn.delivery_properties(routing_key="q"), "B"))

        msgB = q.get(timeout=10)

    def test_window_stop(self):
        """
        Ensure window based flow control reacts to stop correctly
        """
        session = self.session
        #setup subscriber on a test queue
        session.queue_declare(queue = "q", exclusive=True, auto_delete=True)
        session.message_subscribe(queue = "q", destination = "c")
        session.message_set_flow_mode(flow_mode = 1, destination = "c")
        session.message_flow(unit = session.credit_unit.message, value = 5, destination = "c")
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "c")


        #send batch of messages to queue
        for i in range(0, 10):
            session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "Message %d" % (i+1)))

        #retrieve all delivered messages
        q = session.incoming("c")
        for i in range(0, 5):
            msg = q.get(timeout = 1)
            session.receiver._completed.add(msg.id)#TODO: this may be done automatically
            self.assertDataEquals(session, msg, "Message %d" % (i+1))

        session.message_stop(destination = "c")

        #now send completions, normally used to move window forward,
        #but after a stop should not do so
        session.channel.session_completed(session.receiver._completed)

        #check no more messages are sent
        self.assertEmpty(q)

        #re-establish window and check remaining messages
        session.message_flow(unit = session.credit_unit.message, value = 5, destination = "c")
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "c")
        for i in range(0, 5):
            msg = q.get(timeout = 1)
            self.assertDataEquals(session, msg, "Message %d" % (i+6))

    def test_credit_window_after_messagestop(self):
        """
        Tests that the broker's credit window size doesnt exceed the requested value when completing
        previous messageTransfer commands after a message_stop and message_flow.
        """

        session = self.session

        #create queue
        session.queue_declare(queue = self.test_queue_name, exclusive=True, auto_delete=True)

        #send 11 messages
        for i in range(1, 12):
            session.message_transfer(message=Message(session.delivery_properties(routing_key=self.test_queue_name), "message-%d" % (i)))


        #subscribe:
        session.message_subscribe(queue=self.test_queue_name, destination="a")
        a = session.incoming("a")
        session.message_set_flow_mode(flow_mode = 1, destination = "a")
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "a")
        # issue 5 message credits
        session.message_flow(unit = session.credit_unit.message, value = 5, destination = "a")

        # get 5 messages
        ids = RangedSet()
        for i in range(1, 6):
            msg = a.get(timeout = 1)
            self.assertEquals("message-%d" % (i), msg.body)
            ids.add(msg.id)

        # now try and read a 6th message. we expect this to fail due to exhausted message credit.
        try:
            extra = a.get(timeout=1)
            self.fail("Got unexpected message: " + extra.body)
        except Empty: None

        session.message_stop(destination = "a")

        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "a")
        session.message_flow(unit = session.credit_unit.message, value = 5, destination = "a")

        # complete earlier messages after setting the window to 5 message credits
        session.channel.session_completed(ids)

        # Now continue to read the next 5 messages
        for i in range(6, 11):
            msg = a.get(timeout = 1)
            self.assertEquals("message-%d" % (i), msg.body)

        # now try and read the 11th message. we expect this to fail due to exhausted message credit.  If we receive an
        # 11th this indicates the broker is not respecting the client's requested window size.
        try:
            extra = a.get(timeout=1)
            self.fail("Got unexpected message: " + extra.body)
        except Empty: None

    def test_no_credit_wrap(self):
        """
        Ensure that adding credit does not result in wrapround, lowering the balance.
        """
        session = self.session

        session.queue_declare(queue = self.test_queue_name, exclusive=True, auto_delete=True)
        session.message_subscribe(queue=self.test_queue_name, destination="a")
        a = session.incoming("a")
        session.message_set_flow_mode(flow_mode = session.flow_mode.credit, destination = "a")
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "a")
        session.message_flow(unit = session.credit_unit.message, value = 0xFFFFFFFAL, destination = "a")
        #test wraparound of credit balance does not occur
        session.message_flow(unit = session.credit_unit.message, value = 10, destination = "a")
        for i in range(1, 50):
            session.message_transfer(message=Message(session.delivery_properties(routing_key=self.test_queue_name), "message-%d" % (i)))
        session.message_flush(destination = "a")
        for i in range(1, 50):
            msg = a.get(timeout = 1)
            self.assertEquals("message-%d" % (i), msg.body)


    def test_subscribe_not_acquired(self):
        """
        Test the not-acquired modes works as expected for a simple case
        """
        session = self.session
        session.queue_declare(queue = "q", exclusive=True, auto_delete=True)
        for i in range(1, 6):
            session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "Message %s" % i))

        session.message_subscribe(queue = "q", destination = "a", acquire_mode = 1)
        session.message_flow(unit = session.credit_unit.message, value = 0xFFFFFFFFL, destination = "a")
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "a")
        session.message_subscribe(queue = "q", destination = "b", acquire_mode = 1)
        session.message_flow(unit = session.credit_unit.message, value = 0xFFFFFFFFL, destination = "b")
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "b")

        for i in range(6, 11):
            session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "Message %s" % i))

        #both subscribers should see all messages
        qA = session.incoming("a")
        qB = session.incoming("b")
        for i in range(1, 11):
            for q in [qA, qB]:
                msg = q.get(timeout = 1)
                self.assertEquals("Message %s" % i, msg.body)
                #TODO: tidy up completion
                session.receiver._completed.add(msg.id)

        #TODO: tidy up completion
        session.channel.session_completed(session.receiver._completed)
        #messages should still be on the queue:
        self.assertEquals(10, session.queue_query(queue = "q").message_count)

    def test_acquire_with_no_accept_and_credit_flow(self):
        """
        Test that messages recieved unacquired, with accept not
        required in windowing mode can be acquired.
        """
        session = self.session
        session.queue_declare(queue = "q", exclusive=True, auto_delete=True)

        session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "acquire me"))

        session.message_subscribe(queue = "q", destination = "a", acquire_mode = 1, accept_mode = 1)
        session.message_set_flow_mode(flow_mode = session.flow_mode.credit, destination = "a")
        session.message_flow(unit = session.credit_unit.message, value = 0xFFFFFFFFL, destination = "a")
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "a")
        msg = session.incoming("a").get(timeout = 1)
        self.assertEquals("acquire me", msg.body)
        #message should still be on the queue:
        self.assertEquals(1, session.queue_query(queue = "q").message_count)

        transfers = RangedSet(msg.id)
        response = session.message_acquire(transfers)
        #check that we get notification (i.e. message_acquired)
        self.assert_(msg.id in response.transfers)
        #message should have been removed from the queue:
        self.assertEquals(0, session.queue_query(queue = "q").message_count)

    def test_acquire(self):
        """
        Test explicit acquire function
        """
        session = self.session
        session.queue_declare(queue = "q", exclusive=True, auto_delete=True)

        session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "acquire me"))

        session.message_subscribe(queue = "q", destination = "a", acquire_mode = 1)
        session.message_flow(destination="a", unit=session.credit_unit.message, value=0xFFFFFFFFL)
        session.message_flow(destination="a", unit=session.credit_unit.byte, value=0xFFFFFFFFL)
        msg = session.incoming("a").get(timeout = 1)
        self.assertEquals("acquire me", msg.body)
        #message should still be on the queue:
        self.assertEquals(1, session.queue_query(queue = "q").message_count)

        transfers = RangedSet(msg.id)
        response = session.message_acquire(transfers)
        #check that we get notification (i.e. message_acquired)
        self.assert_(msg.id in response.transfers)
        #message should have been removed from the queue:
        self.assertEquals(0, session.queue_query(queue = "q").message_count)
        session.message_accept(transfers)


    def test_release(self):
        """
        Test explicit release function
        """
        session = self.session
        session.queue_declare(queue = "q", exclusive=True, auto_delete=True)

        session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "release me"))

        session.message_subscribe(queue = "q", destination = "a")
        session.message_flow(destination="a", unit=session.credit_unit.message, value=0xFFFFFFFFL)
        session.message_flow(destination="a", unit=session.credit_unit.byte, value=0xFFFFFFFFL)
        msg = session.incoming("a").get(timeout = 1)
        self.assertEquals("release me", msg.body)
        session.message_cancel(destination = "a")
        session.message_release(RangedSet(msg.id))

        #message should not have been removed from the queue:
        self.assertEquals(1, session.queue_query(queue = "q").message_count)

    def test_release_ordering(self):
        """
        Test order of released messages is as expected
        """
        session = self.session
        session.queue_declare(queue = "q", exclusive=True, auto_delete=True)
        for i in range (1, 11):
            session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "released message %s" % (i)))

        session.message_subscribe(queue = "q", destination = "a")
        session.message_flow(unit = session.credit_unit.message, value = 10, destination = "a")
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "a")
        queue = session.incoming("a")
        first = queue.get(timeout = 1)
        for i in range(2, 10):
            msg = queue.get(timeout = 1)
            self.assertEquals("released message %s" % (i), msg.body)
            
        last = queue.get(timeout = 1)
        self.assertEmpty(queue)
        released = RangedSet()
        released.add(first.id, last.id)
        session.message_release(released)

        #TODO: may want to clean this up...
        session.receiver._completed.add(first.id, last.id)
        session.channel.session_completed(session.receiver._completed)
        
        for i in range(1, 11):
            self.assertEquals("released message %s" % (i), queue.get(timeout = 1).body)

    def test_ranged_ack(self):
        """
        Test acking of messages ranges
        """
        session = self.conn.session("alternate-session", timeout=10)

        session.queue_declare(queue = "q", auto_delete=True)
        delivery_properties = session.delivery_properties(routing_key="q")
        for i in range (1, 11):
            session.message_transfer(message=Message(delivery_properties, "message %s" % (i)))

        session.message_subscribe(queue = "q", destination = "a")
        session.message_flow(unit = session.credit_unit.message, value = 10, destination = "a")
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "a")
        queue = session.incoming("a")
        ids = []
        for i in range (1, 11):
            msg = queue.get(timeout = 1)
            self.assertEquals("message %s" % (i), msg.body)
            ids.append(msg.id)
            
        self.assertEmpty(queue)

        #ack all but the fourth message (command id 2)
        accepted = RangedSet()
        accepted.add(ids[0], ids[2])
        accepted.add(ids[4], ids[9])
        session.message_accept(accepted)

        #subscribe from second session here to ensure queue is not
        #auto-deleted when alternate session closes (no need to ack on these):
        self.session.message_subscribe(queue = "q", destination = "checker")

        #now close the session, and see that the unacked messages are
        #then redelivered to another subscriber:
        session.close(timeout=10)

        session = self.session
        session.message_flow(destination="checker", unit=session.credit_unit.message, value=0xFFFFFFFFL)
        session.message_flow(destination="checker", unit=session.credit_unit.byte, value=0xFFFFFFFFL)
        queue = session.incoming("checker")

        self.assertEquals("message 4", queue.get(timeout = 1).body)
        self.assertEmpty(queue)

    def test_subscribe_not_acquired_2(self):
        session = self.session

        #publish some messages
        session.queue_declare(queue = "q", exclusive=True, auto_delete=True)
        for i in range(1, 11):
            session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "message-%d" % (i)))

        #consume some of them
        session.message_subscribe(queue = "q", destination = "a")
        session.message_set_flow_mode(flow_mode = 0, destination = "a")
        session.message_flow(unit = session.credit_unit.message, value = 5, destination = "a")
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "a")

        queue = session.incoming("a")
        for i in range(1, 6):
            msg = queue.get(timeout = 1)
            self.assertEquals("message-%d" % (i), msg.body)
            #complete and accept
            session.message_accept(RangedSet(msg.id))
            #TODO: tidy up completion
            session.receiver._completed.add(msg.id)
            session.channel.session_completed(session.receiver._completed)
        self.assertEmpty(queue)

        #now create a not-acquired subscriber
        session.message_subscribe(queue = "q", destination = "b", acquire_mode=1)
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "b")

        #check it gets those not consumed
        queue = session.incoming("b")
        session.message_flow(unit = session.credit_unit.message, value = 1, destination = "b")
        for i in range(6, 11):
            msg = queue.get(timeout = 1)
            self.assertEquals("message-%d" % (i), msg.body)
            session.message_release(RangedSet(msg.id))
            #TODO: tidy up completion
            session.receiver._completed.add(msg.id)
            session.channel.session_completed(session.receiver._completed)
        session.message_flow(unit = session.credit_unit.message, value = 1, destination = "b")
        self.assertEmpty(queue)

        #check all 'browsed' messages are still on the queue
        self.assertEqual(5, session.queue_query(queue="q").message_count)

    def test_subscribe_not_acquired_3(self):
        session = self.session

        #publish some messages
        session.queue_declare(queue = "q", exclusive=True, auto_delete=True)
        for i in range(1, 11):
            session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "message-%d" % (i)))

        #create a not-acquired subscriber
        session.message_subscribe(queue = "q", destination = "a", acquire_mode=1)
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "a")
        session.message_flow(unit = session.credit_unit.message, value = 10, destination = "a")

        #browse through messages
        queue = session.incoming("a")
        for i in range(1, 11):
            msg = queue.get(timeout = 1)
            self.assertEquals("message-%d" % (i), msg.body)
            if (i % 2):
                #try to acquire every second message
                response = session.message_acquire(RangedSet(msg.id))
                #check that acquire succeeds
                self.assert_(msg.id in response.transfers)
                session.message_accept(RangedSet(msg.id))
            else:
                session.message_release(RangedSet(msg.id))
            session.receiver._completed.add(msg.id)
            session.channel.session_completed(session.receiver._completed)
        self.assertEmpty(queue)

        #create a second not-acquired subscriber
        session.message_subscribe(queue = "q", destination = "b", acquire_mode=1)
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "b")
        session.message_flow(unit = session.credit_unit.message, value = 1, destination = "b")
        #check it gets those not consumed
        queue = session.incoming("b")
        for i in [2,4,6,8,10]:
            msg = queue.get(timeout = 1)
            self.assertEquals("message-%d" % (i), msg.body)
            session.message_release(RangedSet(msg.id))
            session.receiver._completed.add(msg.id)
            session.channel.session_completed(session.receiver._completed)
        session.message_flow(unit = session.credit_unit.message, value = 1, destination = "b")
        self.assertEmpty(queue)

        #check all 'browsed' messages are still on the queue
        self.assertEqual(5, session.queue_query(queue="q").message_count)

    def test_release_unacquired(self):
        session = self.session

        #create queue
        session.queue_declare(queue = "q", exclusive=True, auto_delete=True)

        #send message
        session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "my-message"))

        #create two 'browsers'
        session.message_subscribe(queue = "q", destination = "a", acquire_mode=1)
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "a")
        session.message_flow(unit = session.credit_unit.message, value = 10, destination = "a")
        queueA = session.incoming("a")

        session.message_subscribe(queue = "q", destination = "b", acquire_mode=1)
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "b")
        session.message_flow(unit = session.credit_unit.message, value = 10, destination = "b")
        queueB = session.incoming("b")
        
        #have each browser release the message
        msgA = queueA.get(timeout = 1)
        session.message_release(RangedSet(msgA.id))

        msgB = queueB.get(timeout = 1)
        session.message_release(RangedSet(msgB.id))
        
        #cancel browsers
        session.message_cancel(destination = "a")
        session.message_cancel(destination = "b")
        
        #create consumer
        session.message_subscribe(queue = "q", destination = "c")
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "c")
        session.message_flow(unit = session.credit_unit.message, value = 10, destination = "c")
        queueC = session.incoming("c")
        #consume the message then ack it
        msgC = queueC.get(timeout = 1)
        session.message_accept(RangedSet(msgC.id))
        #ensure there are no other messages
        self.assertEmpty(queueC)

    def test_release_order(self):
        session = self.session

        #create queue
        session.queue_declare(queue = "q", exclusive=True, auto_delete=True)

        #send messages
        for i in range(1, 11):
            session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "message-%d" % (i)))

        #subscribe:
        session.message_subscribe(queue="q", destination="a")
        a = session.incoming("a")
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "a")
        session.message_flow(unit = session.credit_unit.message, value = 10, destination = "a")

        # receive all messages into list
        messages = [];
        for i in range(1, 11):
            msg = a.get(timeout = 1)
            self.assertEquals("message-%d" % (i), msg.body)
            messages.append(msg)

        # accept/release received messages
        for i, msg in enumerate(messages, start=1):
            if (i % 2):
                #accept all odd messages
                session.message_accept(RangedSet(msg.id))
            else:
                #release all even messages
                session.message_release(RangedSet(msg.id))

        session.message_subscribe(queue="q", destination="b", acquire_mode=0)
        b = session.incoming("b")
        b.start()
        for i in [2, 4, 6, 8, 10]:
            msg = b.get(timeout = 1)
            self.assertEquals("message-%d" % (i), msg.body)


    def test_empty_body(self):
        session = self.session
        session.queue_declare(queue="xyz", exclusive=True, auto_delete=True)
        props = session.delivery_properties(routing_key="xyz")
        session.message_transfer(message=Message(props, ""))

        consumer_tag = "tag1"
        session.message_subscribe(queue="xyz", destination=consumer_tag)
        session.message_flow(unit = session.credit_unit.message, value = 0xFFFFFFFFL, destination = consumer_tag)
        session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = consumer_tag)
        queue = session.incoming(consumer_tag)
        msg = queue.get(timeout=1)
        self.assertEquals("", msg.body)
        session.message_accept(RangedSet(msg.id))

    def test_incoming_start(self):
        q = "test_incoming_start"
        session = self.session

        session.queue_declare(queue=q, exclusive=True, auto_delete=True)
        session.message_subscribe(queue=q, destination="msgs")
        messages = session.incoming("msgs")
        assert messages.destination == "msgs"

        dp = session.delivery_properties(routing_key=q)
        session.message_transfer(message=Message(dp, "test"))

        messages.start()
        msg = messages.get()
        assert msg.body == "test"

    def test_ttl(self):
        q = "test_ttl"
        session = self.session

        session.queue_declare(queue=q, exclusive=True, auto_delete=True)

        dp = session.delivery_properties(routing_key=q, ttl=500)#expire in half a second
        session.message_transfer(message=Message(dp, "first"))

        dp = session.delivery_properties(routing_key=q, ttl=300000)#expire in fives minutes
        session.message_transfer(message=Message(dp, "second"))

        d = "msgs"
        session.message_subscribe(queue=q, destination=d)
        messages = session.incoming(d)
        sleep(1)
        session.message_flow(unit = session.credit_unit.message, value=2, destination=d)
        session.message_flow(unit = session.credit_unit.byte, value=0xFFFFFFFFL, destination=d)
        assert messages.get(timeout=1).body == "second"
        self.assertEmpty(messages)

    def assertDataEquals(self, session, msg, expected):
        self.assertEquals(expected, msg.body)

    def assertEmpty(self, queue):
        try:
            extra = queue.get(timeout=1)
            self.fail("Queue not empty, contains: " + extra.body)
        except Empty: None

class SizelessContent(Content):

    def size(self):
        return None