@Test public void testSend() throws Exception { when(destination.isStarted()).thenReturn(true); bufferedDestination.send(ImmutableList.of(firstMutation, secondMutation)); bufferedDestination.processMutations(); verify(destination).send(ImmutableList.of(firstMutation, secondMutation)); bufferedDestination.send(ImmutableList.of(firstMutation)); bufferedDestination.send(ImmutableList.of(secondMutation, thirdMutation)); bufferedDestination.processMutations(); verify(destination).send(mutations); }
@Override public void send(@NonNull final List<? extends Mutation<?>> mutations) { try { if (mutations.isEmpty()) { return; } final Stopwatch stopwatch = Stopwatch.createStarted(); final Mutation.Metadata metadata = mutations.get(0).getMetadata(); if (mutationBuffer.remainingCapacity() == 0) { metrics.bufferFull(metadata); } mutationBuffer.put(mutations); metrics.bufferSize(mutationBuffer.size(), metadata); stopwatch.stop(); final long time = stopwatch.elapsed(TimeUnit.MILLISECONDS); metrics.sendTime(time); } catch (Exception ex) { log.error("Failed to send mutations.", ex); metrics.sendFailed(ex); throw new DestinationException("Failed to send mutations", ex); } }
BufferedDestination extends ListenableDestination { @Override public void send(@NonNull final List<? extends Mutation<?>> mutations) { try { if (mutations.isEmpty()) { return; } final Stopwatch stopwatch = Stopwatch.createStarted(); final Mutation.Metadata metadata = mutations.get(0).getMetadata(); if (mutationBuffer.remainingCapacity() == 0) { metrics.bufferFull(metadata); } mutationBuffer.put(mutations); metrics.bufferSize(mutationBuffer.size(), metadata); stopwatch.stop(); final long time = stopwatch.elapsed(TimeUnit.MILLISECONDS); metrics.sendTime(time); } catch (Exception ex) { log.error("Failed to send mutations.", ex); metrics.sendFailed(ex); throw new DestinationException("Failed to send mutations", ex); } } }
BufferedDestination extends ListenableDestination { @Override public void send(@NonNull final List<? extends Mutation<?>> mutations) { try { if (mutations.isEmpty()) { return; } final Stopwatch stopwatch = Stopwatch.createStarted(); final Mutation.Metadata metadata = mutations.get(0).getMetadata(); if (mutationBuffer.remainingCapacity() == 0) { metrics.bufferFull(metadata); } mutationBuffer.put(mutations); metrics.bufferSize(mutationBuffer.size(), metadata); stopwatch.stop(); final long time = stopwatch.elapsed(TimeUnit.MILLISECONDS); metrics.sendTime(time); } catch (Exception ex) { log.error("Failed to send mutations.", ex); metrics.sendFailed(ex); throw new DestinationException("Failed to send mutations", ex); } } BufferedDestination(
      @NonNull final String name,
      @Min(1) final int bufferSize,
      @NonNull final Destination destination,
      @NonNull final DestinationMetrics metrics); }
BufferedDestination extends ListenableDestination { @Override public void send(@NonNull final List<? extends Mutation<?>> mutations) { try { if (mutations.isEmpty()) { return; } final Stopwatch stopwatch = Stopwatch.createStarted(); final Mutation.Metadata metadata = mutations.get(0).getMetadata(); if (mutationBuffer.remainingCapacity() == 0) { metrics.bufferFull(metadata); } mutationBuffer.put(mutations); metrics.bufferSize(mutationBuffer.size(), metadata); stopwatch.stop(); final long time = stopwatch.elapsed(TimeUnit.MILLISECONDS); metrics.sendTime(time); } catch (Exception ex) { log.error("Failed to send mutations.", ex); metrics.sendFailed(ex); throw new DestinationException("Failed to send mutations", ex); } } BufferedDestination(
      @NonNull final String name,
      @Min(1) final int bufferSize,
      @NonNull final Destination destination,
      @NonNull final DestinationMetrics metrics); int getRemainingCapacity(); @Override void send(@NonNull final List<? extends Mutation<?>> mutations); Mutation<?> getLastPublishedMutation(); synchronized boolean isRunning(); synchronized boolean isTerminated(); @Override synchronized boolean isStarted(); @Override void open(); @Override void close(); void clear(); }
BufferedDestination extends ListenableDestination { @Override public void send(@NonNull final List<? extends Mutation<?>> mutations) { try { if (mutations.isEmpty()) { return; } final Stopwatch stopwatch = Stopwatch.createStarted(); final Mutation.Metadata metadata = mutations.get(0).getMetadata(); if (mutationBuffer.remainingCapacity() == 0) { metrics.bufferFull(metadata); } mutationBuffer.put(mutations); metrics.bufferSize(mutationBuffer.size(), metadata); stopwatch.stop(); final long time = stopwatch.elapsed(TimeUnit.MILLISECONDS); metrics.sendTime(time); } catch (Exception ex) { log.error("Failed to send mutations.", ex); metrics.sendFailed(ex); throw new DestinationException("Failed to send mutations", ex); } } BufferedDestination(
      @NonNull final String name,
      @Min(1) final int bufferSize,
      @NonNull final Destination destination,
      @NonNull final DestinationMetrics metrics); int getRemainingCapacity(); @Override void send(@NonNull final List<? extends Mutation<?>> mutations); Mutation<?> getLastPublishedMutation(); synchronized boolean isRunning(); synchronized boolean isTerminated(); @Override synchronized boolean isStarted(); @Override void open(); @Override void close(); void clear();  }