queues[6] = new LocalClusteredQueue(office1, 1, "sub7", im.getId(), ms, pm, true, true, (QueuedExecutor)pool.get(), null, tr);
queues[6] = new LocalClusteredQueue(office1, 1, "sub7", channelIdManager.getId(), ms, pm, true, true, (QueuedExecutor)pool.get(), null, tr);
   protected void clusteredTransactionalRoute(boolean persistent) throws Throwable   {      ClusteredPostOffice office1 = null;            ClusteredPostOffice office2 = null;            try      {            //Start two offices                  office1 = createClusteredPostOffice(1, "testgroup");         office2 = createClusteredPostOffice(2, "testgroup");              LocalClusteredQueue[] queues = new LocalClusteredQueue[16];         Binding[] bindings = new Binding[16];                  queues[0] = new LocalClusteredQueue(office1, 1, "sub1", im.getId(), ms, pm, true, false, (QueuedExecutor)pool.get(), null, tr);                  bindings[0] = office1.bindClusteredQueue("topic1", queues[0]);                  queues[1] = new LocalClusteredQueue(office1, 1, "sub2", im.getId(), ms, pm, true, false, (QueuedExecutor)pool.get(), null, tr);                  bindings[1] = office1.bindClusteredQueue("topic1", queues[1]);                  queues[2] = new LocalClusteredQueue(office2, 2, "sub3", im.getId(), ms, pm, true, false, (QueuedExecutor)pool.get(), null, tr);                  bindings[2] = office2.bindClusteredQueue("topic1", queues[2]);                  queues[3] = new LocalClusteredQueue(office2, 2, "sub4", im.getId(), ms, pm, true, false, (QueuedExecutor)pool.get(), null, tr);                  bindings[3] = office2.bindClusteredQueue("topic1", queues[3]);                  queues[4] = new LocalClusteredQueue(office2, 2, "sub5", im.getId(), ms, pm, true, true, (QueuedExecutor)pool.get(), null, tr);                  bindings[4] = office2.bindClusteredQueue("topic1", queues[4]);                  queues[5] = new LocalClusteredQueue(office1, 1, "sub6", im.getId(), ms, pm, true, false, (QueuedExecutor)pool.get(), null, tr);                  bindings[5] = office1.bindClusteredQueue("topic1", queues[5]);                  queues[6] = new LocalClusteredQueue(office1, 1, "sub7", im.getId(), ms, pm, true, true, (QueuedExecutor)pool.get(), null, tr);                  bindings[6] = office1.bindClusteredQueue("topic1", queues[6]);                  queues[7] = new LocalClusteredQueue(office1, 1, "sub8", im.getId(), ms, pm, true, true, (QueuedExecutor)pool.get(), null, tr);                  bindings[7] = office1.bindClusteredQueue("topic1", queues[7]);                  queues[8] = new LocalClusteredQueue(office1, 1, "sub9", im.getId(), ms, pm, true, false, (QueuedExecutor)pool.get(), null, tr);                  bindings[8] = office1.bindClusteredQueue("topic2", queues[8]);                  queues[9] = new LocalClusteredQueue(office1, 1, "sub10", im.getId(), ms, pm, true, false, (QueuedExecutor)pool.get(), null, tr);                  bindings[9] = office1.bindClusteredQueue("topic2", queues[9]);                  queues[10] = new LocalClusteredQueue(office2, 2, "sub11", im.getId(), ms, pm, true, false, (QueuedExecutor)pool.get(), null, tr);                  bindings[10] = office2.bindClusteredQueue("topic2", queues[10]);                  queues[11] = new LocalClusteredQueue(office2, 2, "sub12", im.getId(), ms, pm, true, false, (QueuedExecutor)pool.get(), null, tr);                  bindings[11] = office2.bindClusteredQueue("topic2", queues[11]);                  queues[12] = new LocalClusteredQueue(office2, 2, "sub13", im.getId(), ms, pm, true, true, (QueuedExecutor)pool.get(), null, tr);                  bindings[12] = office2.bindClusteredQueue("topic2", queues[12]);                  queues[13] = new LocalClusteredQueue(office1, 1, "sub14", im.getId(), ms, pm, true, false, (QueuedExecutor)pool.get(), null, tr);                  bindings[13] = office1.bindClusteredQueue("topic2", queues[13]);                  queues[14] = new LocalClusteredQueue(office1, 1, "sub15", im.getId(), ms, pm, true, true, (QueuedExecutor)pool.get(), null, tr);                  bindings[14] = office1.bindClusteredQueue("topic2", queues[14]);                  queues[15] = new LocalClusteredQueue(office1, 1, "sub16", im.getId(), ms, pm, true, true, (QueuedExecutor)pool.get(), null, tr);                  bindings[15] = office1.bindClusteredQueue("topic2", queues[15]);         SimpleReceiver[] receivers = new SimpleReceiver[16];                  for (int i = 0; i < 16; i++)         {            receivers[i] = new SimpleReceiver("blah", SimpleReceiver.ACCEPTING);            queues[i].add(receivers[i]);         }                  //First for topic 1                  Message msg1 = CoreMessageFactory.createCoreMessage(1, persistent, null);;               MessageReference ref1 = ms.reference(msg1);                  Message msg2 = CoreMessageFactory.createCoreMessage(2, persistent, null);;               MessageReference ref2 = ms.reference(msg2);                  Transaction tx = tr.createTransaction();         boolean routed = office1.route(ref1, "topic1", tx);                  assertTrue(routed);         routed = office1.route(ref2, "topic1", tx);                  assertTrue(routed);                  for (int i = 0; i < 16; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                  tx.commit();                  //Messages are sent asych so may take some finite time to arrive         Thread.sleep(1000);                  for (int i = 0; i < 8; i++)         {                     List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertEquals(2, msgs.size());            Message msgRec1 = (Message)msgs.get(0);            assertEquals(msg1.getMessageID(), msgRec1.getMessageID());            Message msgRec2 = (Message)msgs.get(1);            assertEquals(msg2.getMessageID(), msgRec2.getMessageID());                        receivers[i].acknowledge(msgRec1, null);            receivers[i].acknowledge(msgRec2, null);            msgs = queues[i].browse();            assertNotNull(msgs);                        assertTrue(msgs.isEmpty());                                    receivers[i].clear();         }                  for (int i = 8; i < 16; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                  msg1 = CoreMessageFactory.createCoreMessage(1, persistent, null);;               ref1 = ms.reference(msg1);                  msg2 = CoreMessageFactory.createCoreMessage(2, persistent, null);;               ref2 = ms.reference(msg2);                  tx = tr.createTransaction();         routed = office1.route(ref1, "topic1", tx);                  assertTrue(routed);         routed = office1.route(ref2, "topic1", tx);                  assertTrue(routed);                  //Messages are sent asych so may take some finite time to arrive         Thread.sleep(1000);                           for (int i = 0; i < 16; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                  tx.rollback();                  for (int i = 0; i < 16; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                  //Now send some non transactionally                  msg1 = CoreMessageFactory.createCoreMessage(1, persistent, null);         ref1 = ms.reference(msg1);                  msg2 = CoreMessageFactory.createCoreMessage(2, persistent, null);         ref2 = ms.reference(msg2);                  routed = office1.route(ref1, "topic1", null);                  assertTrue(routed);         routed = office1.route(ref2, "topic1", null);                  assertTrue(routed);                  //Messages are sent asych so may take some finite time to arrive         Thread.sleep(1000);                           //And acknowledge transactionally                  tx = tr.createTransaction();                  for (int i = 0; i < 8; i++)         {                     List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertEquals(2, msgs.size());            Message msgRec1 = (Message)msgs.get(0);            assertEquals(msg1.getMessageID(), msgRec1.getMessageID());            Message msgRec2 = (Message)msgs.get(1);            assertEquals(msg2.getMessageID(), msgRec2.getMessageID());                                          receivers[i].acknowledge(msgRec1, tx);            receivers[i].acknowledge(msgRec2, tx);                                    msgs = queues[i].browse();                        assertNotNull(msgs);            assertEquals(2, msgs.size());            msgRec1 = (Message)msgs.get(0);            assertEquals(msg1.getMessageID(), msgRec1.getMessageID());            msgRec2 = (Message)msgs.get(1);            assertEquals(msg2.getMessageID(), msgRec2.getMessageID());                        receivers[i].clear();         }                  for (int i = 8; i < 16; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                  tx.commit();                  for (int i = 0; i < 16; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                           // and the rollback                  msg1 = CoreMessageFactory.createCoreMessage(1, persistent, null);;               ref1 = ms.reference(msg1);                  msg2 = CoreMessageFactory.createCoreMessage(2, persistent, null);;               ref2 = ms.reference(msg2);                  routed = office1.route(ref1, "topic1", null);                  assertTrue(routed);         routed = office1.route(ref2, "topic1", null);                  assertTrue(routed);                  Thread.sleep(1000);                          tx = tr.createTransaction();                  for (int i = 0; i < 8; i++)         {                     List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertEquals(2, msgs.size());            Message msgRec1 = (Message)msgs.get(0);            assertEquals(msg1.getMessageID(), msgRec1.getMessageID());            Message msgRec2 = (Message)msgs.get(1);            assertEquals(msg2.getMessageID(), msgRec2.getMessageID());                                          receivers[i].acknowledge(msgRec1, tx);            receivers[i].acknowledge(msgRec2, tx);                                    msgs = queues[i].browse();                        assertNotNull(msgs);            assertEquals(2, msgs.size());            msgRec1 = (Message)msgs.get(0);            assertEquals(msg1.getMessageID(), msgRec1.getMessageID());            msgRec2 = (Message)msgs.get(1);            assertEquals(msg2.getMessageID(), msgRec2.getMessageID());                     }                  for (int i = 8; i < 16; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                  tx.rollback();                  for (int i = 0; i < 8; i++)         {                     List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertEquals(2, msgs.size());            Message msgRec1 = (Message)msgs.get(0);            assertEquals(msg1.getMessageID(), msgRec1.getMessageID());            Message msgRec2 = (Message)msgs.get(1);            assertEquals(msg2.getMessageID(), msgRec2.getMessageID());                                                   msgs = queues[i].browse();                        assertNotNull(msgs);            assertEquals(2, msgs.size());            msgRec1 = (Message)msgs.get(0);            assertEquals(msg1.getMessageID(), msgRec1.getMessageID());            msgRec2 = (Message)msgs.get(1);            assertEquals(msg2.getMessageID(), msgRec2.getMessageID());                        receivers[i].acknowledge(msgRec1, null);            receivers[i].acknowledge(msgRec2, null);                                       receivers[i].clear();         }                  for (int i = 8; i < 16; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                           // Now for topic 2                  msg1 = CoreMessageFactory.createCoreMessage(1, persistent, null);             ref1 = ms.reference(msg1);                  msg2 = CoreMessageFactory.createCoreMessage(2, persistent, null);              ref2 = ms.reference(msg2);                  tx = tr.createTransaction();         routed = office1.route(ref1, "topic2", tx);                  assertTrue(routed);         routed = office1.route(ref2, "topic2", tx);                  assertTrue(routed);                                    for (int i = 0; i < 16; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                  tx.commit();                  //Messages are sent asych so may take some finite time to arrive         Thread.sleep(1000);                  for (int i = 0; i < 8; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                  for (int i = 8; i < 16; i++)         {                     List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertEquals(2, msgs.size());            Message msgRec1 = (Message)msgs.get(0);            assertEquals(msg1.getMessageID(), msgRec1.getMessageID());            Message msgRec2 = (Message)msgs.get(1);            assertEquals(msg2.getMessageID(), msgRec2.getMessageID());                        receivers[i].acknowledge(msgRec1, null);            receivers[i].acknowledge(msgRec2, null);            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());             receivers[i].clear();         }                  msg1 = CoreMessageFactory.createCoreMessage(1, persistent, null);;               ref1 = ms.reference(msg1);                  msg2 = CoreMessageFactory.createCoreMessage(2, persistent, null);;               ref2 = ms.reference(msg2);                  tx = tr.createTransaction();         routed = office1.route(ref1, "topic1", tx);                  assertTrue(routed);         routed = office1.route(ref2, "topic1", tx);                  assertTrue(routed);                  for (int i = 0; i < 16; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                  tx.rollback();                  for (int i = 0; i < 16; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                  //Now send some non transactionally                  msg1 = CoreMessageFactory.createCoreMessage(1, persistent, null);               ref1 = ms.reference(msg1);                  msg2 = CoreMessageFactory.createCoreMessage(2, persistent, null);               ref2 = ms.reference(msg2);                  routed = office1.route(ref1, "topic2", null);                  assertTrue(routed);         routed = office1.route(ref2, "topic2", null);                  assertTrue(routed);                  Thread.sleep(1000);                  //And acknowledge transactionally                  tx = tr.createTransaction();                  for (int i = 0; i < 8; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                  for (int i = 8; i < 16; i++)         {                     List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertEquals(2, msgs.size());            Message msgRec1 = (Message)msgs.get(0);            assertEquals(msg1.getMessageID(), msgRec1.getMessageID());            Message msgRec2 = (Message)msgs.get(1);            assertEquals(msg2.getMessageID(), msgRec2.getMessageID());                                          receivers[i].acknowledge(msgRec1, tx);            receivers[i].acknowledge(msgRec2, tx);                                    msgs = queues[i].browse();                        assertNotNull(msgs);            assertEquals(2, msgs.size());            msgRec1 = (Message)msgs.get(0);            assertEquals(msg1.getMessageID(), msgRec1.getMessageID());            msgRec2 = (Message)msgs.get(1);            assertEquals(msg2.getMessageID(), msgRec2.getMessageID());                        receivers[i].clear();         }                                    tx.commit();                  for (int i = 0; i < 16; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                           // and the rollback                  msg1 = CoreMessageFactory.createCoreMessage(1, persistent, null);;               ref1 = ms.reference(msg1);                  msg2 = CoreMessageFactory.createCoreMessage(2, persistent, null);;               ref2 = ms.reference(msg2);                  routed = office1.route(ref1, "topic2", null);                  assertTrue(routed);         routed = office1.route(ref2, "topic2", null);                  assertTrue(routed);                  Thread.sleep(1000);                   tx = tr.createTransaction();                  for (int i = 0; i < 8; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                  for (int i = 8; i < 16; i++)         {                     List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertEquals(2, msgs.size());            Message msgRec1 = (Message)msgs.get(0);            assertEquals(msg1.getMessageID(), msgRec1.getMessageID());            Message msgRec2 = (Message)msgs.get(1);            assertEquals(msg2.getMessageID(), msgRec2.getMessageID());                                          receivers[i].acknowledge(msgRec1, tx);            receivers[i].acknowledge(msgRec2, tx);                                    msgs = queues[i].browse();                        assertNotNull(msgs);            assertEquals(2, msgs.size());            msgRec1 = (Message)msgs.get(0);            assertEquals(msg1.getMessageID(), msgRec1.getMessageID());            msgRec2 = (Message)msgs.get(1);            assertEquals(msg2.getMessageID(), msgRec2.getMessageID());         }                                    tx.rollback();                  for (int i = 0; i < 8; i++)         {            List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());            msgs = queues[i].browse();            assertNotNull(msgs);            assertTrue(msgs.isEmpty());         }                  for (int i = 8; i < 16; i++)         {                     List msgs = receivers[i].getMessages();            assertNotNull(msgs);            assertEquals(2, msgs.size());            Message msgRec1 = (Message)msgs.get(0);            assertEquals(msg1.getMessageID(), msgRec1.getMessageID());            Message msgRec2 = (Message)msgs.get(1);            assertEquals(msg2.getMessageID(), msgRec2.getMessageID());                                                                msgs = queues[i].browse();                        assertNotNull(msgs);            assertEquals(2, msgs.size());            msgRec1 = (Message)msgs.get(0);            assertEquals(msg1.getMessageID(), msgRec1.getMessageID());            msgRec2 = (Message)msgs.get(1);            assertEquals(msg2.getMessageID(), msgRec2.getMessageID());                        receivers[i].acknowledge(msgRec1, null);            receivers[i].acknowledge(msgRec2, null);                                     receivers[i].clear();         }      }      finally      {         if (office1 != null)         {            try            {               office1.unbindClusteredQueue("sub7");               office1.unbindClusteredQueue("sub8");                          office1.unbindClusteredQueue("sub15");               office1.unbindClusteredQueue("sub16");            }            catch (Exception ignore)            {               ignore.printStackTrace();            }                                    office1.stop();         }                  if (office2 != null)         {            try            {               office2.unbindClusteredQueue("sub5");               office2.unbindClusteredQueue("sub13");            }            catch (Exception ignore)            {               ignore.printStackTrace();            }                        office2.stop();         }                  checkNoMessageData();      }   }