@Test public void itShouldSelectSubSenderConsideringHashCodeOfRequestData() { RpcClientConnectionPoolStub pool = new RpcClientConnectionPoolStub(); RpcSenderStub connection1 = new RpcSenderStub(); RpcSenderStub connection2 = new RpcSenderStub(); RpcSenderStub connection3 = new RpcSenderStub(); int shardsAmount = 3; ShardingFunction<Integer> shardingFunction = item -> item % shardsAmount; RpcStrategy shardingStrategy = sharding(shardingFunction, servers(ADDRESS_1, ADDRESS_2, ADDRESS_3)); RpcSender senderSharding; int timeout = 50; pool.put(ADDRESS_1, connection1); pool.put(ADDRESS_2, connection2); pool.put(ADDRESS_3, connection3); senderSharding = shardingStrategy.createSender(pool); senderSharding.sendRequest(0, timeout, assertNoCalls()); senderSharding.sendRequest(0, timeout, assertNoCalls()); senderSharding.sendRequest(1, timeout, assertNoCalls()); senderSharding.sendRequest(0, timeout, assertNoCalls()); senderSharding.sendRequest(2, timeout, assertNoCalls()); senderSharding.sendRequest(0, timeout, assertNoCalls()); senderSharding.sendRequest(0, timeout, assertNoCalls()); senderSharding.sendRequest(2, timeout, assertNoCalls()); assertEquals(5, connection1.getRequests()); assertEquals(1, connection2.getRequests()); assertEquals(2, connection3.getRequests()); }
@Nullable @Override public RpcSender createSender(RpcClientConnectionPool pool) { List<RpcSender> subSenders = list.listOfNullableSenders(pool); int activeSenders = 0; for (RpcSender subSender : subSenders) { if (subSender != null) { activeSenders++; } } if (activeSenders < minActiveSubStrategies) return null; if (subSenders.size() == 0) return null; if (subSenders.size() == 1) return subSenders.get(0); return new Sender(shardingFunction, subSenders); }
RpcStrategySharding implements RpcStrategy { @Nullable @Override public RpcSender createSender(RpcClientConnectionPool pool) { List<RpcSender> subSenders = list.listOfNullableSenders(pool); int activeSenders = 0; for (RpcSender subSender : subSenders) { if (subSender != null) { activeSenders++; } } if (activeSenders < minActiveSubStrategies) return null; if (subSenders.size() == 0) return null; if (subSenders.size() == 1) return subSenders.get(0); return new Sender(shardingFunction, subSenders); } }
RpcStrategySharding implements RpcStrategy { @Nullable @Override public RpcSender createSender(RpcClientConnectionPool pool) { List<RpcSender> subSenders = list.listOfNullableSenders(pool); int activeSenders = 0; for (RpcSender subSender : subSenders) { if (subSender != null) { activeSenders++; } } if (activeSenders < minActiveSubStrategies) return null; if (subSenders.size() == 0) return null; if (subSenders.size() == 1) return subSenders.get(0); return new Sender(shardingFunction, subSenders); } private  RpcStrategySharding(@NotNull ShardingFunction<?> shardingFunction, @NotNull RpcStrategyList list, int minActiveSubStrategies); }
RpcStrategySharding implements RpcStrategy { @Nullable @Override public RpcSender createSender(RpcClientConnectionPool pool) { List<RpcSender> subSenders = list.listOfNullableSenders(pool); int activeSenders = 0; for (RpcSender subSender : subSenders) { if (subSender != null) { activeSenders++; } } if (activeSenders < minActiveSubStrategies) return null; if (subSenders.size() == 0) return null; if (subSenders.size() == 1) return subSenders.get(0); return new Sender(shardingFunction, subSenders); } private  RpcStrategySharding(@NotNull ShardingFunction<?> shardingFunction, @NotNull RpcStrategyList list, int minActiveSubStrategies); static RpcStrategySharding create(ShardingFunction<?> shardingFunction, RpcStrategyList list); RpcStrategySharding withMinActiveSubStrategies(int minActiveSubStrategies); @Override Set<InetSocketAddress> getAddresses(); @Nullable @Override RpcSender createSender(RpcClientConnectionPool pool); }
RpcStrategySharding implements RpcStrategy { @Nullable @Override public RpcSender createSender(RpcClientConnectionPool pool) { List<RpcSender> subSenders = list.listOfNullableSenders(pool); int activeSenders = 0; for (RpcSender subSender : subSenders) { if (subSender != null) { activeSenders++; } } if (activeSenders < minActiveSubStrategies) return null; if (subSenders.size() == 0) return null; if (subSenders.size() == 1) return subSenders.get(0); return new Sender(shardingFunction, subSenders); } private  RpcStrategySharding(@NotNull ShardingFunction<?> shardingFunction, @NotNull RpcStrategyList list, int minActiveSubStrategies); static RpcStrategySharding create(ShardingFunction<?> shardingFunction, RpcStrategyList list); RpcStrategySharding withMinActiveSubStrategies(int minActiveSubStrategies); @Override Set<InetSocketAddress> getAddresses(); @Nullable @Override RpcSender createSender(RpcClientConnectionPool pool);  }