// at this point we should have sent out only highMark messages
// which have not bee received so will be queued up in the channel
// which should be suspended
assertTrue(_subscription.isSuspended());
UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap();
assertTrue(map.size() == highMark);
//acknowledge messages so we are just above lowMark
_channel.acknowledgeMessage(lowMark - 1, true);
//we should still be suspended
assertTrue(_subscription.isSuspended());
assertTrue(map.size() == lowMark + 1);
//acknowledge one more message
_channel.acknowledgeMessage(lowMark, true);
//and suspension should be lifted
assertTrue(!_subscription.isSuspended());
//pubilsh more msgs so we are just below the limit
publishMessages(lowMark - 1);
//we should not be suspended
assertTrue(!_subscription.isSuspended());
//acknowledge all messages
_channel.acknowledgeMessage(0, true);
try
{
Thread.sleep(3000);
}
catch (InterruptedException e)
{
_log.error("Error: " + e, e);
}
//map will be empty
assertTrue(map.size() == 0);
}