DestinationPool extends ListenableDestination { @Override public synchronized void send(@NonNull final List<? extends Mutation<?>> mutations) { mutations .stream() .collect(groupingBy(this::getPartitionId, LinkedHashMap::new, toList())) .forEach( (id, mutationList) -> { log.debug("Sending {} mutations to destination {}.", mutationList.size(), id); isActive[id] = true; destinations.get(id).send(mutationList); }); } DestinationPool(
      @NonNull final KeyProvider<Mutation<?>, String> keyProvider,
      @NonNull final List<Destination> destinations); int getPoolSize(); @Override synchronized Mutation<?> getLastPublishedMutation(); @Override synchronized void send(@NonNull final List<? extends Mutation<?>> mutations); @Override boolean isStarted(); @Override void open(); @Override void close(); @Override void clear();  }
@Test public void testSend() throws Exception { Mutation<?> firstMutation = mock(Mutation.class); Mutation<?> secondMutation = mock(Mutation.class); Mutation<?> thirdMutation = mock(Mutation.class); Mutation<?> fourthMutation = mock(Mutation.class); Mutation<?> fifthMutation = mock(Mutation.class); Mutation.Metadata firstMetadata = mock(Mutation.Metadata.class); Mutation.Metadata secondMetadata = mock(Mutation.Metadata.class); Mutation.Metadata thirdMetadata = mock(Mutation.Metadata.class); when(firstMetadata.getId()).thenReturn(3L); when(secondMetadata.getId()).thenReturn(2L); when(thirdMetadata.getId()).thenReturn(4L); when(firstMutation.getMetadata()).thenReturn(firstMetadata); when(secondMutation.getMetadata()).thenReturn(secondMetadata); when(thirdMutation.getMetadata()).thenReturn(thirdMetadata); when(keyProvider.get(firstMutation)).thenReturn("4"); when(keyProvider.get(secondMutation)).thenReturn("6"); when(keyProvider.get(thirdMutation)).thenReturn("9"); when(keyProvider.get(fourthMutation)).thenReturn("2"); when(keyProvider.get(fifthMutation)).thenReturn("1"); assertEquals(null, firstDestination.getLastPublishedMutation()); destinationPool.send( Arrays.asList(firstMutation, secondMutation, thirdMutation, fourthMutation, fifthMutation)); verify(firstDestination, times(1)).send(Collections.singletonList(firstMutation)); verify(secondDestination, times(1)).send(ImmutableList.of(thirdMutation, fifthMutation)); verify(thirdDestination, times(1)).send(ImmutableList.of(secondMutation, fourthMutation)); when(firstDestination.getLastPublishedMutation()).thenReturn((Mutation) firstMutation); when(secondDestination.getLastPublishedMutation()).thenReturn((Mutation) thirdMutation); when(thirdDestination.getLastPublishedMutation()).thenReturn((Mutation) secondMutation); when(fourthDestination.getLastPublishedMutation()).thenReturn(null); assertEquals(secondMutation, destinationPool.getLastPublishedMutation()); when(firstDestination.getLastPublishedMutation()).thenReturn(null); assertEquals(null, destinationPool.getLastPublishedMutation()); }
@Test public void testSendFailure() throws Exception { Destination corruptDestination = new TestDestination(); Mutation<?> firstMutation = mock(Mutation.class); Mutation<?> secondMutation = mock(Mutation.class); when(keyProvider.get(firstMutation)).thenReturn("0"); when(keyProvider.get(secondMutation)).thenReturn("1"); Destination.Listener listener = mock(Destination.Listener.class); DestinationPool pool = new DestinationPool( keyProvider, Arrays.asList(mock(Destination.class), corruptDestination)); pool.addListener(listener); pool.send(ImmutableList.of(firstMutation, secondMutation)); verify(listener, times(1)).onError(any(RuntimeException.class)); }