Phase-5: Adding TCP client and different connection types

build.gradle

@@ -13,14 +13,15 @@ repositories {
 dependencies {
     testImplementation platform('org.junit:junit-bom:5.10.0')
     testImplementation 'org.junit.jupiter:junit-jupiter'
+    testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
 }
 
 application {
     mainClass = 'org.cache.Main'
 }
 
 tasks.withType(JavaCompile).configureEach {
-    options.release = 17
+    options.release = 21
 }
 
 test {

src/main/java/org/cache/Main.java

@@ -1,26 +1,31 @@
 package org.cache;
 
-
 import org.cache.core.LocalCache;
+import org.cache.core.ValueType;
 import org.cache.eviction.LruEvictionPolicy;
 import org.cache.network.TcpCacheServer;
 import org.cache.protocol.CommandParser;
 import org.cache.protocol.CommandProcessor;
-import org.cache.protocol.codec.StringCodec;
+import org.cache.protocol.codec.ListValueCodec;
+import org.cache.protocol.codec.StringKeyCodec;
+import org.cache.protocol.codec.StringValueCodec;
+import org.cache.protocol.codec.ValueCodecRegistry;
 
 import java.io.IOException;
 
 public class Main {
     public static void main(String[] args) throws IOException {
         int port = args.length > 0 ? Integer.parseInt(args[0]) : 2020;
-        var cache = new LocalCache<String, String>(1_000, new LruEvictionPolicy<>());
-        var stringCodec = new StringCodec();
-        var commandParser = new CommandParser<>(stringCodec, stringCodec);
-        var commandProcessor = new CommandProcessor<>(cache, commandParser, stringCodec);
+        var cache = new LocalCache<String>(1_000, new LruEvictionPolicy<>());
+        var keyCodec = new StringKeyCodec();
+        var valueCodecs = new ValueCodecRegistry();
+        valueCodecs.register(ValueType.STRING, new StringValueCodec()).register(ValueType.LIST, new ListValueCodec());
+        var commandParser = new CommandParser<>(keyCodec);
+        var commandProcessor = new CommandProcessor<>(cache, commandParser, valueCodecs);
 
 
         try (cache; var server = new TcpCacheServer(port, commandProcessor)) {
             server.start();
         }
     }
-}
+}

src/main/java/org/cache/client/CacheClient.java

@@ -0,0 +1,22 @@
+package org.cache.client;
+
+import org.cache.core.metrics.Snapshot;
+
+import java.util.Optional;
+
+public interface CacheClient<K, V> {
+
+    void put(K key, V value, long ttl);
+
+    void put(K key, V value);
+
+    Optional<V> get(K key);
+
+    void delete(K key);
+
+    Snapshot metrics();
+
+    int size();
+
+    void clear();
+}

src/main/java/org/cache/client/CacheClientException.java

@@ -0,0 +1,12 @@
+package org.cache.client;
+
+public class CacheClientException extends RuntimeException {
+
+    public CacheClientException(String message) {
+        super(message);
+    }
+
+    public CacheClientException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

src/main/java/org/cache/client/TcpCacheClient.java

@@ -0,0 +1,185 @@
+package org.cache.client;
+
+import org.cache.client.serializer.Serializer;
+import org.cache.core.metrics.Snapshot;
+import org.cache.network.connection.RespConnection;
+import org.cache.protocol.codec.KeyCodec;
+import org.cache.protocol.commands.ResponseConstants;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.List;
+import java.util.Optional;
+
+import static org.cache.protocol.commands.ResponseConstants.ERROR;
+import static org.cache.protocol.commands.ResponseConstants.OK;
+
+public class TcpCacheClient<K, V> implements CacheClient<K, V>, AutoCloseable {
+
+    private static final int DEFAULT_CONNECT_TIMEOUT_MILLIS = 2_000;
+    private static final int DEFAULT_READ_TIMEOUT_MILLIS = 5_000;
+
+    private final RespConnection connection;
+    private final KeyCodec<K> keyCodec;
+    private final Serializer<V> valueSerializer;
+
+
+    public TcpCacheClient(String address, int port, KeyCodec<K> keyCodec, Serializer<V> valueSerializer) {
+        this(address, port, keyCodec, valueSerializer, DEFAULT_CONNECT_TIMEOUT_MILLIS, DEFAULT_READ_TIMEOUT_MILLIS);
+    }
+
+    public TcpCacheClient(
+            String address,
+            int port,
+            KeyCodec<K> keyCodec,
+            Serializer<V> valueSerializer,
+            int connectTimeoutMillis,
+            int readTimeoutMillis
+    ) {
+        try {
+            this.connection = new RespConnection(connect(address, port, connectTimeoutMillis, readTimeoutMillis));
+            this.keyCodec = keyCodec;
+            this.valueSerializer = valueSerializer;
+        } catch (IOException exception) {
+            throw new CacheClientException("failed to connect to cache server", exception);
+        }
+    }
+
+    @Override
+    public void put(K key, V value, long ttl) {
+        expectOk(List.of("PUT", keyCodec.encode(key), valueSerializer.encode(value), Long.toString(ttl)));
+    }
+
+    @Override
+    public void put(K key, V value) {
+        expectOk(List.of("PUT", keyCodec.encode(key), valueSerializer.encode(value)));
+    }
+
+    @Override
+    public Optional<V> get(K key) {
+        List<String> response = send(List.of("GET", keyCodec.encode(key)));
+
+        if (isResponse(response, ResponseConstants.NOT_FOUND.name(), 1)) {
+            return Optional.empty();
+        }
+
+        if (isResponse(response, ResponseConstants.VALUE.name(), 2)) {
+            return Optional.of(valueSerializer.decode(response.get(1)));
+        }
+
+        throw new CacheClientException("unexpected cache server response: " + response);
+    }
+
+    @Override
+    public void delete(K key) {
+        expectOk(List.of("DELETE", keyCodec.encode(key)));
+    }
+
+    @Override
+    public Snapshot metrics() {
+        List<String> response = send(List.of("METRICS"));
+
+        if (response.isEmpty() || !response.get(0).equals(ResponseConstants.METRICS.name())) {
+            throw new CacheClientException("unexpected cache server response: " + response);
+        }
+
+        long hits = 0;
+        long misses = 0;
+        long evictions = 0;
+        long expirations = 0;
+        double hitRate = 0;
+
+        if ((response.size() - 1) % 2 != 0) {
+            throw new CacheClientException("unexpected cache server response: " + response);
+        }
+
+        for (int i = 1; i < response.size(); i += 2) {
+            String name = response.get(i);
+            String value = response.get(i + 1);
+
+            try {
+                switch (name) {
+                    case "hits" -> hits = Long.parseLong(value);
+                    case "misses" -> misses = Long.parseLong(value);
+                    case "evictions" -> evictions = Long.parseLong(value);
+                    case "expirations" -> expirations = Long.parseLong(value);
+                    case "hitRate" -> hitRate = Double.parseDouble(value);
+                    default -> throw new CacheClientException("unexpected cache server response: " + response);
+                }
+            } catch (NumberFormatException exception) {
+                throw new CacheClientException("unexpected cache server response: " + response);
+            }
+        }
+
+        return new Snapshot(hits, misses, evictions, expirations, hitRate);
+    }
+
+    @Override
+    public int size() {
+        List<String> response = send(List.of("SIZE"));
+
+        if (!isResponse(response, ResponseConstants.SIZE.name(), 2)) {
+            throw new CacheClientException("unexpected cache server response: " + response);
+        }
+
+        return Integer.parseInt(response.get(1));
+    }
+
+    @Override
+    public void clear() {
+        expectOk(List.of("CLEAR"));
+    }
+
+    @Override
+    public void close() {
+        try {
+            connection.close();
+        } catch (IOException exception) {
+            throw new CacheClientException("failed to close cache client", exception);
+        }
+    }
+
+    private void expectOk(List<String> command) {
+        List<String> response = send(command);
+
+        if (!isResponse(response, OK.name(), 1)) {
+            throw new CacheClientException("unexpected cache server response: " + response);
+        }
+    }
+
+    private List<String> send(List<String> command) {
+        try {
+            List<String> response = connection.sendCommandForResponse(command);
+
+            if (!response.isEmpty() && response.getFirst().equals(ERROR.name())) {
+                throw new CacheClientException("unexpected cache server response: " + response);
+            }
+
+            return response;
+        } catch (IOException exception) {
+            throw new CacheClientException("failed to read response from cache server", exception);
+        }
+    }
+
+    private boolean isResponse(List<String> response, String type, int size) {
+        return response.size() == size && response.getFirst().equals(type);
+    }
+
+    private static Socket connect(
+            String address,
+            int port,
+            int connectTimeoutMillis,
+            int readTimeoutMillis
+    ) throws IOException {
+        Socket socket = new Socket();
+        try {
+            socket.connect(new InetSocketAddress(address, port), connectTimeoutMillis);
+            socket.setSoTimeout(readTimeoutMillis);
+            return socket;
+        } catch (IOException exception) {
+            socket.close();
+            throw exception;
+        }
+    }
+}

src/main/java/org/cache/client/serializer/Serializer.java

@@ -0,0 +1,8 @@
+package org.cache.client.serializer;
+
+public interface Serializer<V> {
+
+    String encode(V value);
+
+    V decode(String value);
+}

src/main/java/org/cache/client/serializer/StringSerializer.java

@@ -0,0 +1,22 @@
+package org.cache.client.serializer;
+
+public class StringSerializer implements Serializer<String> {
+
+    @Override
+    public String encode(String value) {
+        if (value == null) {
+            return "";
+        }
+
+        return value;
+    }
+
+    @Override
+    public String decode(String value) {
+        if (value == null) {
+            return "";
+        }
+
+        return value;
+    }
+}

src/main/java/org/cache/core/Cache.java

@@ -4,11 +4,11 @@
 
 import java.util.Optional;
 
-public interface Cache<K, V> {
+public interface Cache<K> {
 
-    void put(K key, V value, long ttlMillis);
+    void put(K key, byte[] value, ValueType type, long ttlMillis);
 
-    Optional<V> get(K key);
+    Optional<CacheEntry> get(K key);
 
     void delete(K key);
 

src/main/java/org/cache/core/CacheEntry.java

@@ -1,19 +1,27 @@
 package org.cache.core;
 
-public class CacheEntry<V> {
+import java.util.Arrays;
 
-    private final V value;
+public class CacheEntry {
+
+    private final byte[] value;
+    private final ValueType type;
     private final long expiresAt;
 
-    CacheEntry(V value, long ttlMillis) {
-        this.value = value;
+    CacheEntry(byte[] value, ValueType type, long ttlMillis) {
+        this.value = Arrays.copyOf(value, value.length);
+        this.type = type;
         this.expiresAt = ttlMillis > 0
                 ? System.currentTimeMillis() + ttlMillis
                 : Long.MAX_VALUE;
     }
 
-    public V getValue() {
-        return value;
+    public byte[] getValue() {
+        return Arrays.copyOf(value, value.length);
+    }
+
+    public ValueType getType() {
+        return type;
     }
 
     public boolean isExpired() {