code: @Override
  	protected final void onConsumerClose()
      {
      	super.onConsumerClose();
  
      	// Unlock waiting threads
      	prefetchSemaphore.release();
      	
      	try
      	{
          	CloseConsumerQuery query = new CloseConsumerQuery();
      		query.setSessionId(session.getId());
      		query.setConsumerId(id);
      		
      		// Append message IDs to rollback
      		synchronized (prefetchQueue)
              {
      		    while (!prefetchQueue.isEmpty())
      		    {
      		    	AbstractMessage msg = prefetchQueue.removeFirst();
      		        query.addUndeliveredMessageID(msg.getJMSMessageID());
      		    }
              }
      		
      		transportEndpoint.blockingRequest(query);
      	}
          catch (JMSException e)
          {
              ErrorTools.log(e, log);
          }
      }
docs: /* (non-Javadoc)
  @see net.timewalker.ffmq4.common.session.AbstractMessageConsumer#onConsumerClose()