long id = messageIdGenerator.incrementAndGet();
// Each thread will Keep paging until all the messages are depaged.
// This is possible because the depage thread is not actually reading the pages.
// Just using the internal API to remove it from the page file system
ServerMessage msg = createMessage(id, storeImpl, destination, createRandomBuffer(id, 5));
if (storeImpl.page(msg, new RoutingContextImpl(null)))
{
buffers.put(id, msg);
}
else
{
break;
}
if (firstTime)
{
// We have at least one data paged. So, we can start depaging now
latchStart.countDown();
firstTime = false;
}
}
}
catch (Exception e)
{
e.printStackTrace();
this.e = e;
}
finally
{
aliveProducers.decrementAndGet();
}
}
}
class ReaderThread extends Thread
{
Exception e;
@Override
public void run()
{
try
{
// Wait every producer to produce at least one message
latchStart.await();
while (aliveProducers.get() > 0)
{
Page page = storeImpl.depage();
if (page != null)
{
readPages.add(page);
}
}
}
catch (Exception e)
{
e.printStackTrace();
this.e = e;
}
}
}
WriterThread producerThread[] = new WriterThread[numberOfThreads];
for (int i = 0; i < numberOfThreads; i++)
{
producerThread[i] = new WriterThread();
producerThread[i].start();
}
ReaderThread consumer = new ReaderThread();
consumer.start();
for (int i = 0; i < numberOfThreads; i++)
{
producerThread[i].join();
if (producerThread[i].e != null)
{
throw producerThread[i].e;
}
}
consumer.join();
if (consumer.e != null)
{
throw consumer.e;
}
final ConcurrentMap<Long, ServerMessage> buffers2 = new ConcurrentHashMap<Long, ServerMessage>();
for (Page page : readPages)
{
page.open();
List<PagedMessage> msgs = page.read(new NullStorageManager());
page.close();
for (PagedMessage msg : msgs)
{
long id = msg.getMessage().getBodyBuffer().readLong();
msg.getMessage().getBodyBuffer().resetReaderIndex();
ServerMessage msgWritten = buffers.remove(id);
buffers2.put(id, msg.getMessage());
Assert.assertNotNull(msgWritten);
Assert.assertEquals(msg.getMessage().getAddress(), msgWritten.getAddress());
UnitTestCase.assertEqualsBuffers(10, msgWritten.getBodyBuffer(), msg.getMessage().getBodyBuffer());
}
}
Assert.assertEquals(0, buffers.size());
List<String> files = factory.listFiles("page");
Assert.assertTrue(files.size() != 0);
for (String file : files)
{
SequentialFile fileTmp = factory.createSequentialFile(file, 1);
fileTmp.open();
Assert.assertTrue("The page file size (" + fileTmp.size() + ") shouldn't be > " + MAX_SIZE,
fileTmp.size() <= MAX_SIZE);
fileTmp.close();
}
TestSupportPageStore storeImpl2 = new PagingStoreImpl(PagingStoreImplTest.destinationTestName,
null,
100,
createMockManager(),
createStorageManagerMock(),
factory,
storeFactory,
new SimpleString("test"),
settings,
getExecutorFactory().getExecutor(),
true);
storeImpl2.start();
int numberOfPages = storeImpl2.getNumberOfPages();
Assert.assertTrue(numberOfPages != 0);
storeImpl2.startPaging();
storeImpl2.startPaging();
Assert.assertEquals(numberOfPages, storeImpl2.getNumberOfPages());
long lastMessageId = messageIdGenerator.incrementAndGet();
ServerMessage lastMsg = createMessage(lastMessageId, storeImpl, destination, createRandomBuffer(lastMessageId, 5));
storeImpl2.forceAnotherPage();
storeImpl2.page(lastMsg, new RoutingContextImpl(null));
buffers2.put(lastMessageId, lastMsg);
Page lastPage = null;
while (true)
{
Page page = storeImpl2.depage();
if (page == null)
{
break;
}
lastPage = page;
page.open();
List<PagedMessage> msgs = page.read(new NullStorageManager());
page.close();
for (PagedMessage msg : msgs)
{
long id = msg.getMessage().getBodyBuffer().readLong();
ServerMessage msgWritten = buffers2.remove(id);
Assert.assertNotNull(msgWritten);
Assert.assertEquals(msg.getMessage().getAddress(), msgWritten.getAddress());
UnitTestCase.assertEqualsByteArrays(msgWritten.getBodyBuffer().writerIndex(),
msgWritten.getBodyBuffer().toByteBuffer().array(),
msg.getMessage().getBodyBuffer().toByteBuffer().array());
}
}
lastPage.open();