@Test(timeout = 1000) public void batcher_batches_upto_maxBatchSize_under_pressure() throws Exception { List<List<Integer>> batches = new ArrayList<>(); Function<List<Integer>, List<Integer>> batchHandler = l -> { batches.add(l); return l.stream().map(i -> i * 10).collect(toList()); }; AtomicReference<Runnable> processQueue = new AtomicReference<>(); ThreadFactory factory = r -> { processQueue.set(r); return new Thread(); }; Thread t = null; try (MessageBatcher<Integer, Integer> batcher = new MessageBatcher<>(5, 2, EnqueueBehaviour.COMPLETE_EXCEPTIONALLY, batchHandler, factory)) { CompletableFuture<Integer> result1 = batcher.apply(1); CompletableFuture<Integer> result2 = batcher.apply(2); CompletableFuture<Integer> result3 = batcher.apply(3); CompletableFuture<Integer> result4 = batcher.apply(4); CompletableFuture<Integer> result5 = batcher.apply(5); t = new Thread(processQueue.get()); t.start(); assertThat(result1.get(), is(10)); assertThat(result2.get(), is(20)); assertThat(result3.get(), is(30)); assertThat(result4.get(), is(40)); assertThat(result5.get(), is(50)); } t.join(); assertThat(batches.size(), is(3)); assertThat(batches.get(0).size(), is(2)); assertThat(batches.get(0), contains(1, 2)); assertThat(batches.get(1).size(), is(2)); assertThat(batches.get(1), contains(3, 4)); assertThat(batches.get(2).size(), is(1)); assertThat(batches.get(2), contains(5)); }
@Override public CompletableFuture<RESPONSE> apply(MESSAGE t) { try { return enqueueBehaviour.enqueueMessage(queue, t); } catch (Exception e) { CompletableFuture<RESPONSE> callback = new CompletableFuture<>(); callback.completeExceptionally(e); return callback; } }
MessageBatcher implements CloseableFunction<MESSAGE, CompletableFuture<RESPONSE>> { @Override public CompletableFuture<RESPONSE> apply(MESSAGE t) { try { return enqueueBehaviour.enqueueMessage(queue, t); } catch (Exception e) { CompletableFuture<RESPONSE> callback = new CompletableFuture<>(); callback.completeExceptionally(e); return callback; } } }
MessageBatcher implements CloseableFunction<MESSAGE, CompletableFuture<RESPONSE>> { @Override public CompletableFuture<RESPONSE> apply(MESSAGE t) { try { return enqueueBehaviour.enqueueMessage(queue, t); } catch (Exception e) { CompletableFuture<RESPONSE> callback = new CompletableFuture<>(); callback.completeExceptionally(e); return callback; } } MessageBatcher(
      int bufferSize,
      int maxBatchSize,
      EnqueueBehaviour enqueueBehaviour,
      Function<List<MESSAGE>, List<RESPONSE>> batchHandler);   MessageBatcher(
      int bufferSize,
      int maxBatchSize,
      EnqueueBehaviour enqueueBehaviour,
      Function<List<MESSAGE>, List<RESPONSE>> batchHandler,
      ThreadFactory threadFactory); }
MessageBatcher implements CloseableFunction<MESSAGE, CompletableFuture<RESPONSE>> { @Override public CompletableFuture<RESPONSE> apply(MESSAGE t) { try { return enqueueBehaviour.enqueueMessage(queue, t); } catch (Exception e) { CompletableFuture<RESPONSE> callback = new CompletableFuture<>(); callback.completeExceptionally(e); return callback; } } MessageBatcher(
      int bufferSize,
      int maxBatchSize,
      EnqueueBehaviour enqueueBehaviour,
      Function<List<MESSAGE>, List<RESPONSE>> batchHandler);   MessageBatcher(
      int bufferSize,
      int maxBatchSize,
      EnqueueBehaviour enqueueBehaviour,
      Function<List<MESSAGE>, List<RESPONSE>> batchHandler,
      ThreadFactory threadFactory); @Override CompletableFuture<RESPONSE> apply(MESSAGE t); @Override void close(); }
MessageBatcher implements CloseableFunction<MESSAGE, CompletableFuture<RESPONSE>> { @Override public CompletableFuture<RESPONSE> apply(MESSAGE t) { try { return enqueueBehaviour.enqueueMessage(queue, t); } catch (Exception e) { CompletableFuture<RESPONSE> callback = new CompletableFuture<>(); callback.completeExceptionally(e); return callback; } } MessageBatcher(
      int bufferSize,
      int maxBatchSize,
      EnqueueBehaviour enqueueBehaviour,
      Function<List<MESSAGE>, List<RESPONSE>> batchHandler);   MessageBatcher(
      int bufferSize,
      int maxBatchSize,
      EnqueueBehaviour enqueueBehaviour,
      Function<List<MESSAGE>, List<RESPONSE>> batchHandler,
      ThreadFactory threadFactory); @Override CompletableFuture<RESPONSE> apply(MESSAGE t); @Override void close();  }