{
locator.setBlockOnNonDurableSend(true);
locator.setBlockOnDurableSend(true);
locator.setReconnectAttempts(-1);
ClientSessionFactoryInternal sf = createSessionFactoryAndWaitForTopology(locator, 2);
ClientSession session = createSession(sf, false, false);
session.createQueue(FailoverTestBase.ADDRESS, FailoverTestBase.ADDRESS, null, true);
// create a consumer and start the session before failover
ClientConsumer consumer = session.createConsumer(FailoverTestBase.ADDRESS);
session.start();
ClientProducer producer = session.createProducer(FailoverTestBase.ADDRESS);
final int numMessages = 100;
for (int i = 0; i < numMessages; i++)
{
ClientMessage message = session.createMessage(i % 2 == 0);
setBody(i, message);
message.putIntProperty("counter", i);
producer.send(message);
}
// messages will be delivered to the consumer when the session is committed
session.commit();
Assert.assertFalse(session.isRollbackOnly());
crash(session);
session.commit();
session.close();
session = createSession(sf, false, false);
consumer = session.createConsumer(FailoverTestBase.ADDRESS);
session.start();
for (int i = 0; i < numMessages; i++)
{
// Only the persistent messages will survive
if (i % 2 == 0)
{
ClientMessage message = consumer.receive(1000);
Assert.assertNotNull(message);
assertMessageBody(i, message);
Assert.assertEquals(i, message.getIntProperty("counter").intValue());
message.acknowledge();
}
}
Assert.assertNull(consumer.receiveImmediate());
session.commit();
session.close();
sf.close();
Assert.assertEquals(0, sf.numSessions());
Assert.assertEquals(0, sf.numConnections());
}