#
# 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 __future__ import unicode_literals
from __future__ import division
from __future__ import absolute_import
from __future__ import print_function

from time import sleep
import json, os
import logging
from threading import Timer
from subprocess import PIPE, STDOUT
from proton import Message, Timeout, Delivery, symbol, Condition
from system_test import Logger, TestCase, Process, Qdrouterd, main_module, TIMEOUT, DIR, TestTimeout
from system_test import AsyncTestReceiver
from system_test import AsyncTestSender
from system_test import get_inter_router_links
from system_test import unittest
from test_broker import FakeService

from proton.handlers import MessagingHandler
from proton.reactor import Container, AtLeastOnce
from proton.utils import BlockingConnection
from qpid_dispatch.management.client import Node
CONNECTION_PROPERTIES_UNICODE_STRING = {u'connection': u'properties', u'int_property': 6451}

class TwoRouterTest(TestCase):

    inter_router_port = None

    @classmethod
    def setUpClass(cls):
        """Start a router and a messenger"""
        super(TwoRouterTest, cls).setUpClass()

        def router(name, client_server, connection):
            policy_config_path = os.path.join(DIR, 'two-router-policy')

            config = [
                # Use the deprecated attributes helloInterval, raInterval, raIntervalFlux, remoteLsMaxAge
                # The routers should still start successfully after using these deprecated entities.
                ('router', {'remoteLsMaxAge': 60, 'helloInterval': 1, 'raInterval': 30, 'raIntervalFlux': 4,
                            'mode': 'interior', 'id': 'QDR.%s'%name, 'allowUnsettledMulticast': 'yes'}),
                ('listener', {'port': cls.tester.get_port(), 'stripAnnotations': 'no', 'linkCapacity': 500}),

                ('listener', {'port': cls.tester.get_port(), 'stripAnnotations': 'no'}),
                ('listener', {'port': cls.tester.get_port(), 'stripAnnotations': 'both'}),
                ('listener', {'port': cls.tester.get_port(), 'stripAnnotations': 'out'}),
                ('listener', {'port': cls.tester.get_port(), 'stripAnnotations': 'in'}),

                ('address', {'prefix': 'closest', 'distribution': 'closest'}),
                ('address', {'prefix': 'balanced', 'distribution': 'balanced'}),
                ('address', {'prefix': 'multicast', 'distribution': 'multicast'}),

                # for testing pattern matching
                ('address', {'pattern': 'a.b.c.d',
                             'distribution': 'closest'}),
                ('address', {'pattern': '#.b.c.d',
                             'distribution': 'multicast'}),
                ('address', {'pattern': 'a/*/#/d',
                             'distribution': 'closest'}),
                ('address', {'pattern': '*/b/c/d',
                             'distribution': 'multicast'}),
                ('address', {'pattern': 'a.x.d',
                             'distribution': 'closest'}),
                ('address', {'pattern': 'a.*.d',
                             'distribution': 'multicast'}),
                connection
            ]

            config = Qdrouterd.Config(config)

            cls.routers.append(cls.tester.qdrouterd(name, config, wait=True))

        cls.routers = []

        inter_router_port = cls.tester.get_port()

        router('A', 'server',
               ('listener', {'role': 'inter-router', 'port': inter_router_port}))

        router('B', 'client',
               ('connector', {'name': 'connectorToA', 'role': 'inter-router', 'port': inter_router_port}))

        cls.routers[0].wait_router_connected('QDR.B')
        cls.routers[1].wait_router_connected('QDR.A')

    def address(self):
        return self.routers[0].addresses[0]

    def run_qdmanage(self, cmd, input=None, expect=Process.EXIT_OK, address=None):
        p = self.popen(
            ['qdmanage'] + cmd.split(' ') + ['--bus', address or self.address(), '--indent=-1', '--timeout', str(TIMEOUT)],
            stdin=PIPE, stdout=PIPE, stderr=STDOUT, expect=expect,
            universal_newlines=True)
        out = p.communicate(input)[0]
        try:
            p.teardown()
        except Exception as e:
            raise Exception(out if out else str(e))
        return out

    def test_01_pre_settled(self):
        test = DeliveriesInTransit(self.routers[0].addresses[0], self.routers[1].addresses[0])
        test.run()
        self.assertEqual(None, test.error)

        local_node = Node.connect(self.routers[0].addresses[0], timeout=TIMEOUT)
        outs = local_node.query(type='org.apache.qpid.dispatch.router')

        # deliveriesTransit must most surely be greater than num_msgs
        pos = outs.attribute_names.index("deliveriesTransit")
        results = outs.results[0]
        self.assertTrue(results[pos] > 104)

    def test_02a_multicast_unsettled(self):
        test = MulticastUnsettled(self.routers[0].addresses[0])
        test.run()
        self.assertEqual(None, test.error)

    def test_02c_sender_settles_first(self):
        test = SenderSettlesFirst(self.routers[0].addresses[0], self.routers[1].addresses[0])
        test.run()
        self.assertEqual(None, test.error)

    def test_03_message_annotations(self):
        test = MessageAnnotationsTest(self.routers[0].addresses[0], self.routers[1].addresses[0])
        test.run()
        self.assertEqual(None, test.error)

    def test_03a_test_strip_message_annotations_no(self):
        test = MessageAnnotationsStripTest(self.routers[0].addresses[1], self.routers[1].addresses[1])
        test.run()
        self.assertEqual(None, test.error)

    def test_03a_test_strip_message_annotations_no_add_trace(self):
        test = MessageAnnotationsStripAddTraceTest(self.routers[0].addresses[1], self.routers[1].addresses[1])
        test.run()
        # Dump the logger output only if there is a test error, otherwise dont bother
        if test.error:
            test.logger.dump()
        self.assertEqual(None, test.error)

    def test_03a_test_strip_message_annotations_both_add_ingress_trace(self):
        test = MessageAnnotationsStripBothAddIngressTrace(self.routers[0].addresses[2], self.routers[1].addresses[2])
        test.run()
        self.assertEqual(None, test.error)

    def test_03a_test_strip_message_annotations_out(self):
        test = MessageAnnotationsStripMessageAnnotationsOut(self.routers[0].addresses[3], self.routers[1].addresses[3])
        test.run()
        self.assertEqual(None, test.error)

    def test_03a_test_strip_message_annotations_in(self):
        test = MessageAnnotationStripMessageAnnotationsIn(self.routers[0].addresses[4], self.routers[1].addresses[4])
        test.run()
        self.assertEqual(None, test.error)

    def test_04_management(self):
        test = ManagementTest(self.routers[0].addresses[0])
        test.run()
        self.assertEqual(None, test.error)

    def test_06_semantics_closest_is_local(self):
        test = SemanticsClosestIsLocal(self.routers[0].addresses[0], self.routers[1].addresses[0])
        test.run()
        self.assertEqual(None, test.error)

    def test_07_semantics_closest_is_remote(self):
        test = SemanticsClosestIsRemote(self.routers[0].addresses[0], self.routers[1].addresses[0])
        test.run()
        self.assertEqual(None, test.error)

    def test_08_semantics_balanced(self):
        test = SemanticsBalanced(self.routers[0].addresses[0], self.routers[0].addresses[1],
                                 self.routers[1].addresses[0])
        test.run()
        self.assertEqual(None, test.error)

    def test_09_to_override(self):
        test = MessageAnnotaionsPreExistingOverride(self.routers[0].addresses[0], self.routers[1].addresses[0])
        test.run()
        self.assertEqual(None, test.error)

    def test_10_propagated_disposition(self):
        test = PropagatedDisposition(self, self.routers[0].addresses[0], self.routers[1].addresses[0])
        test.run()
        self.assertTrue(test.passed)

    def test_10a_propagated_disposition_data(self):
        test = PropagatedDispositionData(self, self.routers[0].addresses[0], self.routers[1].addresses[0])
        test.run()
        self.assertTrue(test.passed)

    def test_11_three_ack(self):
        test = ThreeAck(self, self.routers[0].addresses[0], self.routers[1].addresses[0])
        test.run()

    def test_12_excess_deliveries_released(self):
        """
        Message-route a series of deliveries where the receiver provides credit for a subset and
        once received, closes the link.  The remaining deliveries should be released back to the sender.
        """
        test = ExcessDeliveriesReleasedTest(self.routers[0].addresses[0], self.routers[1].addresses[0])
        test.run()
        self.assertEqual(None, test.error)

    def test_15_attach_on_inter_router(self):
        test = AttachOnInterRouterTest(self.routers[0].addresses[5])
        test.run()
        self.assertEqual(None, test.error)

    def test_17_address_wildcard(self):
        # verify proper distribution is selected by wildcard
        addresses = [
            # (address, count of messages expected to be received)
            ('a.b.c.d',   1), # closest 'a.b.c.d'
            ('b.c.d',     2), # multi   '#.b.c.d'
            ('f.a.b.c.d', 2), # multi   '#.b.c.d
            ('a.c.d',     2), # multi   'a.*.d'
            ('a/c/c/d',   1), # closest 'a/*/#.d
            ('a/x/z/z/d', 1), # closest 'a/*/#.d
            ('a/x/d',     1), # closest 'a.x.d'
            ('a.x.e',     1), # balanced  ----
            ('m.b.c.d',   2)  # multi   '*/b/c/d'
        ]

        # two receivers per address - one for each router
        receivers = []
        for a in addresses:
            for x in range(2):
                ar = AsyncTestReceiver(address=self.routers[x].addresses[0],
                                       source=a[0])
                receivers.append(ar)

        # wait for the consumer info to propagate
        for a in addresses:
            self.routers[0].wait_address(a[0], 1, 1)
            self.routers[1].wait_address(a[0], 1, 1)

        # send one message to each address
        conn = BlockingConnection(self.routers[0].addresses[0])
        sender = conn.create_sender(address=None, options=AtLeastOnce())
        for a in addresses:
            sender.send(Message(address=a[0], body={'address': a[0]}))

        # count received messages by address
        msgs_recvd = {}
        for M in receivers:
            try:
                while True:
                    i = M.queue.get(timeout=0.2).body.get('address', "ERROR")
                    if i not in msgs_recvd:
                        msgs_recvd[i] = 0
                    msgs_recvd[i] += 1
            except AsyncTestReceiver.Empty:
                pass

        # verify expected count == actual count
        self.assertNotIn("ERROR", msgs_recvd)
        for a in addresses:
            self.assertIn(a[0], msgs_recvd)
            self.assertEqual(a[1], msgs_recvd[a[0]])

        for M in receivers:
            M.stop()
        conn.close()

    def test_17_large_streaming_test(self):
        test = LargeMessageStreamTest(self.routers[0].addresses[0], self.routers[1].addresses[0])
        test.run()
        self.assertEqual(None, test.error)

    def test_18_single_char_dest_test(self):
        test = SingleCharacterDestinationTest(self.routers[0].addresses[0], self.routers[1].addresses[0])
        test.run()
        self.assertEqual(None, test.error)

    def test_19_delete_inter_router_connection(self):
        """
        This test tries to delete an inter-router connection but is
        prevented from doing so.
        """
        query_command = 'QUERY --type=connection'
        outputs = json.loads(self.run_qdmanage(query_command))
        identity = None
        passed = False

        for output in outputs:
            if "inter-router" == output['role']:
                identity = output['identity']
                if identity:
                    update_command = 'UPDATE --type=connection adminStatus=deleted --id=' + identity
                    try:
                        json.loads(self.run_qdmanage(update_command))
                    except Exception as e:
                        if "Forbidden" in str(e):
                            passed = True

        # The test has passed since we were forbidden from deleting
        # inter-router connections even though we are allowed to update the adminStatus field.
        self.assertTrue(passed)

    def test_20_delete_connection(self):
        """
        This test creates a blocking connection and tries to delete that connection.
        Since  there is no policy associated with this router, the default for allowAdminStatusUpdate is true,
        the delete operation will be permitted.
        """

        # Create a connection with some properties so we can easily identify the connection
        connection = BlockingConnection(self.address(),
                                        properties=CONNECTION_PROPERTIES_UNICODE_STRING)
        query_command = 'QUERY --type=connection'
        outputs = json.loads(self.run_qdmanage(query_command))
        identity = None
        passed = False

        print ()

        for output in outputs:
            if output.get('properties'):
                conn_properties = output['properties']
                # Find the connection that has our properties - CONNECTION_PROPERTIES_UNICODE_STRING
                # Delete that connection and run another qdmanage to see
                # if the connection is gone.
                if conn_properties.get('int_property'):
                    identity = output.get("identity")
                    if identity:
                        update_command = 'UPDATE --type=connection adminStatus=deleted --id=' + identity
                        try:
                            self.run_qdmanage(update_command)
                            query_command = 'QUERY --type=connection'
                            outputs = json.loads(
                                self.run_qdmanage(query_command))
                            no_properties = True
                            for output in outputs:
                                if output.get('properties'):
                                    no_properties = False
                                    conn_properties = output['properties']
                                    if conn_properties.get('int_property'):
                                        passed = False
                                        break
                                    else:
                                        passed = True
                            if no_properties:
                                passed = True
                        except Exception as e:
                            passed = False

        # The test has passed since we were allowed to delete a connection
        # because we have the policy permission to do so.
        self.assertTrue(passed)

    def test_21_delete_connection_with_receiver(self):
        test = DeleteConnectionWithReceiver(self.routers[0].addresses[0])
        self.assertEqual(test.error, None)
        test.run()

    def test_30_huge_address(self):
        # try a link with an extremely long address
        # DISPATCH-1461
        addr = "A" * 2019
        rx = AsyncTestReceiver(self.routers[0].addresses[0],
                               source=addr)
        tx = AsyncTestSender(self.routers[1].addresses[0],
                             target=addr,
                             count=100)
        tx.wait()

        i = 100
        while i:
            try:
                rx.queue.get(timeout=TIMEOUT)
                i -= 1
            except AsyncTestReceiver.Empty:
                break;
        self.assertEqual(0, i)
        rx.stop()


class DeleteConnectionWithReceiver(MessagingHandler):
    def __init__(self, address):
        super(DeleteConnectionWithReceiver, self).__init__()
        self.address = address
        self.mgmt_receiver = None
        self.mgmt_receiver_1 = None
        self.mgmt_receiver_2 = None
        self.conn_to_kill = None
        self.mgmt_conn = None
        self.mgmt_sender = None
        self.success = False
        self.error = None


    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))

        # Create a receiver connection with some properties so it
        # can be easily identified.
        self.conn_to_kill = event.container.connect(self.address, properties=CONNECTION_PROPERTIES_UNICODE_STRING)
        self.receiver_to_kill = event.container.create_receiver(self.conn_to_kill, "hello_world")
        self.mgmt_conn = event.container.connect(self.address)
        self.mgmt_sender = event.container.create_sender(self.mgmt_conn)
        self.mgmt_receiver = event.container.create_receiver(self.mgmt_conn, None, dynamic=True)
        self.mgmt_receiver_1 = event.container.create_receiver(self.mgmt_conn,
                                                             None,
                                                             dynamic=True)
        self.mgmt_receiver_2 = event.container.create_receiver(self.mgmt_conn,
                                                             None,
                                                             dynamic=True)
    def timeout(self):
        self.error = "Timeout Expired: sent=%d, received=%d" % (self.n_sent, self.n_received)
        self.mgmt_conn.close()

    def bail(self, error):
        self.error = error
        self.timer.cancel()
        self.mgmt_conn.close()
        self.conn_to_kill.close()

    def on_link_opened(self, event):
        if event.receiver == self.mgmt_receiver:
            request = Message()
            request.address = "amqp:/_local/$management"
            request.properties = {
                u'type': u'org.apache.qpid.dispatch.connection',
                u'operation': u'QUERY'}
            request.reply_to = self.mgmt_receiver.remote_source.address
            self.mgmt_sender.send(request)

    def on_message(self, event):
        if event.receiver == self.mgmt_receiver:
            attribute_names = event.message.body['attributeNames']
            property_index = attribute_names .index('properties')
            identity_index = attribute_names .index('identity')

            for result in event.message.body['results']:
                if result[property_index]:
                    properties = result[property_index]
                    if properties.get('int_property'):
                        identity = result[identity_index]
                        if identity:
                            request = Message()
                            request.address = "amqp:/_local/$management"
                            request.properties = {
                                u'identity': identity,
                                u'type': u'org.apache.qpid.dispatch.connection',
                                u'operation': u'UPDATE'
                            }
                            request.body = {
                                u'adminStatus':  u'deleted'}
                            request.reply_to = self.mgmt_receiver_1.remote_source.address
                            self.mgmt_sender.send(request)
        elif event.receiver == self.mgmt_receiver_1:
            if event.message.properties['statusDescription'] == 'OK' and event.message.body['adminStatus'] == 'deleted':
                request = Message()
                request.address = "amqp:/_local/$management"
                request.properties = {u'type': u'org.apache.qpid.dispatch.connection',
                                      u'operation': u'QUERY'}
                request.reply_to = self.mgmt_receiver_2.remote_source.address
                self.mgmt_sender.send(request)

        elif event.receiver == self.mgmt_receiver_2:
            attribute_names = event.message.body['attributeNames']
            property_index = attribute_names .index('properties')
            identity_index = attribute_names .index('identity')

            for result in event.message.body['results']:
                if result[property_index]:
                    properties = result[property_index]
                    if properties and properties.get('int_property'):
                        self.bail("Connection not deleted")
            self.bail(None)

    def run(self):
        Container(self).run()


class SingleCharacterDestinationTest(MessagingHandler):
    def __init__(self, address1, address2):
        super(SingleCharacterDestinationTest, self).__init__()
        self.address1 = address1
        self.address2 = address2
        self.dest = "x"
        self.error = None
        self.conn1 = None
        self.conn2 = None
        self.count = 1
        self.n_sent = 0
        self.timer = None
        self.sender = None
        self.receiver = None
        self.n_received = 0
        self.body = "xyz"

    def check_if_done(self):
        if self.n_received == self.count:
            self.timer.cancel()
            self.conn1.close()
            self.conn2.close()

    def timeout(self):
        self.error = "Timeout Expired: sent=%d, received=%d" % (self.n_sent, self.n_received)
        self.conn1.close()
        self.conn2.close()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn1 = event.container.connect(self.address1)
        self.conn2 = event.container.connect(self.address2)
        self.sender = event.container.create_sender(self.conn1, self.dest)
        self.receiver = event.container.create_receiver(self.conn2, self.dest)

    def on_sendable(self, event):
        if self.n_sent < self.count:
            msg = Message(body=self.body)
            event.sender.send(msg)
            self.n_sent += 1

    def on_message(self, event):
        self.n_received += 1
        self.check_if_done()

    def run(self):
        Container(self).run()


class LargeMessageStreamTest(MessagingHandler):
    def __init__(self, address1, address2):
        super(LargeMessageStreamTest, self).__init__()
        self.address1 = address1
        self.address2 = address2
        self.dest = "LargeMessageStreamTest"
        self.error = None
        self.conn1 = None
        self.conn2 = None
        self.count = 10
        self.n_sent = 0
        self.timer = None
        self.sender = None
        self.receiver = None
        self.n_received = 0
        self.body = ""
        for i in range(10000):
            self.body += "0123456789101112131415"

    def check_if_done(self):
        if self.n_received == self.count:
            self.timer.cancel()
            self.conn1.close()
            self.conn2.close()

    def timeout(self):
        self.error = "Timeout Expired: sent=%d, received=%d" % (self.n_sent, self.n_received)
        self.conn1.close()
        self.conn2.close()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn1 = event.container.connect(self.address1)
        self.conn2 = event.container.connect(self.address2)
        self.sender = event.container.create_sender(self.conn1, self.dest)
        self.receiver = event.container.create_receiver(self.conn2, self.dest)

    def on_sendable(self, event):
        if self.n_sent < self.count:
            msg = Message(body=self.body)
            # send(msg) calls the stream function which streams data from sender to the router
            event.sender.send(msg)
            self.n_sent += 1

    def on_message(self, event):
        self.n_received += 1
        self.check_if_done()

    def run(self):
        Container(self).run()


class ExcessDeliveriesReleasedTest(MessagingHandler):
    def __init__(self, address1, address2):
        super(ExcessDeliveriesReleasedTest, self).__init__(prefetch=0)
        self.address1 = address1
        self.address2 = address2
        self.dest = "closest.EDRtest"
        self.error = None
        self.sender = None
        self.receiver = None
        self.n_sent  = 0
        self.n_received = 0
        self.n_accepted = 0
        self.n_released = 0
        self.timer = None
        self.conn1 = None
        self.conn2 = None

    def timeout(self):
        self.error = "Timeout Expired"
        self.conn1.close()
        self.conn2.close()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn1 = event.container.connect(self.address1)
        self.conn2 = event.container.connect(self.address2)
        self.sender   = event.container.create_sender(self.conn1, self.dest)
        self.receiver = event.container.create_receiver(self.conn2, self.dest)
        self.receiver.flow(6)

    def on_sendable(self, event):
        for i in range(10 - self.n_sent):
            msg = Message(body=i)
            event.sender.send(msg)
            self.n_sent += 1

    def on_accepted(self, event):
        self.n_accepted += 1

    def on_released(self, event):
        self.n_released += 1
        if self.n_released == 4:
            if self.n_accepted != 6:
                self.error = "Expected 6 accepted, got %d" % self.n_accepted
            if self.n_received != 6:
                self.error = "Expected 6 received, got %d" % self.n_received
            self.conn1.close()
            self.conn2.close()
            self.timer.cancel()

    def on_message(self, event):
        self.n_received += 1
        if self.n_received == 6:
            self.receiver.close()

    def run(self):
        Container(self).run()


class AttachOnInterRouterTest(MessagingHandler):
    """Expect an error when attaching a link to an inter-router listener"""
    def __init__(self, address):
        super(AttachOnInterRouterTest, self).__init__(prefetch=0)
        self.address = address
        self.dest = "AOIRtest"
        self.error = None
        self.sender = None
        self.timer = None
        self.conn = None

    def timeout(self):
        self.error = "Timeout Expired"
        self.conn.close()

    def on_start(self, event):
        self.timer  = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn   = event.container.connect(self.address)
        self.sender = event.container.create_sender(self.conn, self.dest)

    def on_link_remote_close(self, event):
        self.conn.close()
        self.timer.cancel()

    def run(self):
        logging.disable(logging.ERROR) # Hide expected log errors
        try:
            Container(self).run()
        finally:
            logging.disable(logging.NOTSET) # Restore to normal


class DeliveriesInTransit(MessagingHandler):
    def __init__(self, address1, address2):
        super(DeliveriesInTransit, self).__init__()
        self.address1 = address1
        self.address2 = address2
        self.dest = "pre_settled.1"
        self.error = "All messages not received"
        self.n_sent = 0
        self.timer = None
        self.conn1 = None
        self.conn2 = None
        self.sender = None
        self.num_msgs = 104
        self.sent_count = 0
        self.received_count = 0
        self.receiver = None

    def timeout(self):
        self.error = "Timeout Expired: n_sent=%d n_received_count=%d" % (self.n_sent, self.received_count)
        self.conn1.close()
        self.conn2.close()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn1 = event.container.connect(self.address1)
        self.sender = event.container.create_sender(self.conn1, self.dest)
        self.conn2 = event.container.connect(self.address2)
        self.receiver = event.container.create_receiver(self.conn2, self.dest)

    def on_sendable(self, event):
        if self.n_sent <= self.num_msgs-1:
            msg = Message(body="Hello World")
            self.sender.send(msg)
            self.n_sent += 1

    def check_if_done(self):
        if self.n_sent == self.received_count:
            self.error = None
            self.timer.cancel()
            self.conn1.close()
            self.conn2.close()

    def on_message(self, event):
        self.received_count+=1
        self.check_if_done()

    def run(self):
        Container(self).run()


class MessageAnnotationsTest(MessagingHandler):
    def __init__(self, address1, address2):
        super(MessageAnnotationsTest, self).__init__()
        self.address1 = address1
        self.address2 = address2
        self.dest = "ma/1"
        self.error = "Message annotations not found"
        self.timer = None
        self.conn1 = None
        self.conn2 = None
        self.sender = None
        self.receiver = None
        self.sent_count = 0
        self.msg_not_sent = True

    def timeout(self):
        self.error = "Timeout Expired: " + self.error
        self.conn1.close()
        self.conn2.close()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn2 = event.container.connect(self.address2)
        self.receiver = event.container.create_receiver(self.conn2, self.dest)

    def on_link_opened(self, event):
        if event.receiver == self.receiver:
            self.conn1 = event.container.connect(self.address1)
            self.sender = event.container.create_sender(self.conn1, self.dest)

    def on_sendable(self, event):
        if self.msg_not_sent:
            msg = Message(body={'number': 0})
            event.sender.send(msg)
            self.msg_not_sent = False

    def on_message(self, event):
        if event.receiver == self.receiver:
            if 0 == event.message.body['number']:
                ma = event.message.annotations
                if ma['x-opt-qd.ingress'] == '0/QDR.A' and ma['x-opt-qd.trace'] == ['0/QDR.A', '0/QDR.B']:
                    self.error = None
            self.timer.cancel()
            self.conn1.close()
            self.conn2.close()

    def run(self):
        Container(self).run()


class MessageAnnotationsStripTest(MessagingHandler):
    def __init__(self, address1, address2):
        super(MessageAnnotationsStripTest, self).__init__()
        self.address1 = address1
        self.address2 = address2
        self.dest = "message_annotations_strip_no/1"
        self.error = "Message annotations not found"
        self.timer = None
        self.conn1 = None
        self.conn2 = None
        self.sender = None
        self.receiver = None
        self.sent_count = 0
        self.msg_not_sent = True

    def timeout(self):
        self.error = "Timeout Expired: " + self.error
        self.conn1.close()
        self.conn2.close()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn2 = event.container.connect(self.address2)
        self.receiver = event.container.create_receiver(self.conn2, self.dest)

    def on_link_opened(self, event):
        if event.receiver == self.receiver:
            self.conn1 = event.container.connect(self.address1)
            self.sender = event.container.create_sender(self.conn1, self.dest)

    def on_sendable(self, event):
        if event.sender == self.sender:
            if self.msg_not_sent:
                msg = Message(body={'number': 0})
                ingress_message_annotations = {'work': 'hard', 'stay': 'humble'}
                msg.annotations = ingress_message_annotations
                event.sender.send(msg)
                self.msg_not_sent = False

    def on_message(self, event):
        if event.receiver == self.receiver:
            if 0 == event.message.body['number']:
                ma = event.message.annotations
                if ma['x-opt-qd.ingress'] == '0/QDR.A' and ma['x-opt-qd.trace'] == ['0/QDR.A', '0/QDR.B'] \
                        and ma['work'] == 'hard' and ma['stay'] == 'humble':
                    self.error = None
            self.accept(event.delivery)
            self.timer.cancel()
            self.conn1.close()
            self.conn2.close()

    def run(self):
        Container(self).run()


class ManagementTest(MessagingHandler):
    def __init__(self, address):
        super(ManagementTest, self).__init__()
        self.address = address
        self.timer = None
        self.conn = None
        self.sender = None
        self.receiver = None
        self.sent_count = 0
        self.msg_not_sent = True
        self.error = None
        self.response1 = False
        self.response2 = False

    def timeout(self):
        if not self.response1:
            self.error = "Incorrect response received for message with correlation id C1"
        if not self.response1:
            self.error = self.error + "Incorrect response received for message with correlation id C2"

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn = event.container.connect(self.address)
        self.sender = event.container.create_sender(self.conn)
        self.receiver = event.container.create_receiver(self.conn, None, dynamic=True)

    def on_link_opened(self, event):
        if event.receiver == self.receiver:
            request = Message()
            request.correlation_id = "C1"
            request.address = "amqp:/_local/$management"
            request.properties = {u'type': u'org.amqp.management', u'name': u'self', u'operation': u'GET-MGMT-NODES'}
            request.reply_to = self.receiver.remote_source.address
            self.sender.send(request)

            request = Message()
            request.address = "amqp:/_topo/0/QDR.B/$management"
            request.correlation_id = "C2"
            request.reply_to = self.receiver.remote_source.address
            request.properties = {u'type': u'org.amqp.management', u'name': u'self', u'operation': u'GET-MGMT-NODES'}
            self.sender.send(request)

    def on_message(self, event):
        if event.receiver == self.receiver:
            if event.message.correlation_id == "C1":
                if event.message.properties['statusCode'] == 200 and \
                        event.message.properties['statusDescription'] is not None \
                        and 'amqp:/_topo/0/QDR.B/$management' in event.message.body:
                    self.response1 = True
            elif event.message.correlation_id == "C2":
                if event.message.properties['statusCode'] == 200 and \
                        event.message.properties['statusDescription'] is not None \
                        and 'amqp:/_topo/0/QDR.A/$management' in event.message.body:
                    self.response2 = True

        if self.response1 and self.response2:
            self.error = None

        if self.error is None:
            self.timer.cancel()
            self.conn.close()

    def run(self):
        Container(self).run()


class MessageAnnotationStripMessageAnnotationsIn(MessagingHandler):
    def __init__(self, address1, address2):
        super(MessageAnnotationStripMessageAnnotationsIn, self).__init__()
        self.address1 = address1
        self.address2 = address2
        self.dest = "strip_message_annotations_in/1"
        self.error = "Message annotations not found"
        self.timer = None
        self.conn1 = None
        self.conn2 = None
        self.sender = None
        self.receiver = None
        self.sent_count = 0
        self.msg_not_sent = True

    def timeout(self):
        self.error = "Timeout Expired: " + self.error
        self.conn1.close()
        self.conn2.close()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn2 = event.container.connect(self.address2)
        self.receiver = event.container.create_receiver(self.conn2, self.dest)

    def on_link_opened(self, event):
        if event.receiver == self.receiver:
            self.conn1 = event.container.connect(self.address1)
            self.sender = event.container.create_sender(self.conn1, self.dest)

    def on_sendable(self, event):
        if event.sender == self.sender:
            if self.msg_not_sent:
                msg = Message(body={'number': 0})
                #
                # Pre-existing ingress and trace
                #
                ingress_message_annotations = {'x-opt-qd.ingress': 'ingress-router', 'x-opt-qd.trace': ['X/QDR']}
                msg.annotations = ingress_message_annotations
                event.sender.send(msg)
                self.msg_not_sent = False

    def on_message(self, event):
        if event.receiver == self.receiver:
            if 0 == event.message.body['number']:
                if event.message.annotations['x-opt-qd.ingress'] == '0/QDR.A' \
                        and event.message.annotations['x-opt-qd.trace'] == ['0/QDR.A', '0/QDR.B']:
                    self.error = None
            self.timer.cancel()
            self.conn1.close()
            self.conn2.close()

    def run(self):
        Container(self).run()


class MessageAnnotaionsPreExistingOverride(MessagingHandler):
    def __init__(self, address1, address2):
        super(MessageAnnotaionsPreExistingOverride, self).__init__()
        self.address1 = address1
        self.address2 = address2
        self.dest = "toov/1"
        self.error = "Pre-existing x-opt-qd.to has been stripped"
        self.timer = None
        self.conn1 = None
        self.conn2 = None
        self.sender = None
        self.receiver = None
        self.msg_not_sent = True

    def timeout(self):
        self.error = "Timeout Expired: " + self.error
        self.conn1.close()
        self.conn2.close()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn2 = event.container.connect(self.address2)
        self.receiver = event.container.create_receiver(self.conn2, self.dest)

    def on_link_opened(self, event):
        if event.receiver == self.receiver:
            self.conn1 = event.container.connect(self.address1)
            self.sender = event.container.create_sender(self.conn1, self.dest)

    def on_sendable(self, event):
        if event.sender == self.sender:
            if self.msg_not_sent:
                msg = Message(body={'number': 0})
                msg.annotations = {'x-opt-qd.to': 'toov/1'}
                event.sender.send(msg)
                self.msg_not_sent = False

    def on_message(self, event):
        if event.receiver == self.receiver:
            if 0 == event.message.body['number']:
                ma = event.message.annotations
                if ma['x-opt-qd.to'] == 'toov/1':
                    self.error = None
            self.timer.cancel()
            self.conn1.close()
            self.conn2.close()

    def run(self):
        Container(self).run()


class MessageAnnotationsStripMessageAnnotationsOut(MessagingHandler):
    def __init__(self, address1, address2):
        super(MessageAnnotationsStripMessageAnnotationsOut, self).__init__()
        self.address1 = address1
        self.address2 = address2
        self.dest = "strip_message_annotations_out/1"
        self.error = "Message annotations not found"
        self.timer = None
        self.conn1 = None
        self.conn2 = None
        self.sender = None
        self.receiver = None
        self.sent_count = 0
        self.msg_not_sent = True

    def timeout(self):
        self.error = "Timeout Expired: " + self.error
        self.conn1.close()
        self.conn2.close()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn2 = event.container.connect(self.address2)
        self.receiver = event.container.create_receiver(self.conn2, self.dest)

    def on_link_opened(self, event):
        if event.receiver == self.receiver:
            self.conn1 = event.container.connect(self.address1)
            self.sender = event.container.create_sender(self.conn1, self.dest)

    def on_sendable(self, event):
        if event.sender == self.sender:
            if self.msg_not_sent:
                msg = Message(body={'number': 0})
                event.sender.send(msg)
                self.msg_not_sent = False

    def on_message(self, event):
        if event.receiver == self.receiver:
            if 0 == event.message.body['number']:
                if event.message.annotations is None:
                    self.error = None
            self.timer.cancel()
            self.conn1.close()
            self.conn2.close()

    def run(self):
        Container(self).run()


class MessageAnnotationsStripBothAddIngressTrace(MessagingHandler):
    def __init__(self, address1, address2):
        super(MessageAnnotationsStripBothAddIngressTrace, self).__init__()
        self.address1 = address1
        self.address2 = address2
        self.dest = "strip_message_annotations_both_add_ingress_trace/1"
        self.error = "Message annotations not found"
        self.timer = None
        self.conn1 = None
        self.conn2 = None
        self.sender = None
        self.receiver = None
        self.sent_count = 0
        self.msg_not_sent = True

    def timeout(self):
        self.error = "Timeout Expired: " + self.error
        self.conn1.close()
        self.conn2.close()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn2 = event.container.connect(self.address2)
        self.receiver = event.container.create_receiver(self.conn2, self.dest)

    def on_link_opened(self, event):
        if event.receiver == self.receiver:
            self.conn1 = event.container.connect(self.address1)
            self.sender = event.container.create_sender(self.conn1, self.dest)

    def on_sendable(self, event):
        if event.sender == self.sender:
            if self.msg_not_sent:
                msg = Message(body={'number': 0})
                ingress_message_annotations = {'work': 'hard',
                                               'x-opt-qd': 'humble',
                                               'x-opt-qd.ingress': 'ingress-router',
                                               'x-opt-qd.trace': ['0/QDR.A']}
                msg.annotations = ingress_message_annotations
                event.sender.send(msg)
                self.msg_not_sent = False

    def on_message(self, event):
        if self.receiver == event.receiver:
            if 0 == event.message.body['number']:
                if event.message.annotations == {'work': 'hard', 'x-opt-qd': 'humble'}:
                    self.error = None
            self.timer.cancel()
            self.conn1.close()
            self.conn2.close()

    def run(self):
        Container(self).run()


class MessageAnnotationsStripAddTraceTest(MessagingHandler):
    def __init__(self, address1, address2):
        super(MessageAnnotationsStripAddTraceTest, self).__init__()
        self.address1 = address1
        self.address2 = address2
        self.dest = "message_annotations_strip_no/1"
        self.error = "Message annotations not found"
        self.timer = None
        self.conn1 = None
        self.conn2 = None
        self.sender = None
        self.receiver = None
        self.sent_count = 0
        self.msg_not_sent = True
        self.logger = Logger(title="MessageAnnotationsStripAddTraceTest")

    def timeout(self):
        self.error = "Timeout Expired: " + self.error
        self.conn1.close()
        self.conn2.close()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn2 = event.container.connect(self.address2)
        self.receiver = event.container.create_receiver(self.conn2, self.dest)
        self.logger.log("on_start(): Receiver link created on self.conn2")

    def on_sendable(self, event):
        if event.sender == self.sender:
            if self.msg_not_sent:
                msg = Message(body={'number': 0})
                ingress_message_annotations = {'x-opt-qd.trace': ['0/QDR.1']}
                msg.annotations = ingress_message_annotations
                event.sender.send(msg)
                self.msg_not_sent = False
                self.logger.log("on_sendable(): Message sent")

    def on_link_opened(self, event):
        if event.receiver == self.receiver:
            self.conn1 = event.container.connect(self.address1)
            self.sender = event.container.create_sender(self.conn1, self.dest)
            self.logger.log("on_link_opened(): Sender link created on self.conn1")

    def on_message(self, event):
        if event.receiver == self.receiver:
            self.logger.log("on_message(): Message received by receiver")
            if 0 == event.message.body['number']:
                self.logger.log("on_message(): Message received by receiver body matches expected body")
                ma = event.message.annotations
                if ma['x-opt-qd.ingress'] == '0/QDR.A' and ma['x-opt-qd.trace'] == ['0/QDR.1', '0/QDR.A', '0/QDR.B']:
                    self.logger.log("on_message(): Message annotations in message match expected message annotations. Success...")
                    self.error = None
                    self.timer.cancel()
                    self.conn1.close()
                    self.conn2.close()
                else:
                    self.logger.log("on_message(): Message received by receiver,  message annotations are not as expected")
                    self.logger.log(ma)
            else:
                self.logger.log("on_message(): Message received by receiver but body is not expected")

    def run(self):
        Container(self).run()


class SenderSettlesFirst(MessagingHandler):
    def __init__(self, address1, address2):
        super(SenderSettlesFirst, self).__init__(auto_accept=False)
        self.address1 = address1
        self.address2 = address2
        self.dest = "closest.senderfirst.1"
        self.error = "Message body received differs from the one sent"
        self.n_sent = 0
        self.timer = None
        self.conn1 = None
        self.conn2 = None
        self.sender = None
        self.sent_count = 0
        self.received_count = 0
        self.receiver = None
        self.msg_not_sent = True

    def timeout(self):
        self.error = "Timeout Expired: " + self.error
        self.conn1.close()
        self.conn2.close()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn2 = event.container.connect(self.address2)
        self.receiver = event.container.create_receiver(self.conn2, self.dest)

    def on_link_opened(self, event):
        if event.receiver == self.receiver:
            self.conn1 = event.container.connect(self.address1)
            self.sender = event.container.create_sender(self.conn1, self.dest)

    def on_sendable(self, event):
        if event.sender == self.sender:
            if self.msg_not_sent:
                msg = Message(body={'number': 0})
                dlv = event.sender.send(msg)
                dlv.settle()
                self.msg_not_sent = False

    def on_message(self, event):
        if event.receiver == self.receiver:
            if 0 == event.message.body['number']:
                self.error = None
            self.accept(event.delivery)
            self.timer.cancel()
            self.conn1.close()
            self.conn2.close()

    def run(self):
        Container(self).run()


class MulticastUnsettled(MessagingHandler):
    def __init__(self, address):
        super(MulticastUnsettled, self).__init__()
        self.address = address
        self.dest = "multicast.2"
        self.error = None
        self.n_sent = 0
        self.count = 3
        self.n_received_a = 0
        self.n_received_b = 0
        self.n_received_c = 0
        self.timer = None
        self.conn = None
        self.sender = None
        self.receiver_a = None
        self.receiver_b = None
        self.receiver_c = None

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn = event.container.connect(self.address)
        self.sender = event.container.create_sender(self.conn, self.dest)
        self.receiver_a = event.container.create_receiver(self.conn, self.dest, name="A")
        self.receiver_b = event.container.create_receiver(self.conn, self.dest, name="B")
        self.receiver_c = event.container.create_receiver(self.conn, self.dest, name="C")

    def timeout(self):
        self.error = "Timeout Expired: sent=%d rcvd=%d/%d/%d" % \
                     (self.n_sent, self.n_received_a, self.n_received_b, self.n_received_c)
        self.conn.close()

    def check_if_done(self):
        if self.n_received_a + self.n_received_b + self.n_received_c == self.count:
            self.timer.cancel()
            self.conn.close()

    def on_sendable(self, event):
        if self.n_sent == 0:
            msg = Message(body="MulticastUnsettled-Test")
            self.sender.send(msg)
            self.n_sent += 1

    def on_message(self, event):
        if event.receiver == self.receiver_a:
            self.n_received_a += 1
        if event.receiver == self.receiver_b:
            self.n_received_b += 1
        if event.receiver == self.receiver_c:
            self.n_received_c += 1

    def on_accepted(self, event):
        self.check_if_done()

    def run(self):
        Container(self).run()


class SemanticsClosestIsLocal(MessagingHandler):
    def __init__(self, address1, address2):
        super(SemanticsClosestIsLocal, self).__init__()
        self.address1 = address1
        self.address2 = address2
        self.dest = "closest.1"
        self.timer = None
        self.conn1 = None
        self.conn2 = None
        self.sender = None
        self.receiver_a = None
        self.receiver_b = None
        self.receiver_c = None
        self.num_messages = 100
        self.n_received_a = 0
        self.n_received_b = 0
        self.n_received_c = 0
        self.error = None
        self.n_sent = 0

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn1 = event.container.connect(self.address1)
        self.conn2 = event.container.connect(self.address2)
        self.sender = event.container.create_sender(self.conn1, self.dest)
        # Receiver on same router as the sender must receive all the messages. The other two
        # receivers are on the other router
        self.receiver_a = event.container.create_receiver(self.conn1, self.dest, name="A")
        self.receiver_b = event.container.create_receiver(self.conn2, self.dest, name="B")
        self.receiver_c = event.container.create_receiver(self.conn2, self.dest, name="C")

    def timeout(self):
        self.error = "Timeout Expired: sent=%d rcvd=%d/%d/%d" % \
                     (self.n_sent, self.n_received_a, self.n_received_b, self.n_received_c)
        self.conn1.close()
        self.conn2.close()

    def check_if_done(self):
        if self.n_received_a == 100 and self.n_received_b + self.n_received_c == 0:
            self.timer.cancel()
            self.conn1.close()
            self.conn2.close()

    def on_sendable(self, event):
        if self.n_sent < self.num_messages:
            msg = Message(body="SemanticsClosestIsLocal-Test")
            self.sender.send(msg)
            self.n_sent += 1

    def on_message(self, event):
        if event.receiver == self.receiver_a:
            self.n_received_a += 1
        if event.receiver == self.receiver_b:
            self.n_received_b += 1
        if event.receiver == self.receiver_c:
            self.n_received_c += 1

    def on_accepted(self, event):
        self.check_if_done()

    def run(self):
        Container(self).run()


class SemanticsClosestIsRemote(MessagingHandler):
    def __init__(self, address1, address2):
        super(SemanticsClosestIsRemote, self).__init__()
        self.address1 = address1
        self.address2 = address2
        self.dest = "closest.1"
        self.timer = None
        self.conn1 = None
        self.conn2 = None
        self.sender = None
        self.receiver_a = None
        self.receiver_b = None
        self.receiver_c = None
        self.num_messages = 100
        self.n_received_a = 0
        self.n_received_b = 0
        self.error = None
        self.n_sent = 0

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn1 = event.container.connect(self.address1)
        self.conn2 = event.container.connect(self.address2)
        self.sender = event.container.create_sender(self.conn1, self.dest)
        # Receiver on same router as the sender must receive all the messages. The other two
        # receivers are on the other router
        self.receiver_a = event.container.create_receiver(self.conn2, self.dest, name="A")
        self.receiver_b = event.container.create_receiver(self.conn2, self.dest, name="B")

    def timeout(self):
        self.error = "Timeout Expired: sent=%d rcvd=%d/%d" % \
                     (self.n_sent, self.n_received_a, self.n_received_b)
        self.conn1.close()
        self.conn2.close()

    def check_if_done(self):
        if self.n_received_a + self.n_received_b == 100 and self.n_received_a > 0 and self.n_received_b > 0:
            self.timer.cancel()
            self.conn1.close()
            self.conn2.close()

    def on_sendable(self, event):
        if self.n_sent < self.num_messages:
            msg = Message(body="SemanticsClosestIsRemote-Test")
            self.sender.send(msg)
            self.n_sent += 1

    def on_message(self, event):
        if event.receiver == self.receiver_a:
            self.n_received_a += 1
        if event.receiver == self.receiver_b:
            self.n_received_b += 1

    def on_accepted(self, event):
        self.check_if_done()

    def run(self):
        Container(self).run()


class CustomTimeout(object):
    def __init__(self, parent):
        self.parent = parent

    def addr_text(self, addr):
        if not addr:
            return ""
        if addr[0] == 'M':
            return addr[2:]
        else:
            return addr[1:]

    def on_timer_task(self, event):
        local_node = Node.connect(self.parent.address1, timeout=TIMEOUT)

        res = local_node.query('org.apache.qpid.dispatch.router.address')
        name = res.attribute_names.index('name')
        found = False
        for results in res.results:
            if "balanced.1" == self.addr_text(results[name]):
                found = True
                break

        if found:
            self.parent.cancel_custom()
            self.parent.create_sender(event)

        else:
            event.reactor.schedule(2, self)


class SemanticsBalanced(MessagingHandler):
    def __init__(self, address1, address2, address3):
        super(SemanticsBalanced, self).__init__(auto_accept=False, prefetch=0)
        self.address1 = address1
        self.address2 = address2
        self.address3 = address3
        self.dest = "balanced.1"
        self.timer = None
        self.conn1 = None
        self.conn2 = None
        self.conn3 = None
        self.sender = None
        self.receiver_a = None
        self.receiver_b = None
        self.receiver_c = None
        self.num_messages = 400
        self.n_received_a = 0
        self.n_received_b = 0
        self.n_received_c = 0
        self.error = None
        self.n_sent = 0
        self.rx_set = []
        self.custom_timer = None

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.custom_timer = event.reactor.schedule(2, CustomTimeout(self))
        self.conn1 = event.container.connect(self.address1)
        self.conn2 = event.container.connect(self.address2)
        self.conn3 = event.container.connect(self.address3)

        # This receiver is on the same router as the sender
        self.receiver_a = event.container.create_receiver(self.conn2, self.dest, name="A")

        # These two receivers are connected to a different router than the sender
        self.receiver_b = event.container.create_receiver(self.conn3, self.dest, name="B")
        self.receiver_c = event.container.create_receiver(self.conn3, self.dest, name="C")

        self.receiver_a.flow(300)
        self.receiver_b.flow(300)
        self.receiver_c.flow(300)

    def cancel_custom(self):
        self.custom_timer.cancel()

    def create_sender(self, event):
        self.sender = event.container.create_sender(self.conn1, self.dest)

    def timeout(self):
        self.error = "Timeout Expired: sent=%d rcvd=%d/%d/%d" % \
                     (self.n_sent, self.n_received_a, self.n_received_b, self.n_received_c)
        self.conn1.close()
        self.conn2.close()
        self.conn3.close()

    def check_if_done(self):
        if self.n_received_a + self.n_received_b + self.n_received_c == self.num_messages and \
                self.n_received_a > 0 and self.n_received_b > 0 and self.n_received_c > 0:
            self.rx_set.sort()
            all_messages_received = True
            for i in range(self.num_messages):
                if not i == self.rx_set[i]:
                    all_messages_received = False

            if all_messages_received:
                self.timer.cancel()
                self.conn1.close()
                self.conn2.close()
                self.conn3.close()

    def on_sendable(self, event):
        if self.n_sent < self.num_messages:
            msg = Message(body={'number': self.n_sent})
            self.sender.send(msg)
            self.n_sent += 1

    def on_message(self, event):
        if event.receiver ==    self.receiver_a:
            self.n_received_a += 1
            self.rx_set.append(event.message.body['number'])
        elif event.receiver == self.receiver_b:
            self.n_received_b += 1
            self.rx_set.append(event.message.body['number'])
        elif event.receiver == self.receiver_c:
            self.n_received_c += 1
            self.rx_set.append(event.message.body['number'])

        self.check_if_done()

    def run(self):
        Container(self).run()


class PropagatedDisposition(MessagingHandler):
    """
    Verify outcomes are properly sent end-to-end
    """
    def __init__(self, test, address1, address2):
        super(PropagatedDisposition, self).__init__(auto_accept=False)
        self.address1 = address1
        self.address2 = address2
        self.settled = []
        self.test = test
        self.sender_conn = None
        self.receiver_conn = None
        self.passed = False

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.sender_conn = event.container.connect(self.address1)
        self.receiver_conn = event.container.connect(self.address2)
        addr = "unsettled/2"
        self.sender = event.container.create_sender(self.sender_conn, addr)
        self.receiver = event.container.create_receiver(self.receiver_conn, addr)
        self.receiver.flow(3)
        self.trackers = {}
        for b in ['accept', 'modified', 'reject']:
            self.trackers[self.sender.send(Message(body=b))] = b

    def timeout(self):
        unique_list = sorted(list(dict.fromkeys(self.settled)))
        self.error = "Timeout Expired: Expected ['accept', 'modified', 'reject'] got %s" % unique_list
        self.sender_conn.close()
        self.receiver_conn.close()

    def check(self):
        unique_list = sorted(list(dict.fromkeys(self.settled)))
        if unique_list == [u'accept', u'modified', u'reject']:
            self.passed = True
            self.sender_conn.close()
            self.receiver_conn.close()
            self.timer.cancel()

    def on_message(self, event):
        if event.message.body == u'accept':
            event.delivery.update(Delivery.ACCEPTED)
            event.delivery.settle()
        elif event.message.body == u'reject':
            self.set_rejected_data(event.delivery.local)
            event.delivery.update(Delivery.REJECTED)
            event.delivery.settle()
        elif event.message.body == u'modified':
            self.set_modified_data(event.delivery.local)
            event.delivery.update(Delivery.MODIFIED)
            event.delivery.settle()

    def on_accepted(self, event):
        self.test.assertEqual(Delivery.ACCEPTED, event.delivery.remote_state)
        self.test.assertEqual('accept', self.trackers[event.delivery])
        self.settled.append('accept')
        self.check()

    def on_rejected(self, event):
        self.test.assertEqual(Delivery.REJECTED, event.delivery.remote_state)
        self.test.assertEqual('reject', self.trackers[event.delivery])
        self.check_rejected_data(event.delivery.remote)
        self.settled.append('reject')
        self.check()

    def on_released(self, event):
        # yes, for some reason Proton triggers on_released when MODIFIED is set
        self.test.assertEqual(Delivery.MODIFIED, event.delivery.remote_state)
        self.test.assertEqual('modified', self.trackers[event.delivery])
        self.check_modified_data(event.delivery.remote)
        self.settled.append('modified')
        self.check()

    def set_rejected_data(self, local_state):
        # use defaults
        pass

    def check_rejected_data(self, remote_state):
        self.test.assertTrue(remote_state.condition is None)

    def set_modified_data(self, local_state):
        # use defaults
        pass

    def check_modified_data(self, remote_state):
        self.test.assertTrue(remote_state.failed)
        self.test.assertFalse(remote_state.undeliverable)
        self.test.assertTrue(remote_state.annotations is None)

    def run(self):
        Container(self).run()


class PropagatedDispositionData(PropagatedDisposition):
    """
    Verify that data associated with a terminal outcome is correctly passed end
    to end
    """
    def set_rejected_data(self, local_state):
        local_state.condition = Condition("name",
                                          str("description"),
                                          {symbol("info"): True})

    def check_rejected_data(self, remote_state):
        cond = remote_state.condition
        self.test.assertEqual("name", cond.name)
        self.test.assertEqual("description", cond.description)
        self.test.assertTrue(cond.info is not None)
        self.test.assertTrue(symbol("info") in cond.info)
        self.test.assertEqual(True, cond.info[symbol("info")])

    def set_modified_data(self, local_state):
        local_state.failed = True
        local_state.undeliverable = True
        local_state.annotations = {symbol('modified'): True}

    def check_modified_data(self, remote_state):
        self.test.assertTrue(remote_state.failed)
        self.test.assertTrue(remote_state.undeliverable)
        self.test.assertTrue(remote_state.annotations is not None)
        self.test.assertTrue(symbol('modified') in remote_state.annotations)
        self.test.assertEqual(True, remote_state.annotations[symbol('modified')])


class ThreeAck(MessagingHandler):
    def __init__(self, test, address1, address2):
        super(ThreeAck, self).__init__(auto_accept=False, auto_settle=False)
        self.addrs = [address1, address2]
        self.settled = []
        self.test = test
        self.phase = 0

    def on_start(self, event):
        connections = [event.container.connect(a) for a in self.addrs]
        addr = "three_ack/1"
        self.sender = event.container.create_sender(connections[0], addr)
        self.receiver = event.container.create_receiver(connections[1], addr)
        self.receiver.flow(1)
        self.tracker = self.sender.send(Message('hello'))

    def on_message(self, event):
        self.test.assertEqual(0, self.phase)
        self.phase = 1
        self.test.assertFalse(event.delivery.settled)
        self.test.assertEqual(0, self.tracker.local_state)
        self.test.assertEqual(0, self.tracker.remote_state)
        event.delivery.update(Delivery.ACCEPTED)
        # NOTE: we don't settle yet for 3-ack

    def on_accepted(self, event):
        self.test.assertTrue(event.sender)
        self.test.assertEqual(1, self.phase)
        self.phase = 2
        self.test.assertEqual(Delivery.ACCEPTED, event.delivery.remote_state)
        self.test.assertFalse(event.delivery.settled)
        self.test.assertEqual(0, event.delivery.local_state)
        event.delivery.settle()
        self.test.assertFalse(event.delivery.settled)
        event.connection.close()

    def on_settled(self, event):
        self.test.assertTrue(event.receiver)
        self.test.assertEqual(2, self.phase)
        self.phase = 3
        event.connection.close()

    def run(self):
        Container(self).run()
        self.test.assertEqual(3, self.phase)


class TwoRouterConnection(TestCase):
    def __init__(self, test_method):
        TestCase.__init__(self, test_method)
        self.success = False
        self.timer_delay = 4
        self.max_attempts = 2
        self.attempts = 0
        self.local_node = None

    @classmethod
    def router(cls, name, config):
        config = Qdrouterd.Config(config)

        cls.routers.append(cls.tester.qdrouterd(name, config, wait=True))

    @classmethod
    def setUpClass(cls):
        super(TwoRouterConnection, cls).setUpClass()

        cls.routers = []

        cls.B_normal_port_1 = cls.tester.get_port()
        cls.B_normal_port_2 = cls.tester.get_port()

        TwoRouterConnection.router('A', [
                        ('router', {'mode': 'interior', 'id': 'A'}),
                        ('listener', {'host': '0.0.0.0', 'role': 'normal',
                                      'port': cls.tester.get_port()}),
                        ]
              )

        TwoRouterConnection.router('B',
                    [
                        ('router', {'mode': 'interior', 'id': 'B'}),
                        ('listener', {'host': '0.0.0.0', 'role': 'normal',
                                      'port': cls.B_normal_port_1}),
                        ('listener', {'host': '0.0.0.0', 'role': 'normal',
                                      'port': cls.B_normal_port_2}),

                    ]
               )

    def address(self):
        return self.routers[0].addresses[0]

    def run_qdmanage(self, cmd, input=None, expect=Process.EXIT_OK, address=None):
        p = self.popen(
            ['qdmanage'] + cmd.split(' ') + ['--bus', address or self.address(), '--indent=-1', '--timeout', str(TIMEOUT)],
            stdin=PIPE, stdout=PIPE, stderr=STDOUT, expect=expect,
            universal_newlines=True)
        out = p.communicate(input)[0]
        try:
            p.teardown()
        except Exception as e:
            raise Exception(out if out else str(e))
        return out

    def can_terminate(self):
        if self.attempts == self.max_attempts:
            return True

        if self.success:
            return True

        return False

    def check_connections(self):
        res = self.local_node.query(type='org.apache.qpid.dispatch.connection')
        results = res.results

        # If DISPATCH-1093 was not fixed, there would be an additional
        # connection created and hence the len(results) would be 4

        # Since DISPATCH-1093 is fixed, len(results would be 3 which is what
        # we would expect.

        if len(results) != 3:
            self.schedule_num_connections_test()
        else:
            self.success = True

    def schedule_num_connections_test(self):
        if self.attempts < self.max_attempts:
            if not self.success:
                Timer(self.timer_delay, self.check_connections).start()
                self.attempts += 1

    def test_create_connectors(self):
        self.local_node = Node.connect(self.routers[0].addresses[0],
                                       timeout=TIMEOUT)

        res = self.local_node.query(type='org.apache.qpid.dispatch.connection')
        results = res.results

        self.assertEqual(1, len(results))

        long_type = 'org.apache.qpid.dispatch.connector' ''

        create_command = 'CREATE --type=' + long_type + ' --name=foo' + ' host=0.0.0.0 port=' + str(TwoRouterConnection.B_normal_port_1)

        self.run_qdmanage(create_command)

        create_command = 'CREATE --type=' + long_type + ' --name=bar' + ' host=0.0.0.0 port=' + str(TwoRouterConnection.B_normal_port_2)

        self.run_qdmanage(create_command)

        self.schedule_num_connections_test()

        while not self.can_terminate():
            pass

        self.assertTrue(self.success)

class PropagationTest(TestCase):

    inter_router_port = None

    @classmethod
    def setUpClass(cls):
        """Start a router and a messenger"""
        super(PropagationTest, cls).setUpClass()

        def router(name, extra_config):

            config = [
                ('router', {'mode': 'interior', 'id': 'QDR.%s'%name}),

                ('listener', {'port': cls.tester.get_port()}),

            ] + extra_config

            config = Qdrouterd.Config(config)

            cls.routers.append(cls.tester.qdrouterd(name, config, wait=True))

        cls.routers = []

        inter_router_port = cls.tester.get_port()
        router('A', [('listener', {'role': 'inter-router', 'port': inter_router_port}), ('address', {'prefix': 'multicast', 'distribution': 'multicast'})])
        router('B', [('connector', {'role': 'inter-router', 'port': inter_router_port})])

        cls.routers[0].wait_router_connected('QDR.B')
        cls.routers[1].wait_router_connected('QDR.A')

    def test_propagation_of_locally_undefined_address(self):
        test = MulticastTestClient(self.routers[0].addresses[0], self.routers[1].addresses[0])
        test.run()
        self.assertEqual(None, test.error)
        self.assertEqual(test.received, 2)

class CreateReceiver(MessagingHandler):
    def __init__(self, connection, address):
        super(CreateReceiver, self).__init__()
        self.connection = connection
        self.address = address

    def on_timer_task(self, event):
        event.container.create_receiver(self.connection, self.address)

class DelayedSend(MessagingHandler):
    def __init__(self, connection, address, message):
        super(DelayedSend, self).__init__()
        self.connection = connection
        self.address = address
        self.message = message

    def on_timer_task(self, event):
        event.container.create_sender(self.connection, self.address).send(self.message)

class MulticastTestClient(MessagingHandler):
    def __init__(self, router1, router2):
        super(MulticastTestClient, self).__init__()
        self.routers = [router1, router2]
        self.received = 0
        self.error = None

    def on_start(self, event):
        self.connections = [event.container.connect(r) for r in self.routers]
        event.container.create_receiver(self.connections[0], "multicast")
        # wait for knowledge of receiver1 to propagate to second router
        event.container.schedule(5, CreateReceiver(self.connections[1], "multicast"))
        event.container.schedule(7, DelayedSend(self.connections[1], "multicast", Message(body="testing1,2,3")))
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))

    def on_message(self, event):
        self.received += 1
        event.connection.close()
        if self.received == 2:
            self.timer.cancel()

    def timeout(self):
        self.error = "Timeout Expired:received=%d" % self.received
        for c in self.connections:
            c.close()

    def run(self):
        Container(self).run()


class StreamingLinkScrubberTest(TestCase):
    """
    Verify that unused inter-router streaming links are eventually reclaimed
    """

    @classmethod
    def setUpClass(cls):
        super(StreamingLinkScrubberTest, cls).setUpClass()

        def router(name, extra):
            config = [
                ('router', {'id': 'Router%s' % name,
                            'mode': 'interior'}),
                ('listener', {'port': cls.tester.get_port(),
                              'stripAnnotations': 'no'}),
                ('address', {'prefix': 'closest', 'distribution': 'closest'}),
                ('address', {'prefix': 'balanced', 'distribution': 'balanced'}),
                ('address', {'prefix': 'multicast', 'distribution': 'multicast'})

            ]

            if extra:
                config.extend(extra)

            config = Qdrouterd.Config(config)

            # run routers in test mode to shorten the streaming link scrubber
            # interval to 5 seconds an the maximum pool size to two links
            cls.routers.append(cls.tester.qdrouterd(name, config, wait=True, cl_args=["--test-hooks"]))

        cls.routers = []

        inter_router_port = cls.tester.get_port()

        router('A',
               [('listener', {'role': 'inter-router',
                              'port': inter_router_port})])
        cls.RouterA = cls.routers[-1]
        cls.RouterA.listener = cls.RouterA.addresses[0]

        router('B',
               [('connector', {'name': 'connectorToA', 'role':
                               'inter-router',
                               'port': inter_router_port})])
        cls.RouterB = cls.routers[-1]
        cls.RouterB.listener = cls.RouterB.addresses[0]

        cls.RouterA.wait_router_connected('RouterB')
        cls.RouterB.wait_router_connected('RouterA')

    def test_01_streaming_link_scrubber(self):
        """
        Ensure extra streaming links are closed by the periodic scrubber
        """
        address = "closest/scrubber"

        # scrubber removes at most 10 links per scan, the test pool size is 2
        sender_count = 12

        # fire up a receiver on RouterB to get 1 message from each sender:
        env = dict(os.environ, PN_TRACE_FRM="1")
        cmd = ["test-receiver",
               "-a", self.RouterB.listener,
               "-s", address,
               "-c", str(sender_count)]
        rx = self.popen(cmd, env=env)

        self.RouterA.wait_address(address)

        # remember the count of inter-router links on A before we start streaming
        pre_count = len(get_inter_router_links(self.RouterA.listener))

        # fire off the senders
        cmd = ["test-sender",
               "-a", self.RouterA.listener,
               "-t", address,
               "-c", "1",
               "-sx"
        ]
        senders = [self.popen(cmd, env=env) for x in range(sender_count)]

        for tx in senders:
            out_text, out_error = tx.communicate(timeout=TIMEOUT)
            if tx.returncode:
                raise Exception("Sender failed: %s %s" % (out_text, out_error))

        # expect: more inter-router links opened.  Should be 12 more, but
        # depending on when the scrubber runs it may be as low as two
        post_count = len(get_inter_router_links(self.RouterA.listener))
        self.assertTrue(post_count > pre_count)

        # expect: after 5 seconds 10 of the links should be closed and 2
        # should remain (--test-hooks router option sets these parameters)
        while (post_count - pre_count) > 2:
            sleep(0.1)
            post_count = len(get_inter_router_links(self.RouterA.listener))

        out_text, out_error = rx.communicate(timeout=TIMEOUT)
        if rx.returncode:
            raise Exception("Receiver failed: %s %s" % (out_text, out_error))


class TwoRouterExtensionStateTest(TestCase):
    """
    Verify that routers propagate extended Disposition state correctly.
    See DISPATCH-1703
    """

    @classmethod
    def setUpClass(cls):
        super(TwoRouterExtensionStateTest, cls).setUpClass()

        def router(name, extra_config):

            config = [
                ('router', {'mode': 'interior',
                            'id': name}),

                ('listener', {'port': cls.tester.get_port() }),

                ('address', {'prefix': 'closest', 'distribution': 'closest'}),
                ('address', {'prefix': 'balanced', 'distribution': 'balanced'}),
                ('address', {'prefix': 'multicast', 'distribution': 'multicast'}),
            ] + extra_config

            config = Qdrouterd.Config(config)
            return cls.tester.qdrouterd(name, config, wait=False)

        inter_router_port = cls.tester.get_port()
        service_port = cls.tester.get_port()

        cls.RouterA = router('RouterA',
                             [
                                 ('listener', {'role': 'inter-router',
                                               'host': '0.0.0.0',
                                               'port': inter_router_port,
                                               'saslMechanisms': 'ANONYMOUS'}),
                             ])

        cls.RouterB = router('RouterB',
                             [
                                 ('connector', {'name': 'toRouterA',
                                                'role': 'inter-router',
                                                'port': inter_router_port}),


                                 ('listener', {'role': 'route-container',
                                               'host': '0.0.0.0',
                                               'port': service_port,
                                               'saslMechanisms': 'ANONYMOUS'}),

                                 ('linkRoute', {'prefix': 'RoutieMcRouteFace',
                                                'containerId': 'FakeService',
                                                'direction': 'in'}),
                                 ('linkRoute', {'prefix': 'RoutieMcRouteFace',
                                                'containerId': 'FakeService',
                                                'direction': 'out'}),
                             ])


        cls.RouterA.wait_router_connected('RouterB')
        cls.RouterB.wait_router_connected('RouterA')

    def test_01_link_route(self):
        """
        Verify non-terminal state and data propagates over a link route
        """
        class MyExtendedService(FakeService):
            """
            This service saves any outcome and extension data that arrives in a
            transfer
            """
            def __init__(self, url, container_id=None):
                self.remote_state = None
                self.remote_data = None
                super(MyExtendedService, self).__init__(url, container_id)

            def on_message(self, event):
                self.remote_state = event.delivery.remote_state;
                self.remote_data = event.delivery.remote.data;
                super(MyExtendedService, self).on_message(event)


        fs = MyExtendedService(self.RouterB.addresses[1],
                               container_id="FakeService")
        self.RouterA.wait_address("RoutieMcRouteFace", remotes=1, count=2)

        tx = MyExtendedSender(self.RouterA.addresses[0],
                              "RoutieMcRouteFace")
        tx.wait()
        fs.join()
        self.assertEqual(999, fs.remote_state)
        self.assertEqual([1, 2, 3], fs.remote_data)

    def test_02_closest(self):
        """
        Verify non-terminal state and data propagates over anycase
        """
        test = ExtensionStateTester(self.RouterA.addresses[0],
                                    self.RouterB.addresses[0],
                                    "closest/fleabag")
        test.run()
        self.assertEqual(None, test.error)

    def test_03_multicast(self):
        """
        Verify that disposition state set by the publisher is available to all
        consumers
        """
        rxs = [MyExtendedReceiver(self.RouterA.addresses[0],
                                  "multicast/thingy")
               for x in range(3)]
        self.RouterA.wait_address("multicast/thingy", subscribers=3)
        sleep(0.5)  # let subscribers grant credit
        tx = MyExtendedSender(self.RouterB.addresses[0],
                              "multicast/thingy")
        tx.wait()

        # DISPATCH-1705: only one of the receivers gets the data, but all
        # should get the state

        ext_data = None
        for rx in rxs:
            rx.stop()
            try:
                while True:
                    dispo = rx.remote_states.pop()
                    self.assertEqual(999, dispo[0])
                    ext_data = dispo[1] or ext_data
            except IndexError:
                pass
        self.assertEqual([1, 2, 3], ext_data)


class MyExtendedSender(AsyncTestSender):
    """
    This sender sets a non-terminal outcome and data on the outgoing
    transfer
    """
    def on_sendable(self, event):
        if self.sent < self.total:
            dlv = event.sender.delivery(str(self.sent))
            dlv.local.data = [1, 2, 3]
            dlv.update(999)
            event.sender.stream(self._message.encode())
            event.sender.advance()
            self.sent += 1


class MyExtendedReceiver(AsyncTestReceiver):
    """
    This receiver stores any remote delivery state that arrives with a message
    transfer
    """
    def __init__(self, *args, **kwargs):
        self.remote_states = []
        super(MyExtendedReceiver, self).__init__(*args, **kwargs)

    def on_message(self, event):
        self.remote_states.append((event.delivery.remote_state,
                                   event.delivery.remote.data))
        super(MyExtendedReceiver, self).on_message(event)


class ExtensionStateTester(MessagingHandler):
    """
    Verify the routers propagate non-terminal outcome and extended state
    disposition information in both message transfer and disposition frames.

    This tester creates a receiver and a sender link to a given address.

    The sender transfers a message with a non-terminal delivery state and
    associated extension data.  The receiver expects to find this state in the
    incoming delivery.

    The receiver then responds with a non-terminal disposition that also has
    extension state data.  The sender expects to find this new state associated
    with its delivery.
    """
    def __init__(self, ingress_router, egress_router, address):
        super(ExtensionStateTester, self).__init__(auto_settle=False,
                                                   auto_accept=False)
        self._in_router = ingress_router
        self._out_router = egress_router
        self._address = address
        self._sender_conn = None
        self._recvr_conn = None
        self._sender = None
        self._receiver = None
        self._sent = 0
        self._received = 0
        self._settled = 0
        self._total = 10
        self._message = Message(body="XYZ" * (1024 * 1024 * 2))
        self.error = None

    def on_start(self, event):
        self._reactor = event.reactor
        self._sender_conn = event.container.connect(self._in_router)
        self._sender = event.container.create_sender(self._sender_conn,
                                                     target=self._address,
                                                     name="ExtensionSender")
        self._recvr_conn = event.container.connect(self._out_router)
        self._receiver = event.container.create_receiver(self._recvr_conn,
                                                         source=self._address,
                                                         name="ExtensionReceiver")
    def _done(self, error=None):
        self.error = error or self.error
        self._sender.close()
        self._sender_conn.close()
        self._receiver.close()
        self._recvr_conn.close()

    def on_sendable(self, event):
        if self._sent < self._total:
            self._sent += 1
            dlv = event.sender.delivery(str(self._sent))
            dlv.local.data = [1, 2, 3, self._sent]
            dlv.update(666)  # non-terminal state
            self._message.id = self._sent
            event.sender.stream(self._message.encode())
            event.sender.advance()

    def on_message(self, event):
        dlv = event.delivery
        msg_id = event.message.id
        if dlv.remote_state != 666:
            return self._done(error="Unexpected outcome '%s', expected '666'"
                              % dlv.remote_state)
        remote_data = dlv.remote.data
        expected_data = [1, 2, 3, msg_id]
        if remote_data != expected_data:
            return self._done(error="Unexpected dispo data '%s', expected '%s'"
                              % (remote_data, expected_data))

        # send back a non-terminal outcome and more data
        dlv.local.data = [10, 9, 8, msg_id]
        dlv.update(777)
        self._received += 1

    def _handle_sender_update(self, event):
        dlv = event.delivery
        if dlv.local_state != 666 or len(dlv.local.data) != 4:
            return self._done(error="Unexpected local state at sender: %s %s" %
                              (dlv.local_state, dlv.local.data))

        if dlv.remote_state != 777 or len(dlv.remote.data) != 4:
            return self._done(error="Unexpected remote state at sender: %s %s" %
                              (dlv.remote_state, dlv.remote.data))
        dlv.settle()

    def _handle_receiver_update(self, event):
        dlv = event.delivery
        if dlv.settled:
            if dlv.local_state != 777 or len(dlv.local.data) != 4:
                return self._done(error="Unexpected local state at sender: %s %s" %
                                  (dlv.local_state, dlv.local.data))

            if dlv.remote_state != 666 or len(dlv.remote.data) != 4:
                return self._done(error="Unexpected remote state at sender: %s %s" %
                                  (dlv.remote_state, dlv.remote.data))
            dlv.settle()
            self._settled += 1
            if self._settled == self._total:
                self._done()

    def on_delivery(self, event):
        if event.delivery.link.is_sender:
            self._handle_sender_update(event)
        else:
            self._handle_receiver_update(event)

    def run(self):
        Container(self).run()


if __name__ == '__main__':
    unittest.main(main_module())