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 interface MemcacheKeys {
    String MEMCACHE_CLUSTER_KEY = "memcacheClusterConfig";
    String CLUSTERS_KEY = "clusters";
    String MEMCACHE_KEY = "memcacheConfig";
    String SERVERS_KEY = "servers";
    String EVENT_BUS_ADDRESS_KEY = "eventBusAddress";
    String EVENT_BUS_ADDRESS_PREFIX_KEY = "eventBusAddressPrefix";
    String NAMESPACE_KEY = "namespace";
    String POINTS_PER_SERVER = "pointsPerServer";
    String RETRY_INTERVAL = "retryInterval";
    String ALGORITHM_KEY = "algorithm";
    String CONTINUUM_KEY = "continuum";
} public class TestMessage<T> implements Message<T> {
    private String address;
    private T body;
    private Object reply;


    public TestMessage(T body) {
        this(null, body);
    }

    public TestMessage(String address, T body) {
        this.address = address;
        this.body = body;
    }

    @Override
    public String address() {
        return address;
    }

    @Override
    public MultiMap headers() {
        return null;
    }

    @Override
    public T body() {
        return body;
    }

    @Override
    public String replyAddress() {
        return address;
    }

    @Override
    public boolean isSend() {
        return false;
    }

    @Override
    public void reply(Object value) {
        this.reply = value;
    }

    @Override
    public <R> void reply(Object o, Handler<AsyncResult<Message<R>>> handler) {
        this.reply = o;
    }

    @Override
    public void reply(Object o, DeliveryOptions deliveryOptions) {
        this.reply = o;
    }

    @Override
    public <R> void reply(Object o, DeliveryOptions deliveryOptions, Handler<AsyncResult<Message<R>>> handler) {
        this.reply = o;
    }

    public Object getReply() {
        return reply;
    }

    @Override
    public void fail(int i, String s) {
    }
} 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 enum HashAlgorithm {
    NATIVE_HASH,
    CRC_HASH,
    FNV1_64_HASH,
    FNV1A_64_HASH,
    FNV1_32_HASH,
    FNV1A_32_HASH,
    KETAMA_HASH;

    private static final Charset ENCODING = Charset.forName("UTF-8");

    private static final long FNV_64_INIT = 0xcbf29ce484222325L;
    private static final long FNV_64_PRIME = 0x100000001b3L;

    private static final long FNV_32_INIT = 2166136261L;
    private static final long FNV_32_PRIME = 16777619;

    private static MessageDigest md5Digest = null;
    private static CRC32 crc32 = null;

    static {
        try {
            md5Digest = MessageDigest.getInstance("MD5");
        } catch (NoSuchAlgorithmException e) {
            throw new RuntimeException("MD5 not supported", e);
        }

        crc32 = new CRC32();
    }

    /**
     * Compute the hash for the given key.
     *
     * @param k - String key
     * @return a positive integer hash
     */
    public long hash(final String k) {
        long rv = 0;
        int len = k.length();
        // CS.OFF: MagicNumber
        switch (this) {
            case NATIVE_HASH:
                rv = k.hashCode();
                break;
            case FNV1_64_HASH:
                // Thanks to pierre@demartines.com for the pointer
                rv = FNV_64_INIT;
                for (int i = 0; i < len; i++) {
                    rv *= FNV_64_PRIME;
                    rv ^= k.charAt(i);
                }
                break;
            case FNV1A_64_HASH:
                rv = FNV_64_INIT;
                for (int i = 0; i < len; i++) {
                    rv ^= k.charAt(i);
                    rv *= FNV_64_PRIME;
                }
                break;
            case FNV1_32_HASH:
                rv = FNV_32_INIT;
                for (int i = 0; i < len; i++) {
                    rv *= FNV_32_PRIME;
                    rv ^= k.charAt(i);
                }
                break;
            case FNV1A_32_HASH:
                rv = FNV_32_INIT;
                for (int i = 0; i < len; i++) {
                    rv ^= k.charAt(i);
                    rv *= FNV_32_PRIME;
                }
                break;
            default:
                rv = hash(k.getBytes(ENCODING));
                break;
        }
        return rv & 0xffffffffL; /* Truncate to 32-bits */
        // CS.ON: MagicNumber
    }

    /**
     * Compute the hash for the given key.
     *
     * @param k - byte array key
     * @return a positive integer hash
     */
    private long hash(final byte[] k) {
        long rv = 0;
        // CS.OFF: MagicNumber
        switch (this) {
            case CRC_HASH:
                crc32.reset();
                crc32.update(k);
                rv = (crc32.getValue() >> 16) & 0x7fff;
                break;
            case KETAMA_HASH:
                byte[] bKey = md5Digest.digest(k);
                rv = ((long) (bKey[3] & 0xFF) << 24) | ((long) (bKey[2] & 0xFF) << 16) |
                        ((long) (bKey[1] & 0xFF) << 8) | (bKey[0] & 0xFF);
                break;
            default:
                assert false;
        }
        // CS.ON: MagicNumber
        return rv;
    }
}
import static org.junit.Assert.assertEquals;
import static org.mockito.Matchers.argThat;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import io.vertx.core.AsyncResult;
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.eventbus.DeliveryOptions;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import org.hamcrest.FeatureMatcher;
import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.mockito.internal.matchers.Equals;
import com.groupon.vertx.memcache.MemcacheConfig;
import com.groupon.vertx.memcache.MemcacheKeys;
import com.groupon.vertx.memcache.TestMessage;
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.hash.HashAlgorithm;
/**
 * 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;





/**
 * Tests for <code>MemcacheClient</code>.
 *
 * @author Stuart Siegrist (fsiegrist at groupon dot com)
 * @since 1.0.0
 */
public class MemcacheClientTest implements MemcacheKeys {

    @Mock
    private EventBus eventBus;

    @Captor
    private ArgumentCaptor<MemcacheClientResponseHandler<ModifyCommandResponse>> modifyCaptor;

    @Captor
    private ArgumentCaptor<MemcacheClientResponseHandler<StoreCommandResponse>> storeCaptor;

    @Captor
    private ArgumentCaptor<TranslateKeyResponseHandler> getCaptor;

    @Captor
    private ArgumentCaptor<MemcacheClientResponseHandler<DeleteCommandResponse>> deleteCaptor;

    @Captor
    private ArgumentCaptor<MemcacheClientResponseHandler<TouchCommandResponse>> touchCaptor;

    @Captor
    private ArgumentCaptor<MemcacheCommand> commandCaptor;

    private MemcacheClient client;

    @Before
    public void setUp() {
        MockitoAnnotations.initMocks(this);

        JsonObject configObj = new JsonObject();
        configObj.put(SERVERS_KEY, new JsonArray().add("server1"));
        configObj.put(EVENT_BUS_ADDRESS_KEY, "address");
        configObj.put(NAMESPACE_KEY, "namespace");
        configObj.put(POINTS_PER_SERVER, 10);
        configObj.put(ALGORITHM_KEY, HashAlgorithm.CRC_HASH.name());

        MemcacheConfig config = new MemcacheConfig(configObj);
        client = new MemcacheClient(eventBus, config);
    }

    @Test
    public void testIncr() {
        final ModifyCommandResponse response = new ModifyCommandResponse.Builder()
                .setStatus(JsendStatus.success)
                .build();

        client.incr("key", 1).setHandler(new Handler<AsyncResult<ModifyCommandResponse>>() {
            @Override
            public void handle(AsyncResult<ModifyCommandResponse> result) {
                assertEquals("Result doesn't match", response, result.result());
            }
        });

        MemcacheCommand command = new MemcacheCommand(MemcacheCommandType.incr, "namespacekey", "1", null);

        verify(eventBus, times(1)).send(eq("address_server1"), commandCaptor.capture(), withTimeout(Long.MAX_VALUE), modifyCaptor.capture());

        verifyCommand(command, commandCaptor.getValue());

        modifyCaptor.getValue().handle(Future.succeededFuture(new TestMessage<>(response)));