@Test public void canAcceptOversizeMessages() { NoopBatchWrite x = new NoopBatchWrite(0, 0, Duration.ofMillis(0)); CompletableFuture.allOf(x.apply("x"), x.apply("x")).join(); assertEquals(2, x.batchCount); }
@Override public CompletableFuture<Void> apply(InputT input) { AtomicReference<CompletableFuture<Void>> output = new AtomicReference<>(); EncodedT encodedInput = encodeInput(input); batches.compute(getBatchKey(input), (batchKey, batch) -> { if (batch != null) { batch.add(encodedInput).ifPresent(output::set); } if (output.get() == null) { batch = getBatch(batchKey); batch.add(encodedInput).ifPresent(output::set); batch.init.complete(null); } return batch; }); return Optional.ofNullable(output.get()) .orElseThrow(() -> new IllegalArgumentException("Empty batch rejected input")); }
BatchWrite implements Function<InputT, CompletableFuture<Void>> { @Override public CompletableFuture<Void> apply(InputT input) { AtomicReference<CompletableFuture<Void>> output = new AtomicReference<>(); EncodedT encodedInput = encodeInput(input); batches.compute(getBatchKey(input), (batchKey, batch) -> { if (batch != null) { batch.add(encodedInput).ifPresent(output::set); } if (output.get() == null) { batch = getBatch(batchKey); batch.add(encodedInput).ifPresent(output::set); batch.init.complete(null); } return batch; }); return Optional.ofNullable(output.get()) .orElseThrow(() -> new IllegalArgumentException("Empty batch rejected input")); } }
BatchWrite implements Function<InputT, CompletableFuture<Void>> { @Override public CompletableFuture<Void> apply(InputT input) { AtomicReference<CompletableFuture<Void>> output = new AtomicReference<>(); EncodedT encodedInput = encodeInput(input); batches.compute(getBatchKey(input), (batchKey, batch) -> { if (batch != null) { batch.add(encodedInput).ifPresent(output::set); } if (output.get() == null) { batch = getBatch(batchKey); batch.add(encodedInput).ifPresent(output::set); batch.init.complete(null); } return batch; }); return Optional.ofNullable(output.get()) .orElseThrow(() -> new IllegalArgumentException("Empty batch rejected input")); } BatchWrite(long maxBytes, int maxMessages, Duration maxDelay,
      PubsubMessageToTemplatedString batchKeyTemplate, Executor executor); }
BatchWrite implements Function<InputT, CompletableFuture<Void>> { @Override public CompletableFuture<Void> apply(InputT input) { AtomicReference<CompletableFuture<Void>> output = new AtomicReference<>(); EncodedT encodedInput = encodeInput(input); batches.compute(getBatchKey(input), (batchKey, batch) -> { if (batch != null) { batch.add(encodedInput).ifPresent(output::set); } if (output.get() == null) { batch = getBatch(batchKey); batch.add(encodedInput).ifPresent(output::set); batch.init.complete(null); } return batch; }); return Optional.ofNullable(output.get()) .orElseThrow(() -> new IllegalArgumentException("Empty batch rejected input")); } BatchWrite(long maxBytes, int maxMessages, Duration maxDelay,
      PubsubMessageToTemplatedString batchKeyTemplate, Executor executor); Function<InputT, CompletableFuture<Void>> withOpenCensusMetrics(); @Override CompletableFuture<Void> apply(InputT input); }
BatchWrite implements Function<InputT, CompletableFuture<Void>> { @Override public CompletableFuture<Void> apply(InputT input) { AtomicReference<CompletableFuture<Void>> output = new AtomicReference<>(); EncodedT encodedInput = encodeInput(input); batches.compute(getBatchKey(input), (batchKey, batch) -> { if (batch != null) { batch.add(encodedInput).ifPresent(output::set); } if (output.get() == null) { batch = getBatch(batchKey); batch.add(encodedInput).ifPresent(output::set); batch.init.complete(null); } return batch; }); return Optional.ofNullable(output.get()) .orElseThrow(() -> new IllegalArgumentException("Empty batch rejected input")); } BatchWrite(long maxBytes, int maxMessages, Duration maxDelay,
      PubsubMessageToTemplatedString batchKeyTemplate, Executor executor); Function<InputT, CompletableFuture<Void>> withOpenCensusMetrics(); @Override CompletableFuture<Void> apply(InputT input); @VisibleForTesting
  final ConcurrentMap<BatchKeyT, Batch> batches; }