@Test public void itShouldNotBeCreatedWhenNoSendersWereAdded() { RpcClientConnectionPoolStub pool = new RpcClientConnectionPoolStub(); HashFunction<Object> hashFunction = new RpcMessageDataStubWithKeyHashFunction(); RpcStrategy rendezvousHashing = rendezvousHashing(hashFunction); assertNull(rendezvousHashing.createSender(pool)); }
@Override @Nullable public RpcSender createSender(RpcClientConnectionPool pool) { Map<Object, RpcSender> shardsSenders = new HashMap<>(); for (Map.Entry<Object, RpcStrategy> entry : shards.entrySet()) { Object shardId = entry.getKey(); RpcStrategy strategy = entry.getValue(); RpcSender sender = strategy.createSender(pool); if (sender != null) { shardsSenders.put(shardId, sender); } } if (shardsSenders.size() < minShards) { return null; } if (shardsSenders.size() == 1) { return first(shardsSenders.values()); } RpcSender[] sendersBuckets = new RpcSender[buckets]; for (int n = 0; n < sendersBuckets.length; n++) { RpcSender chosenSender = null; int max = Integer.MIN_VALUE; for (Map.Entry<Object, RpcSender> entry : shardsSenders.entrySet()) { Object key = entry.getKey(); RpcSender sender = entry.getValue(); int hash = hashBucketFunction.hash(key, n); if (hash >= max) { chosenSender = sender; max = hash; } } assert chosenSender != null; sendersBuckets[n] = chosenSender; } return new Sender(hashFunction, sendersBuckets); }
RpcStrategyRendezvousHashing implements RpcStrategy { @Override @Nullable public RpcSender createSender(RpcClientConnectionPool pool) { Map<Object, RpcSender> shardsSenders = new HashMap<>(); for (Map.Entry<Object, RpcStrategy> entry : shards.entrySet()) { Object shardId = entry.getKey(); RpcStrategy strategy = entry.getValue(); RpcSender sender = strategy.createSender(pool); if (sender != null) { shardsSenders.put(shardId, sender); } } if (shardsSenders.size() < minShards) { return null; } if (shardsSenders.size() == 1) { return first(shardsSenders.values()); } RpcSender[] sendersBuckets = new RpcSender[buckets]; for (int n = 0; n < sendersBuckets.length; n++) { RpcSender chosenSender = null; int max = Integer.MIN_VALUE; for (Map.Entry<Object, RpcSender> entry : shardsSenders.entrySet()) { Object key = entry.getKey(); RpcSender sender = entry.getValue(); int hash = hashBucketFunction.hash(key, n); if (hash >= max) { chosenSender = sender; max = hash; } } assert chosenSender != null; sendersBuckets[n] = chosenSender; } return new Sender(hashFunction, sendersBuckets); } }
RpcStrategyRendezvousHashing implements RpcStrategy { @Override @Nullable public RpcSender createSender(RpcClientConnectionPool pool) { Map<Object, RpcSender> shardsSenders = new HashMap<>(); for (Map.Entry<Object, RpcStrategy> entry : shards.entrySet()) { Object shardId = entry.getKey(); RpcStrategy strategy = entry.getValue(); RpcSender sender = strategy.createSender(pool); if (sender != null) { shardsSenders.put(shardId, sender); } } if (shardsSenders.size() < minShards) { return null; } if (shardsSenders.size() == 1) { return first(shardsSenders.values()); } RpcSender[] sendersBuckets = new RpcSender[buckets]; for (int n = 0; n < sendersBuckets.length; n++) { RpcSender chosenSender = null; int max = Integer.MIN_VALUE; for (Map.Entry<Object, RpcSender> entry : shardsSenders.entrySet()) { Object key = entry.getKey(); RpcSender sender = entry.getValue(); int hash = hashBucketFunction.hash(key, n); if (hash >= max) { chosenSender = sender; max = hash; } } assert chosenSender != null; sendersBuckets[n] = chosenSender; } return new Sender(hashFunction, sendersBuckets); } private  RpcStrategyRendezvousHashing(@NotNull HashFunction<?> hashFunction, int minShards,
			@NotNull HashBucketFunction hashBucketFunction, int buckets,
			Map<Object, RpcStrategy> shards); }
RpcStrategyRendezvousHashing implements RpcStrategy { @Override @Nullable public RpcSender createSender(RpcClientConnectionPool pool) { Map<Object, RpcSender> shardsSenders = new HashMap<>(); for (Map.Entry<Object, RpcStrategy> entry : shards.entrySet()) { Object shardId = entry.getKey(); RpcStrategy strategy = entry.getValue(); RpcSender sender = strategy.createSender(pool); if (sender != null) { shardsSenders.put(shardId, sender); } } if (shardsSenders.size() < minShards) { return null; } if (shardsSenders.size() == 1) { return first(shardsSenders.values()); } RpcSender[] sendersBuckets = new RpcSender[buckets]; for (int n = 0; n < sendersBuckets.length; n++) { RpcSender chosenSender = null; int max = Integer.MIN_VALUE; for (Map.Entry<Object, RpcSender> entry : shardsSenders.entrySet()) { Object key = entry.getKey(); RpcSender sender = entry.getValue(); int hash = hashBucketFunction.hash(key, n); if (hash >= max) { chosenSender = sender; max = hash; } } assert chosenSender != null; sendersBuckets[n] = chosenSender; } return new Sender(hashFunction, sendersBuckets); } private  RpcStrategyRendezvousHashing(@NotNull HashFunction<?> hashFunction, int minShards,
			@NotNull HashBucketFunction hashBucketFunction, int buckets,
			Map<Object, RpcStrategy> shards); static RpcStrategyRendezvousHashing create(HashFunction<?> hashFunction); RpcStrategyRendezvousHashing withMinActiveShards(int minShards); RpcStrategyRendezvousHashing withHashBucketFunction(HashBucketFunction hashBucketFunction); RpcStrategyRendezvousHashing withHashBuckets(int buckets); RpcStrategyRendezvousHashing withShard(Object shardId, @NotNull RpcStrategy strategy); RpcStrategyRendezvousHashing withShards(InetSocketAddress... addresses); RpcStrategyRendezvousHashing withShards(List<InetSocketAddress> addresses); @Override Set<InetSocketAddress> getAddresses(); @Override @Nullable RpcSender createSender(RpcClientConnectionPool pool); }
RpcStrategyRendezvousHashing implements RpcStrategy { @Override @Nullable public RpcSender createSender(RpcClientConnectionPool pool) { Map<Object, RpcSender> shardsSenders = new HashMap<>(); for (Map.Entry<Object, RpcStrategy> entry : shards.entrySet()) { Object shardId = entry.getKey(); RpcStrategy strategy = entry.getValue(); RpcSender sender = strategy.createSender(pool); if (sender != null) { shardsSenders.put(shardId, sender); } } if (shardsSenders.size() < minShards) { return null; } if (shardsSenders.size() == 1) { return first(shardsSenders.values()); } RpcSender[] sendersBuckets = new RpcSender[buckets]; for (int n = 0; n < sendersBuckets.length; n++) { RpcSender chosenSender = null; int max = Integer.MIN_VALUE; for (Map.Entry<Object, RpcSender> entry : shardsSenders.entrySet()) { Object key = entry.getKey(); RpcSender sender = entry.getValue(); int hash = hashBucketFunction.hash(key, n); if (hash >= max) { chosenSender = sender; max = hash; } } assert chosenSender != null; sendersBuckets[n] = chosenSender; } return new Sender(hashFunction, sendersBuckets); } private  RpcStrategyRendezvousHashing(@NotNull HashFunction<?> hashFunction, int minShards,
			@NotNull HashBucketFunction hashBucketFunction, int buckets,
			Map<Object, RpcStrategy> shards); static RpcStrategyRendezvousHashing create(HashFunction<?> hashFunction); RpcStrategyRendezvousHashing withMinActiveShards(int minShards); RpcStrategyRendezvousHashing withHashBucketFunction(HashBucketFunction hashBucketFunction); RpcStrategyRendezvousHashing withHashBuckets(int buckets); RpcStrategyRendezvousHashing withShard(Object shardId, @NotNull RpcStrategy strategy); RpcStrategyRendezvousHashing withShards(InetSocketAddress... addresses); RpcStrategyRendezvousHashing withShards(List<InetSocketAddress> addresses); @Override Set<InetSocketAddress> getAddresses(); @Override @Nullable RpcSender createSender(RpcClientConnectionPool pool);  }