DuraStoreRetrievalSource implements RetrievalSource { @Override public synchronized ContentItem getNextContentItem() { if (currentContentList != null && currentContentList.hasNext()) { return new ContentItem(currentSpaceId, currentContentList.next()); } else if (spaceIds.hasNext()) { getNextSpace(); return getNextContentItem(); } else { return null; } } DuraStoreRetrievalSource(ContentStore store,
                                    List<String> spaces,
                                    boolean allSpaces); @Override synchronized ContentItem getNextContentItem(); @Override Map<String, String> getSourceProperties(ContentItem contentItem); @Override String getSourceChecksum(ContentItem contentItem); @Override ContentStream getSourceContent(ContentItem contentItem, RetrievalListener listener);  }
@Test public void testGetNextItem() throws Exception { List<String> spaces = new ArrayList<String>(); spaces.add("space1"); spaces.add("space2"); RetrievalSource source = new DuraStoreRetrievalSource(store, spaces, false); int count; for (count = 0; source.getNextContentItem() != null; count++) { } assertEquals(6, count); source = new DuraStoreRetrievalSource(store, null, true); for (count = 0; source.getNextContentItem() != null; count++) { } assertEquals(9, count); }
@Test public void testGetNextItemMultithreaded() throws Exception { int spaceCount = 4; int contentCount = 100; store = createMockContentStore(spaceCount, contentCount); RetrievalSource source = new DuraStoreRetrievalSource(store, null, true); int threadCount = 20; final CountDownLatch latch = new CountDownLatch(spaceCount * contentCount); for (int i = 0; i < threadCount; i++) { new Thread(new Runnable() { @Override public void run() { ContentItem item = null; while ((item = source.getNextContentItem()) != null) { assertNotNull(item.getSpaceId()); assertNotNull(item.getContentId()); latch.countDown(); } } }).start(); } Assert.assertTrue(latch.await(5, TimeUnit.SECONDS)); }