public class MemcacheConfig implements MemcacheKeys {
    private static final Logger log = Logger.getLogger(MemcacheConfig.class);
    protected static final int DEFAULT_POINTS_PER_SERVER = 160;
    protected static final long DEFAULT_RETRY_INTERVAL = 50;
    protected static final String VALID_SERVER = "[a-zA-Z0-9-.]+(:\\d+){0,2}$";

    private LinkedList<String> servers = new LinkedList<>();
    private String eventBusAddress;
    private String namespace;
    private ContinuumType continuum;
    private HashAlgorithm algorithm;
    private int pointsPerServer = DEFAULT_POINTS_PER_SERVER;
    private long retryInterval = DEFAULT_RETRY_INTERVAL;

    public MemcacheConfig(JsonObject jsonConfig) {
        if (jsonConfig == null) {
            log.error("initialize", "exception", "noConfigFound");
            throw new MemcacheException("No Memcache config found");
        }

        if (jsonConfig.getJsonArray(SERVERS_KEY) != null && jsonConfig.getString(EVENT_BUS_ADDRESS_KEY) != null &&
                !jsonConfig.getString(EVENT_BUS_ADDRESS_KEY).isEmpty()) {
            this.servers.addAll(processServers(jsonConfig.getJsonArray(SERVERS_KEY)));
            this.eventBusAddress = jsonConfig.getString(EVENT_BUS_ADDRESS_KEY);
            this.namespace = jsonConfig.getString(NAMESPACE_KEY);
            this.pointsPerServer = jsonConfig.getInteger(POINTS_PER_SERVER, DEFAULT_POINTS_PER_SERVER);
            this.retryInterval = jsonConfig.getLong(RETRY_INTERVAL, DEFAULT_RETRY_INTERVAL);

            final HashAlgorithm defaultHashAlgorithm = HashAlgorithm.FNV1_32_HASH;
            String algorithmStr = jsonConfig.getString(ALGORITHM_KEY, defaultHashAlgorithm.name());
            this.algorithm = algorithmStr == null ? defaultHashAlgorithm : HashAlgorithm.valueOf(algorithmStr);

            final ContinuumType defaultContinuumType = ContinuumType.KETAMA;
            String continuumStr = jsonConfig.getString(CONTINUUM_KEY, defaultContinuumType.name());
            this.continuum = continuumStr == null ? defaultContinuumType : ContinuumType.valueOf(continuumStr);
        } else {
            log.error("initialize", "exception", "invalidConfigFound", new String[] {"config"}, jsonConfig.encode());
            throw new MemcacheException("Invalid Memcache config defined");
        }

        log.info("initialize", "success", new String[]{"eventBusAddress", "namespace", "servers", "algorithm"}, eventBusAddress,
                namespace, servers.size(), algorithm);
    }

    public Collection<String> getServers() {
        return servers;
    }

    public String getEventBusAddress() {
        return eventBusAddress;
    }

    public String getNamespace() {
        return namespace;
    }

    public HashAlgorithm getHashAlgorithm() {
        return algorithm;
    }

    public ContinuumType getContinuumType() {
        return continuum;
    }

    public int getPointsPerServer() {
        return pointsPerServer;
    }

    public long getRetryInterval() {
        return retryInterval;
    }

    /**
     * Process a JsonArray of server strings and return a collection of the valid entries.  Valid server strings should
     * be in the format [hostname]:[port].
     *
     * @param serverStrings
     * @return A Collection of valid servers.
     */
    private Collection<String> processServers(JsonArray serverStrings) {
        LinkedList<String> validServers = new LinkedList<>();
        for (Object server : serverStrings) {
            if (server == null || !Pattern.matches(VALID_SERVER, (String) server)) {
                log.warn("processServers", "invalidServer", new String[]{"server"}, server);
                continue;
            }

            validServers.add((String) server);
        }

        return validServers;
    }
} public final class DeleteCommandResponse extends MemcacheCommandResponse {
    private final String data;

    private DeleteCommandResponse(Builder builder) {
        super(builder);
        data = builder.data;
    }

    public String getData() {
        return data;
    }

    /**
     * Builder for the DeleteCommandResponse
     */
    public static class Builder extends AbstractBuilder<Builder, DeleteCommandResponse> {
        private String data;

        @Override
        protected Builder self() {
            return this;
        }

        public Builder setData(String value) {
            data = value;
            return self();
        }

        @Override
        public DeleteCommandResponse build() {
            return new DeleteCommandResponse(this);
        }
    }
} public final class ModifyCommandResponse extends MemcacheCommandResponse {
    private final Integer data;

    private ModifyCommandResponse(Builder builder) {
        super(builder);
        data = builder.data;
    }

    public Integer getData() {
        return data;
    }

    /**
     * Builder for the ModifyCommandResponse
     */
    public static class Builder extends AbstractBuilder<Builder, ModifyCommandResponse> {
        private Integer data;

        @Override
        protected Builder self() {
            return this;
        }

        public Builder setData(Integer value) {
            data = value;
            return self();
        }

        @Override
        public ModifyCommandResponse build() {
            return new ModifyCommandResponse(this);
        }
    }
} public final class RetrieveCommandResponse extends MemcacheCommandResponse {
    private final Map<String, String> data;

    private RetrieveCommandResponse(Builder builder) {
        super(builder);
        data = Collections.unmodifiableMap(new HashMap<>(builder.data));
    }

    public Map<String, String> getData() {
        return data;
    }

    /**
     * Builder for the RetrieveCommandResponse
     */
    public static class Builder extends AbstractBuilder<Builder, RetrieveCommandResponse> {
        private Map<String, String> data = new HashMap<>();

        @Override
        protected Builder self() {
            return this;
        }

        public Builder setData(Map<String, String> value) {
            if (value == null) {
                data = new HashMap<>();
            } else {
                data = value;
            }
            return self();
        }

        public Builder addData(String key, String value) {
            data.put(key, value);
            return self();
        }

        @Override
        public RetrieveCommandResponse build() {
            return new RetrieveCommandResponse(this);
        }
    }
} public final class StoreCommandResponse extends MemcacheCommandResponse {
    private final String data;

    private StoreCommandResponse(Builder builder) {
        super(builder);
        data = builder.data;
    }

    public String getData() {
        return data;
    }

    /**
     * Builder for the StoreCommandResponse
     */
    public static class Builder extends AbstractBuilder<Builder, StoreCommandResponse> {
        private String data;

        @Override
        protected Builder self() {
            return this;
        }

        public Builder setData(String value) {
            data = value;
            return self();
        }

        @Override
        public StoreCommandResponse build() {
            return new StoreCommandResponse(this);
        }
    }
} public final class TouchCommandResponse extends MemcacheCommandResponse {
    private final String data;

    private TouchCommandResponse(Builder builder) {
        super(builder);
        data = builder.data;
    }

    public String getData() {
        return data;
    }

    /**
     * Builder for the TouchCommandResponse
     */
    public static class Builder extends AbstractBuilder<Builder, TouchCommandResponse> {
        private String data;

        @Override
        protected Builder self() {
            return this;
        }

        public Builder setData(String value) {
            data = value;
            return self();
        }

        @Override
        public TouchCommandResponse build() {
            return new TouchCommandResponse(this);
        }
    }
} public class MemcacheCommand {
    private static final Logger log = Logger.getLogger(MemcacheCommand.class);
    private MemcacheCommandType type;
    private String key;
    private String value;
    private Integer expires;
    private LineParser parser;
    private Handler<MemcacheCommandResponse> commandResponseHandler;

    /**
     * A helper method for manually building a command.
     *
     * @param type - An enum for the command type.
     * @param key - A String containing the key.
     * @param value - The String value to be sent using the command.
     * @param expires - An integer with the seconds before expiration.
     */
    public MemcacheCommand(MemcacheCommandType type, String key, String value, Integer expires) {
        if (type == null || key == null) {
            log.warn("initMemcacheCommand", "failure", new String[]{"reason"}, "Invalid command format");
            throw new IllegalArgumentException("Invalid command format");
        }

        setLineParser(type.getLineParserType());

        this.type = type;
        this.key = key;
        this.value = value;
        this.expires = expires;
    }

    /**
     * The MemcacheCommandType enum which represents the command type being sent to Memcache.
     *
     * @return - A MemcacheCommandType representing the current command.
     */
    public MemcacheCommandType getType() {
        return type;
    }

    /**
     * The String name for the command.
     *
     * @return - A String containing the name for the Memcache command.
     */
    public String getCommand() {
        return type.getCommand();
    }

    /**
     * The key for inserting, updating, or deleting an object in the store.
     *
     * @return - A String containing the key.
     */
    public String getKey() {
        return key;
    }

    /**
     * The value to be stored or updated in the store.
     *
     * @return - A String containing the value.
     */
    public String getValue() {
        return value;
    }

    /**
     * The number of seconds for updating the expiration of a key.
     *
     * @return - An integer representing the number of seconds before a key expires.
     */
    public Integer getExpires() {
        return expires;
    }

    /**
     * Calling this method will execute the handler associate with this command.  If no
     * handler is specified the response will be ignored.
     *
     * @param response - The Memcache response for this command.
     */
    public void setResponse(MemcacheCommandResponse response) {
        if (commandResponseHandler != null) {
            commandResponseHandler.handle(response);
        } else {
            log.trace("setResponse", "missingHandler");
        }
    }

    /**
     * This handler will be executed when the response has been received from Memcache.
     *
     * @param handler - A handler for the JsonObject Memcache response.
     */
    public void commandResponseHandler(Handler<MemcacheCommandResponse> handler) {
        this.commandResponseHandler = handler;
    }

    /**
     * This is the parser for processing the current response for this command.
     *
     * @return - The currently initialized line parser.
     */
    public LineParser getLineParser() {
        return parser;
    }

    /**
     * This initializes the line parser for processing the response for this command.
     *
     * @param lineParserType - The type of line parser to create.
     */
    private void setLineParser(LineParserType lineParserType) {
        switch (lineParserType) {
            case RETRIEVE:
                parser = new RetrieveLineParser();
                break;
            case STORE:
                parser = new StoreLineParser();
                break;
            case MODIFY:
                parser = new ModifyLineParser();
                break;
            case DELETE:
                parser = new DeleteLineParser();
                break;
            case TOUCH:
                parser = new TouchLineParser();
                break;
            default:
                throw new MemcacheException("Unable to initialize line parser.");
        }
    }
} public enum MemcacheCommandType {
    set(LineParserType.STORE),
    add(LineParserType.STORE),
    replace(LineParserType.STORE),
    append(LineParserType.STORE),
    prepend(LineParserType.STORE),
    get(LineParserType.RETRIEVE),
    delete(LineParserType.DELETE),
    touch(LineParserType.TOUCH),
    incr(LineParserType.MODIFY),
    decr(LineParserType.MODIFY);


    private final String command;
    private final LineParserType lineParserType;

    MemcacheCommandType(LineParserType lineParserType) {
        this(null, lineParserType);
    }

    MemcacheCommandType(String command, LineParserType lineParserType) {
        this.command = command == null ? this.name() : command;
        this.lineParserType = lineParserType;
    }

    public String getCommand() {
        return command;
    }

    public LineParserType getLineParserType() {
        return lineParserType;
    }
} public abstract class Continuum {
    private static final Logger log = Logger.getLogger(Continuum.class);
    public static final Charset ENCODING = Charset.forName("UTF-8");

    private final TreeMap<Long, MemcacheServer> servers = new TreeMap<>();
    private final HashAlgorithm hashAlgorithm;

    protected Continuum(HashAlgorithm hashAlgorithm) {
        if (hashAlgorithm == null) {
            throw new MemcacheException("Invalid hash algorithm provided");
        }
        this.hashAlgorithm = hashAlgorithm;
    }

    public MemcacheServer getServer(String key) {
        if (servers.size() == 1) {
            log.debug("getServer", "singleServer");
            return servers.firstEntry().getValue();
        } else {
            long hash = hashAlgorithm.hash(key);
            log.debug("getServer", "hashingKey", new String[] {"alorithm", "hash"}, hashAlgorithm.name(), hash);
            Map.Entry<Long, MemcacheServer> serverEntry = servers.ceilingEntry(hash);
            if (serverEntry == null) {
                return servers.firstEntry().getValue();
            }
            return serverEntry.getValue();
        }
    }

    protected int getServerEntryCount(MemcacheServer server, int pointsPerServer, int totalServers, int totalWeight) {
        return (int) (((double) (totalServers * pointsPerServer * server.getWeight())) / (double) totalWeight);
    }

    protected TreeMap<Long, MemcacheServer> getServerContinuum() {
        return servers;
    }

    protected abstract void addServerEntry(MemcacheServer server, int index);

    protected void addServerEntries(MemcacheServer server, int entries) {
        for (int i = 0; i < entries; i++) {
            addServerEntry(server, i);
        }
    }
} public final class ContinuumFactory {
    private ContinuumFactory() { }

    public static Continuum create(MemcacheConfig config) {
        ArrayList<MemcacheServer> memcacheServerList = new ArrayList<>();
        for (String server : config.getServers()) {
            MemcacheServer memcacheServer = new MemcacheServer(server);
            memcacheServerList.add(memcacheServer);
        }

        if (memcacheServerList.size() < 1) {
            throw new IllegalArgumentException("No servers defined");
        }

        Continuum result;

        switch (config.getContinuumType()) {
            case KETAMA:
                result = new KetamaContinuum(memcacheServerList, config.getHashAlgorithm(), config.getPointsPerServer());
                break;
            default:
                result = new DefaultContinuum(memcacheServerList, config.getHashAlgorithm(), config.getPointsPerServer());
                break;
        }

        return result;
    }
} public class MemcacheServer {
    private static final int SERVER_PARTS = 3;
    public static final int DEFAULT_PORT = 11211;
    public static final int DEFAULT_WEIGHT = 8;

    private String host;
    private int port = DEFAULT_PORT;
    private int weight = DEFAULT_WEIGHT;
    private String server;

    public MemcacheServer(String server) {
        this.server = server;

        String[] parts = server.split(":", SERVER_PARTS);
        host = parts[0];
        if (parts.length > 1) {
            port = Integer.parseInt(parts[1]);
        }
        if (parts.length > 2) {
            weight = Integer.parseInt(parts[2]);
        }
    }

    public String getServer() {
        return server;
    }

    public String getHost() {
        return host;
    }

    public int getPort() {
        return port;
    }

    public int getWeight() {
        return weight;
    }
}
import com.groupon.vertx.memcache.server.ContinuumFactory;
import com.groupon.vertx.memcache.server.MemcacheServer;
import com.groupon.vertx.utils.Logger;
import java.util.Collection;
import io.vertx.core.Future;
import io.vertx.core.eventbus.DeliveryOptions;
import io.vertx.core.eventbus.EventBus;
import com.groupon.vertx.memcache.MemcacheConfig;
import com.groupon.vertx.memcache.client.response.DeleteCommandResponse;
import com.groupon.vertx.memcache.client.response.ModifyCommandResponse;
import com.groupon.vertx.memcache.client.response.RetrieveCommandResponse;
import com.groupon.vertx.memcache.client.response.StoreCommandResponse;
import com.groupon.vertx.memcache.client.response.TouchCommandResponse;
import com.groupon.vertx.memcache.command.MemcacheCommand;
import com.groupon.vertx.memcache.command.MemcacheCommandType;
import com.groupon.vertx.memcache.server.Continuum;
/**
 * Copyright 2014 Groupon.com
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.groupon.vertx.memcache.client;




/**
 * This is a helper class for constructing the eventbus messages to execute the memcache commands.
 *
 * @author Stuart Siegrist (fsiegrist at groupon dot com)
 * @since 1.0.0
 */
public class MemcacheClient {
    private static final Logger log = Logger.getLogger(MemcacheClient.class);
    private static final long INFINITE_REPLY_TIMEOUT = Long.MAX_VALUE;
    private Continuum continuum;
    private EventBus eventBus;
    private String eventBusAddress;
    private String namespace;

    public MemcacheClient(EventBus eventBus, MemcacheConfig config) {
        this.eventBus = eventBus;
        this.eventBusAddress = config.getEventBusAddress();
        this.namespace = config.getNamespace();

        log.info("initialize", "createContinuum", new String[]{"servers", "pointsPerServer"}, config.getServers().size(),
                config.getPointsPerServer());
        continuum = ContinuumFactory.create(config);
    }

    public Future<ModifyCommandResponse> incr(String key, long value) {
        return modify(MemcacheCommandType.incr, key, String.valueOf(value));
    }

    public Future<ModifyCommandResponse> decr(String key, long value) {
        return modify(MemcacheCommandType.decr, key, String.valueOf(value));
    }

    public Future<StoreCommandResponse> set(String key, String data, int expires) {
        return store(MemcacheCommandType.set, key, data, expires);
    }

    public Future<StoreCommandResponse> add(String key, String data, int expires) {
        return store(MemcacheCommandType.add, key, data, expires);
    }

    public Future<StoreCommandResponse> replace(String key, String data, int expires) {
        return store(MemcacheCommandType.replace, key, data, expires);
    }

    public Future<ModifyCommandResponse> append(String key, String data) {
        return modify(MemcacheCommandType.append, key, data);
    }

    public Future<ModifyCommandResponse> prepend(String key, String data) {
        return modify(MemcacheCommandType.prepend, key, data);
    }

    public Future<RetrieveCommandResponse> get(String key) {
        return retrieve(MemcacheCommandType.get, key);
    }

    public Future<RetrieveCommandResponse> get(Collection<String> keys) {
        Future<RetrieveCommandResponse> finalResult = Future.future();

        MemcacheClientMultiResponseHandler handleWrapper = new MemcacheClientMultiResponseHandler(finalResult, keys.size());
        for (String key : keys) {
            retrieve(MemcacheCommandType.get, key).setHandler(handleWrapper);
        }

        return finalResult;
    }

    public Future<DeleteCommandResponse> delete(String key) {
        Future<DeleteCommandResponse> finalResult = Future.future();

        MemcacheCommand command = new MemcacheCommand(MemcacheCommandType.delete, getCacheKey(key), null, null);