@Test public void testSend() throws Exception { Mutation.Metadata metadata = mock(Mutation.Metadata.class); when(firstMutation.getMetadata()).thenReturn(metadata); when(secondMutation.getMetadata()).thenReturn(metadata); when(thirdMutation.getMetadata()).thenReturn(metadata); when(metadata.getTimestamp()).thenReturn(0L); destination.send(mutations); assertEquals(3, destination.getPublishedMutations()); assertEquals(thirdMutation, destination.getLastPublishedMutation()); verify(metrics).publishSucceeded(mutations); verify(listener).onSend(mutations); }
@SuppressWarnings("unchecked") @Override public void send(@NonNull final List<? extends Mutation<?>> mutations) { if (mutations.isEmpty()) { return; } try { final Stopwatch stopwatch = Stopwatch.createStarted(); final List<T> messages = mapper.apply(mutations.stream().collect(Collectors.toList())); final Mutation<?> latestMutation = mutations.get(mutations.size() - 1); delay(latestMutation); publish(messages); lastPublishedMutation.set(latestMutation); stopwatch.stop(); final long time = stopwatch.elapsed(TimeUnit.MILLISECONDS); metrics.publishTime(time); metrics.publishSucceeded(mutations); log(mutations); notifySend(mutations); } catch (Exception ex) { log.error("Failed to send {} mutations.", mutations.size(), ex); mutations.forEach(mutation -> metrics.publishFailed(mutation, ex)); throw new DestinationException("Failed to send mutations", ex); } }
AbstractDestination extends ListenableDestination { @SuppressWarnings("unchecked") @Override public void send(@NonNull final List<? extends Mutation<?>> mutations) { if (mutations.isEmpty()) { return; } try { final Stopwatch stopwatch = Stopwatch.createStarted(); final List<T> messages = mapper.apply(mutations.stream().collect(Collectors.toList())); final Mutation<?> latestMutation = mutations.get(mutations.size() - 1); delay(latestMutation); publish(messages); lastPublishedMutation.set(latestMutation); stopwatch.stop(); final long time = stopwatch.elapsed(TimeUnit.MILLISECONDS); metrics.publishTime(time); metrics.publishSucceeded(mutations); log(mutations); notifySend(mutations); } catch (Exception ex) { log.error("Failed to send {} mutations.", mutations.size(), ex); mutations.forEach(mutation -> metrics.publishFailed(mutation, ex)); throw new DestinationException("Failed to send mutations", ex); } } }
AbstractDestination extends ListenableDestination { @SuppressWarnings("unchecked") @Override public void send(@NonNull final List<? extends Mutation<?>> mutations) { if (mutations.isEmpty()) { return; } try { final Stopwatch stopwatch = Stopwatch.createStarted(); final List<T> messages = mapper.apply(mutations.stream().collect(Collectors.toList())); final Mutation<?> latestMutation = mutations.get(mutations.size() - 1); delay(latestMutation); publish(messages); lastPublishedMutation.set(latestMutation); stopwatch.stop(); final long time = stopwatch.elapsed(TimeUnit.MILLISECONDS); metrics.publishTime(time); metrics.publishSucceeded(mutations); log(mutations); notifySend(mutations); } catch (Exception ex) { log.error("Failed to send {} mutations.", mutations.size(), ex); mutations.forEach(mutation -> metrics.publishFailed(mutation, ex)); throw new DestinationException("Failed to send mutations", ex); } }  }
AbstractDestination extends ListenableDestination { @SuppressWarnings("unchecked") @Override public void send(@NonNull final List<? extends Mutation<?>> mutations) { if (mutations.isEmpty()) { return; } try { final Stopwatch stopwatch = Stopwatch.createStarted(); final List<T> messages = mapper.apply(mutations.stream().collect(Collectors.toList())); final Mutation<?> latestMutation = mutations.get(mutations.size() - 1); delay(latestMutation); publish(messages); lastPublishedMutation.set(latestMutation); stopwatch.stop(); final long time = stopwatch.elapsed(TimeUnit.MILLISECONDS); metrics.publishTime(time); metrics.publishSucceeded(mutations); log(mutations); notifySend(mutations); } catch (Exception ex) { log.error("Failed to send {} mutations.", mutations.size(), ex); mutations.forEach(mutation -> metrics.publishFailed(mutation, ex)); throw new DestinationException("Failed to send mutations", ex); } }  @Override Mutation<?> getLastPublishedMutation(); @SuppressWarnings("unchecked") @Override void send(@NonNull final List<? extends Mutation<?>> mutations); abstract void publish(List<T> messages); @Override boolean isStarted(); @Override void open(); @Override void close(); @Override void clear(); }
AbstractDestination extends ListenableDestination { @SuppressWarnings("unchecked") @Override public void send(@NonNull final List<? extends Mutation<?>> mutations) { if (mutations.isEmpty()) { return; } try { final Stopwatch stopwatch = Stopwatch.createStarted(); final List<T> messages = mapper.apply(mutations.stream().collect(Collectors.toList())); final Mutation<?> latestMutation = mutations.get(mutations.size() - 1); delay(latestMutation); publish(messages); lastPublishedMutation.set(latestMutation); stopwatch.stop(); final long time = stopwatch.elapsed(TimeUnit.MILLISECONDS); metrics.publishTime(time); metrics.publishSucceeded(mutations); log(mutations); notifySend(mutations); } catch (Exception ex) { log.error("Failed to send {} mutations.", mutations.size(), ex); mutations.forEach(mutation -> metrics.publishFailed(mutation, ex)); throw new DestinationException("Failed to send mutations", ex); } }  @Override Mutation<?> getLastPublishedMutation(); @SuppressWarnings("unchecked") @Override void send(@NonNull final List<? extends Mutation<?>> mutations); abstract void publish(List<T> messages); @Override boolean isStarted(); @Override void open(); @Override void close(); @Override void clear();  }