MessageReference ref = ms.reference(routable);
MessageReference ref; if (!routable.isReference()) { ref = ms.reference((Message)routable); } else { ref = (MessageReference)routable; }
   public Set handle(DeliveryObserver observer, Routable routable, Transaction tx)   {      if (!joined)      {         return null;      }      if (log.isTraceEnabled()) { log.trace(this + " handles " + routable); }      Set outputs = getOutputs();      if (outputs.isEmpty())      {         if (log.isTraceEnabled()) { log.trace(this + " has no outputs, rejecting message"); }         return Collections.EMPTY_SET;      }      Set deliveries = new HashSet();      MessageReference ref = ms.reference(routable);      // expect asynchronous acknowledgments only for reliable messages      if (!routable.isReliable())      {         if (log.isTraceEnabled()) { log.trace("non-reliable message, acknowledged immediately"); }         deliveries.add(new SimpleDelivery(ref, true));      }      else      {         for(Iterator i = outputs.iterator(); i.hasNext(); )         {            Delivery d = new ReplicatorOutputDelivery(observer, ref,                                                      ((PeerIdentity)i.next()).getPeerID(),                                                      cancelOnMessageRejection);            deliveries.add(d);         }         collector.startCollecting(deliveries);      }      // dereference it and send the message      Message message = ref.getMessage();      // TODO when sharing a peristent store, it is possible to only send references here      message.putHeader(Routable.REPLICATOR_ID, getReplicatorID());      message.putHeader(Routable.COLLECTOR_ID, collector.getID());      try      {         if (log.isTraceEnabled()) { log.trace(this + " sending " + message + " on the channel"); }         dispatcher.getChannel().send(null, null, message);         if (log.isTraceEnabled()) { log.trace(this + " multicast " + message); }      }      catch(Throwable t)      {         log.error("Failed to put the message on the channel", t);         collector.remove(deliveries);         return null;      }      return deliveries;   }