class PartitionSelectionStrategy implements EndpointSelectionStrategy {

  @Override
  public <R extends Request<? extends Response>> Endpoint select(final R request, final List<Endpoint> endpoints) {
    int size = endpoints.size();
    if (size == 0) {
      return null;
    }

    short partition = ((KeyValueRequest<?>) request).partition();
    Endpoint endpoint = size == 1 ? endpoints.get(0) : endpoints.get(partition % size);
    if (endpoint != null && endpoint.state() == EndpointState.CONNECTED && endpoint.freeToWrite()) {
      return endpoint;
    }

    return null;
  }

}