@Test(timeout = 1000) public void batcher_blocks_when_queue_is_full() throws Exception { Function<List<Integer>, List<Integer>> batchHandler = Function.identity(); AtomicReference<Runnable> processQueue = new AtomicReference<>(); ThreadFactory factory = r -> { processQueue.set(r); return new Thread(); }; CountDownLatch latch = new CountDownLatch(1); Thread thread = null; try (MessageBatcher<Integer, Integer> batcher = new MessageBatcher<>(4, 2, EnqueueBehaviour.BLOCK_AND_WAIT, batchHandler, factory)) { thread = new Thread(() -> { batcher.apply(1); batcher.apply(2); batcher.apply(3); batcher.apply(4); latch.countDown(); batcher.apply(5); }); thread.start(); latch.await(); while (thread.getState() != State.WAITING) {} } finally { if (thread != null) { thread.interrupt(); thread.join(); } } }
@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();  }