@Test public void batcher_passes_exceptions_up() throws Exception { Function<List<Integer>, List<Integer>> batchHandler = l -> { throw new RuntimeException("exception message"); }; try (MessageBatcher<Integer, Integer> batcher = new MessageBatcher<>(5, 2, EnqueueBehaviour.COMPLETE_EXCEPTIONALLY, batchHandler)) { batcher.apply(1).get(); fail("Shouldn't get here"); } catch (ExecutionException e) { assertThat(e.getCause(), instanceOf(RuntimeException.class)); assertThat(e.getCause().getMessage(), is("exception message")); } }
@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();  }