MessagePartitioner implements CloseableFunction<MESSAGE, RESPONSE> { @Override public RESPONSE apply(MESSAGE message) { int partitionNumber = Math.abs(hasherSupplier.get().apply(message)) % delegates.size(); return delegates.get(partitionNumber).apply(message); } MessagePartitioner(
      @NonNull Supplier<Function<MESSAGE, Integer>> hasherSupplier,
      @NonNull List<Function<MESSAGE, RESPONSE>> delegates); @Override RESPONSE apply(MESSAGE message); @Override void close();  }
@Test public void partitionTest() throws Exception { Supplier<Function<Integer, Integer>> hasherSupplier = () -> i -> i % 10; List<Integer> partition1 = new ArrayList<>(); List<Integer> partition2 = new ArrayList<>(); Function<Integer, Integer> client1 = i -> { partition1.add(i); return i * 2; }; Function<Integer, Integer> client2 = i -> { partition2.add(i); return i * 2; }; List<Function<Integer, Integer>> delegates = Arrays.asList(client1, client2); try (MessagePartitioner<Integer, Integer> partitioner = new MessagePartitioner<>(hasherSupplier, delegates)) { for (int i = 0; i < 20; i++) { assertThat(partitioner.apply(i), is(i * 2)); } assertThat(partition1.size() + partition2.size(), is(20)); for (int i = 0; i < 10; i++) { if (partition1.contains(i)) { assertThat(partition1.contains(i + 10), is(true)); assertThat(partition2.contains(i), is(false)); assertThat(partition2.contains(i + 10), is(false)); } if (partition2.contains(i)) { assertThat(partition2.contains(i + 10), is(true)); assertThat(partition1.contains(i), is(false)); assertThat(partition1.contains(i + 10), is(false)); } } } }
@Test public void hasherThrowsExceptionWhichIsPropogated() throws Exception { Supplier<Function<Integer, Integer>> hasherSupplier = () -> i -> { throw new RuntimeException(Integer.toString(i)); }; List<Function<Integer, Integer>> delegates = Arrays.asList(Function.identity()); try (MessagePartitioner<Integer, Integer> partitioner = new MessagePartitioner<>(hasherSupplier, delegates)) { try { partitioner.apply(1); fail("Should throw exception before reaching this point"); } catch (RuntimeException e) { assertThat(e.getMessage(), is("1")); } } }
@Test public void delegateExceptionIsPropogated() throws Exception { List<Function<Integer, Integer>> delegates = Arrays.asList(i -> { throw new RuntimeException(Integer.toString(i)); }); try (MessagePartitioner<Integer, Integer> partitioner = new MessagePartitioner<>(() -> i -> 1, delegates)) { try { partitioner.apply(1); fail("Should throw exception before reaching this point"); } catch (RuntimeException e) { assertThat(e.getMessage(), is("1")); } } }