if (immediateAsynchronousAcknowledgment) { log.info("simulating an asynchronous ACK that arrives before we return the delivery to channel"); try { delivery.acknowledge(null); } catch(Throwable t) { log.error("Cannot acknowledge", t); } }
   public Delivery handle(DeliveryObserver observer, Routable r, Transaction tx)   {      try      {         if (r == null)         {            log.info("Receiver [" + name + "] is rejecting a null message");            return null;         }         if (BROKEN.equals(state))         {            throw new RuntimeException("THIS IS AN EXCEPTION THAT SIMULATES "+                                       "THE BEHAVIOUR OF A BROKEN RECEIVER");         }         boolean done = ACKING.equals(state) ? true : false;         log.info("Receiver [" + name + "] is " + (done ? "ACKing" : "NACKing") +  " message " + r);                  MessageReference ref = (MessageReference)r;         Message m = ref.getMessage();                  Delivery delivery = new SimpleDelivery(observer, ref, done);         messages.add(new Object[] {m, done ? null : delivery});         return delivery;      }      finally      {         synchronized(waitingArea)         {            if (futureState != null && --invocationsToFutureStateCount == 0)            {               state = futureState;               futureState = null;            }            Integer crt = (Integer)waitingArea.get(INVOCATION_COUNT);            waitingArea.put(INVOCATION_COUNT, new Integer(crt.intValue() + 1));            waitingArea.notifyAll();         }      }   }