queueNameNodeIdMap.put(queue.getName(), queue.getNodeId());
queueNameNodeIdMap.put(queue.getName(), new Integer(queue.getNodeId()));
   public boolean route(MessageReference ref, String condition, Transaction tx) throws Exception   {      if (ref == null)      {         throw new IllegalArgumentException("Message reference is null");      }            if (condition == null)      {         throw new IllegalArgumentException("Condition is null");      }            boolean routed = false;            lock.readLock().acquire();            log.info(this.nodeId + " routing reference " + ref);            try      {               ClusteredBindings cb = (ClusteredBindings)conditionMap.get(condition);                  boolean startInternalTx = false;                  String lastNodeId = null;                  if (cb != null)         {            if (tx == null && ref.isReliable())            {                               if (!(cb.getDurableCount() == 0 || (cb.getDurableCount() == 1 && cb.getLocalDurableCount() == 1)))               {                  // When routing a persistent message without a transaction then we may need to start an                   // internal transaction in order to route it.                  // This is so we can guarantee the message is delivered to all or none of the subscriptions.                  // We need to do this if there is anything other than                  // No durable subs or exactly one local durable sub                  startInternalTx = true;               }            }                                                if (startInternalTx)            {               tx = tr.createTransaction();            }                            int numberRemote = 0;                        Map queueNameNodeIdMap = null;                        long lastChannelId = -1;                        Collection routers = cb.getRouters();            Iterator iter = routers.iterator();                                 while (iter.hasNext())            {               ClusterRouter router = (ClusterRouter)iter.next();                              Delivery del = router.handle(null, ref, tx);                              if (del != null && del.isSelectorAccepted())               {                  routed = true;                                 ClusteredQueue queue = (ClusteredQueue)del.getObserver();                                    log.info("Routing message to queue:" + queue.getName() + " on node " + queue.getNodeId());                                    if (router.numberOfReceivers() > 1)                  {                     //We have now chosen which one will receive the message so we need to add this                     //information to a map which will get sent when casting - so the the queue                     //on the receiving node knows whether to receive the message                     if (queueNameNodeIdMap == null)                     {                        queueNameNodeIdMap = new HashMap();                     }                                          queueNameNodeIdMap.put(queue.getName(), queue.getNodeId());                  }                                    if (!queue.isLocal())                  {                     //We need to send the message remotely                     numberRemote++;                                          lastNodeId = queue.getNodeId();                                                                                                                              lastChannelId = queue.getChannelID();                  }               }            }                        //Now we've sent the message to any local queues, we might also need            //to send the message to the other office instances on the cluster if there are            //queues on those nodes that need to receive the message                        //TODO - there is an innefficiency here, numberRemote does not take into account that more than one            //of the number remote may be on the same node, so we could end up multicasting            //when unicast would do            if (numberRemote > 0)            {               if (tx == null)               {                  if (numberRemote == 1)                  {                  //   log.info("unicast no tx");                     //Unicast - only one node is interested in the message                                          //TODO - temporarily commented out until can get unicast to work                                          //asyncSendRequest(new MessageRequest(condition, ref.getMessage(), null), lastNodeId);                     asyncSendRequest(new MessageRequest(condition, ref.getMessage(), queueNameNodeIdMap));                  }                  else                  {                  //   log.info("multicast no tx");                     //Multicast - more than one node is interested                     asyncSendRequest(new MessageRequest(condition, ref.getMessage(), queueNameNodeIdMap));                  }                                                }               else               {                  CastMessagesCallback callback = (CastMessagesCallback)tx.getCallback(this);                                    if (callback == null)                  {                     callback = new CastMessagesCallback(nodeId, tx.getId(), DefaultClusteredPostOffice.this);                                          //This callback MUST be executed first                                          //Execution order is as follows:                     //Before commit:                     //1. Cast messages across network - get added to holding area (if persistent) on receiving                     //nodes                     //2. Persist messages in persistent store                     //After commit                     //1. Cast commit message across network                     tx.addFirstCallback(callback, this);                  }                                        callback.addMessage(condition, ref.getMessage(), queueNameNodeIdMap,                                      numberRemote == 1 ? lastNodeId : null,                                      lastChannelId);                   }            }                                                            if (startInternalTx)            {                              tx.commit();            }         }      }      finally      {                           lock.readLock().release();      }               return routed;    }