@Test public void messages_fail_when_buffer_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(); }; Thread t = null; try (MessageBatcher<Integer, Integer> batcher = new MessageBatcher<>(4, 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(1)); assertThat(result2.get(), is(2)); assertThat(result3.get(), is(3)); assertThat(result4.get(), is(4)); try { result5.get(); fail(); } catch (ExecutionException e) { assertTrue(e.getCause() instanceof IllegalStateException); } } t.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();  }