MessageReference ref = ms.reference(r);
MessageReference ref; if (r.isReference()) { ref = ms.reference((MessageReference)r); } else { ref = ms.reference((Message)r); }
      public void receive(org.jgroups.Message jgroupsMessage)      {         Object  o = jgroupsMessage.getObject();         try         {            if (!(o instanceof Routable))            {               if (log.isTraceEnabled()) { log.trace(this + " discarding non-routable (" + o + ")"); }               return;            }            Routable r = (Routable)o;            Object replicatorID = r.getHeader(Routable.REPLICATOR_ID);            if (!getGroupID().equals(replicatorID))            {               if (log.isTraceEnabled()) { log.trace(this + " discarding routable from a different replicator (" + replicatorID + ")"); }               return;            }            if (log.isTraceEnabled()) { log.trace(this + " received " + r); }            Object collectorID = r.getHeader(Routable.COLLECTOR_ID);            boolean isReliable = r.isReliable();            Address collectorAddress = null;            if (isReliable)            {               // we only need to send asynchronous acknowledgments back for reliable messages               Set ids = viewKeeper.getRemotePeers();               for(Iterator i = ids.iterator(); i.hasNext(); )               {                  PeerIdentity pid = (PeerIdentity)i.next();                  if (pid.getPeerID().equals(collectorID))                  {                     collectorAddress = pid.getAddress();                     break;                  }               }               if (collectorAddress == null)               {                  log.error("Collector " + Util.guidToString(collectorID) +                            " not found among the peers of " + ReplicatorOutput.this);                  return;               }            }            if (ignoredReplicatorPeerID != null && ignoredReplicatorPeerID.equals(collectorID))            {               if (log.isTraceEnabled()) { log.trace(this + ": message " + r + " sent by an ignored replicator, discarding"); }               // However, I need to send an acknowledgment back for reliable messages, otherwise               // the originator replicator input will never get rid of the associated delivery               // from its RemoteTopic state.               //               // TODO: this solution is inneficient and confusing. A more efficient mechanism would               //       be for to code the replicator input to not even *WAIT* for acknowledments               //       from ignored outputs.               if (isReliable)               {                  sendAsynchronousResponse(collectorAddress, r.getMessageID(), Acknowledgment.ACCEPTED);               }               return;            }            if (receiver == null)            {               if (isReliable)               {                  // no receiver to forward to, asynchronously reject the message                  sendAsynchronousResponse(collectorAddress, r.getMessageID(), Acknowledgment.REJECTED);               }               return;            }            MessageReference ref = ms.reference(r);            // Mark the reference as being received from a remote endpoint            ref.putHeader(Routable.REMOTE_ROUTABLE, Routable.REMOTE_ROUTABLE);            ref.removeHeader(Routable.REPLICATOR_ID);            ref.removeHeader(Routable.COLLECTOR_ID);            Delivery d = null;            try            {               ref.putHeader(REPLICATOR_OUTPUT_COLLECTOR_ADDRESS, collectorAddress);               d = receiver.handle(this, ref, null);               if (log.isTraceEnabled()) { log.trace(this + ": receiver " + receiver + " handled " + r + " and returned " + d); }            }            catch(Throwable t)            {               log.warn(ReplicatorOutput.this + "'s receiver is broken", t);               if (isReliable)               {                  // broken receiver, asynchronously reject the message                  sendAsynchronousResponse(collectorAddress, r.getMessageID(), Acknowledgment.REJECTED);               }            }            if (isReliable)            {               // we only need to send asynchronous acknowledgments back for reliable messages               if (d.isDone())               {                  acknowledge(d, null);               }               else if (d.isCancelled())               {                  try                  {                     cancel(d);                  }                  catch(Throwable t)                  {                     // will never throw Throwable                  }               }               else               {                  if (log.isTraceEnabled()) { log.trace(this + " starts waiting asynchronous acknowledgment from " + receiver); }               }            }         }         finally         {            if (delegate != null)            {               // forward the message to delegate, there may be other ReplicatorOutput listeners               // entitled to get the message or other generic listeners               if (log.isTraceEnabled()) { log.trace(this + " forwards message to delegate listener " + delegate); }               delegate.receive(jgroupsMessage);            }         }      }