@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)); }
@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 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 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); }
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(); }
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();  }