@Test public void read_THREAD_SAFE_CHECK() throws InterruptedException { final int threadsCount = 5; final int judgmentsCount = 300; List<Long> ids = IntStream .rangeClosed(1, judgmentsCount) .mapToObj(x -> Long.valueOf(x)) .collect(Collectors.toList()); List<JudgmentIndexingItem> judgmentIndexingItems = IntStream .rangeClosed(1, judgmentsCount) .mapToObj(x -> new JudgmentIndexingItem(Long.valueOf(x), 0L)) .collect(Collectors.toList()); when(judgmentIndexingItemFetcher.fetchJudgmentIndexingItems()).thenReturn(judgmentIndexingItems); when(judgmentEnrichmentService.findOneAndEnrich(Mockito.anyLong())).thenAnswer(new Answer<Judgment>( ) { @Override public Judgment answer(InvocationOnMock invocation) throws Throwable { long id = (long)(invocation.getArguments()[0]); return (id >=1 && id <=judgmentsCount) ? createCcJudgment(id) : null; } }); judgmentIndexingReader.open(new ExecutionContext()); final CountDownLatch latch = new CountDownLatch(threadsCount); final List<List<Long>> threadsResultsIds = Lists.newLinkedList(); for (int i=0; i<threadsCount; ++i) { List<Long> threadResultsIds = runReaderThread(latch, i); threadsResultsIds.add(threadResultsIds); } latch.await(); List<Long> actualIds = threadsResultsIds.stream().flatMap(List::stream).collect(Collectors.toList()); assertThat(actualIds, containsInAnyOrder(ids.toArray())); }
@Override public void open(ExecutionContext executionContext) throws ItemStreamException { judgmentIndexingItems = new ConcurrentLinkedQueue<JudgmentIndexingItem>( judgmentIndexingItemFetcher.fetchJudgmentIndexingItems()); }
JudgmentIndexingReader implements ItemStreamReader<JudgmentIndexingData> { @Override public void open(ExecutionContext executionContext) throws ItemStreamException { judgmentIndexingItems = new ConcurrentLinkedQueue<JudgmentIndexingItem>( judgmentIndexingItemFetcher.fetchJudgmentIndexingItems()); } }
JudgmentIndexingReader implements ItemStreamReader<JudgmentIndexingData> { @Override public void open(ExecutionContext executionContext) throws ItemStreamException { judgmentIndexingItems = new ConcurrentLinkedQueue<JudgmentIndexingItem>( judgmentIndexingItemFetcher.fetchJudgmentIndexingItems()); }  }
JudgmentIndexingReader implements ItemStreamReader<JudgmentIndexingData> { @Override public void open(ExecutionContext executionContext) throws ItemStreamException { judgmentIndexingItems = new ConcurrentLinkedQueue<JudgmentIndexingItem>( judgmentIndexingItemFetcher.fetchJudgmentIndexingItems()); }  @Override void open(ExecutionContext executionContext); @Override JudgmentIndexingData read(); @Override void update(ExecutionContext executionContext); @Override void close(); @Autowired void setJudgmentEnrichmentService(JudgmentEnrichmentService judgmentEnrichmentService); @Autowired void setJudgmentIndexingItemFetcher(JudgmentIndexingItemFetcher judgmentIndexingItemFetcher); }
JudgmentIndexingReader implements ItemStreamReader<JudgmentIndexingData> { @Override public void open(ExecutionContext executionContext) throws ItemStreamException { judgmentIndexingItems = new ConcurrentLinkedQueue<JudgmentIndexingItem>( judgmentIndexingItemFetcher.fetchJudgmentIndexingItems()); }  @Override void open(ExecutionContext executionContext); @Override JudgmentIndexingData read(); @Override void update(ExecutionContext executionContext); @Override void close(); @Autowired void setJudgmentEnrichmentService(JudgmentEnrichmentService judgmentEnrichmentService); @Autowired void setJudgmentIndexingItemFetcher(JudgmentIndexingItemFetcher judgmentIndexingItemFetcher);  }