FindChildrenStreamEndPlugin implements StreamEndPlugin { @Override public void streamEnded(Context context) { context.modifyParentOrChild( new Function<AtomicBoolean, Void>() { @Override public Void apply(AtomicBoolean isParentDirty) { if (isParentDirty.get()) { context .getParentMetacard() .ifPresent(parentMetacard -> handleParentMetacard(context, parentMetacard)); isParentDirty.set(false); } return null; } }); } FindChildrenStreamEndPlugin(FilterBuilder filterBuilder, Factory factory); Factory getFactory(); @Override void streamEnded(Context context); static final int MAX_SUBSEQUENT_ERROR_COUNT; static final int BATCH_SIZE; }
@Test public void testIrregularBatching() throws UnsupportedQueryException, SourceUnavailableException, FederationException { long expectedCount = 10; Metacard metacard1 = mock(Metacard.class); Metacard metacard2 = mock(Metacard.class); Metacard metacard3 = mock(Metacard.class); Metacard metacard4 = mock(Metacard.class); Metacard metacard5 = mock(Metacard.class); Metacard metacard6 = mock(Metacard.class); Metacard metacard7 = mock(Metacard.class); Metacard metacard8 = mock(Metacard.class); Metacard metacard9 = mock(Metacard.class); Metacard metacard10 = mock(Metacard.class); Result result1 = mock(Result.class); when(result1.getMetacard()).thenReturn(metacard1); Result result2 = mock(Result.class); when(result2.getMetacard()).thenReturn(metacard2); Result result3 = mock(Result.class); when(result3.getMetacard()).thenReturn(metacard3); Result result4 = mock(Result.class); when(result4.getMetacard()).thenReturn(metacard4); Result result5 = mock(Result.class); when(result5.getMetacard()).thenReturn(metacard5); Result result6 = mock(Result.class); when(result6.getMetacard()).thenReturn(metacard6); Result result7 = mock(Result.class); when(result7.getMetacard()).thenReturn(metacard7); Result result8 = mock(Result.class); when(result8.getMetacard()).thenReturn(metacard8); Result result9 = mock(Result.class); when(result9.getMetacard()).thenReturn(metacard9); Result result10 = mock(Result.class); when(result10.getMetacard()).thenReturn(metacard10); QueryResponse queryResponse1 = mock(QueryResponse.class); when(queryResponse1.getHits()).thenReturn(expectedCount); when(queryResponse1.getResults()).thenReturn(Arrays.asList(result1, result2)); QueryResponse queryResponse2 = mock(QueryResponse.class); when(queryResponse2.getHits()).thenReturn(expectedCount); when(queryResponse2.getResults()).thenReturn(Arrays.asList(result3, result4, result5)); QueryResponse queryResponse3 = mock(QueryResponse.class); when(queryResponse3.getHits()).thenReturn(expectedCount); when(queryResponse3.getResults()) .thenReturn(Arrays.asList(result6, result7, result8, result9, result10)); when(catalogFramework.query(any())) .thenThrow(SourceUnavailableException.class) .thenReturn(queryResponse1, queryResponse2) .thenThrow(SourceUnavailableException.class) .thenReturn(queryResponse3); setIsParentDirty(true); findChildrenStreamEndPlugin.streamEnded(context); ArgumentCaptor<List> argumentCaptor = ArgumentCaptor.forClass(List.class); verify(handler, times(3)) .handle(Matchers.eq(context), Matchers.eq(parentMetacard), argumentCaptor.capture()); verify(handler).end(Matchers.eq(context), Matchers.eq(parentMetacard)); long capturedMetacardCount = argumentCaptor.getAllValues().stream().mapToLong(List::size).sum(); assertThat(capturedMetacardCount, is(expectedCount)); }
@Test(timeout = 10000) public void testSubsequentErrors() throws UnsupportedQueryException, SourceUnavailableException, FederationException { Throwable[] exceptions = new Throwable[FindChildrenStreamEndPlugin.MAX_SUBSEQUENT_ERROR_COUNT + 1]; IntStream.range(0, FindChildrenStreamEndPlugin.MAX_SUBSEQUENT_ERROR_COUNT + 1) .forEach(i -> exceptions[i] = new SourceUnavailableException()); when(catalogFramework.query(any())).thenThrow(exceptions); setIsParentDirty(true); findChildrenStreamEndPlugin.streamEnded(context); verify(handler, never()) .handle(Matchers.eq(context), Matchers.eq(parentMetacard), any(List.class)); verify(handler, times(1)).end(Matchers.eq(context), Matchers.eq(parentMetacard)); }
@Test public void testBatching() throws UnsupportedQueryException, SourceUnavailableException, FederationException { setIsParentDirty(true); findChildrenStreamEndPlugin.streamEnded(context); assertHandlerCalls(2, 0, 1, 2, 0, 1); }
@Test public void testHandlerThrowsRuntimeException() throws UnsupportedQueryException, SourceUnavailableException, FederationException { when(catalogFramework.query(any())) .thenAnswer( invocationOnMock -> { QueryRequest queryRequest = invocationOnMock.getArgument(0); switch (queryRequest.getQuery().getStartIndex()) { case 1: return queryResponse1; case 2: return queryResponse2; } return null; }); doThrow(new RuntimeException()) .doNothing() .when(handler) .handle(Matchers.eq(context), Matchers.eq(parentMetacard), any(List.class)); setIsParentDirty(true); findChildrenStreamEndPlugin.streamEnded(context); assertHandlerCalls(3, 1, 2, 3, 1, 2); }
@Test public void testIsParentDirty() throws UnsupportedQueryException, SourceUnavailableException, FederationException { setIsParentDirty(false); findChildrenStreamEndPlugin.streamEnded(context); verify(catalogFramework, never()).query(Matchers.any()); }