@Test(timeout = 1000*60*60)
public void testPartitionByQueue() throws Exception {

    partitioning.byQueue = new HashMap<String, Target>();
    partitioning.byQueue.put("foo", new Target("broker1"));
    createBrokerCluster(2);
    Connection connection2 = createConnectionTo("broker2");
    within(5, TimeUnit.SECONDS, new Task() {
        public void run() throws Exception {
            assertEquals(0, getTransportConnector("broker1").getConnections().size());
            assertEquals(1, getTransportConnector("broker2").getConnections().size());
        }
    });
    Session session2 = connection2.createSession(false, Session.AUTO_ACKNOWLEDGE);
    MessageConsumer consumer = session2.createConsumer(session2.createQueue("foo"));
    within(5, TimeUnit.SECONDS, new Task() {
        public void run() throws Exception {
            assertEquals(1, getTransportConnector("broker1").getConnections().size());
            assertEquals(0, getTransportConnector("broker2").getConnections().size());
        }
    });
    Connection connection1 = createConnectionTo("broker2");
    Session session1 = connection1.createSession(false, Session.AUTO_ACKNOWLEDGE);
    MessageProducer producer = session1.createProducer(session1.createQueue("foo"));
    within(5, TimeUnit.SECONDS, new Task() {
        public void run() throws Exception {
            assertEquals(1, getTransportConnector("broker1").getConnections().size());
            assertEquals(1, getTransportConnector("broker2").getConnections().size());
        }
    });
    for (int i = 0; i < 100; i++) {
        producer.send(session1.createTextMessage("#" + i));
    }
    within(5, TimeUnit.SECONDS, new Task() {
        public void run() throws Exception {
            assertEquals(2, getTransportConnector("broker1").getConnections().size());
            assertEquals(0, getTransportConnector("broker2").getConnections().size());
        }
    });
}