diff --git a/build.gradle b/build.gradle index 51276ba..eb4ae53 100644 --- a/build.gradle +++ b/build.gradle @@ -13,6 +13,7 @@ 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 { @@ -20,7 +21,7 @@ application { } tasks.withType(JavaCompile).configureEach { - options.release = 17 + options.release = 21 } test { diff --git a/src/main/java/org/cache/Main.java b/src/main/java/org/cache/Main.java index d32c17e..e066ed6 100644 --- a/src/main/java/org/cache/Main.java +++ b/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(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(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(); } } -} \ No newline at end of file +} diff --git a/src/main/java/org/cache/client/CacheClient.java b/src/main/java/org/cache/client/CacheClient.java new file mode 100644 index 0000000..8912bae --- /dev/null +++ b/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 { + + void put(K key, V value, long ttl); + + void put(K key, V value); + + Optional get(K key); + + void delete(K key); + + Snapshot metrics(); + + int size(); + + void clear(); +} diff --git a/src/main/java/org/cache/client/CacheClientException.java b/src/main/java/org/cache/client/CacheClientException.java new file mode 100644 index 0000000..219e8c7 --- /dev/null +++ b/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); + } +} diff --git a/src/main/java/org/cache/client/TcpCacheClient.java b/src/main/java/org/cache/client/TcpCacheClient.java new file mode 100644 index 0000000..fdc3b01 --- /dev/null +++ b/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 implements CacheClient, 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 keyCodec; + private final Serializer valueSerializer; + + + public TcpCacheClient(String address, int port, KeyCodec keyCodec, Serializer valueSerializer) { + this(address, port, keyCodec, valueSerializer, DEFAULT_CONNECT_TIMEOUT_MILLIS, DEFAULT_READ_TIMEOUT_MILLIS); + } + + public TcpCacheClient( + String address, + int port, + KeyCodec keyCodec, + Serializer 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 get(K key) { + List 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 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 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 command) { + List response = send(command); + + if (!isResponse(response, OK.name(), 1)) { + throw new CacheClientException("unexpected cache server response: " + response); + } + } + + private List send(List command) { + try { + List 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 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; + } + } +} diff --git a/src/main/java/org/cache/client/serializer/Serializer.java b/src/main/java/org/cache/client/serializer/Serializer.java new file mode 100644 index 0000000..554ac4a --- /dev/null +++ b/src/main/java/org/cache/client/serializer/Serializer.java @@ -0,0 +1,8 @@ +package org.cache.client.serializer; + +public interface Serializer { + + String encode(V value); + + V decode(String value); +} diff --git a/src/main/java/org/cache/client/serializer/StringSerializer.java b/src/main/java/org/cache/client/serializer/StringSerializer.java new file mode 100644 index 0000000..72f6c85 --- /dev/null +++ b/src/main/java/org/cache/client/serializer/StringSerializer.java @@ -0,0 +1,22 @@ +package org.cache.client.serializer; + +public class StringSerializer implements Serializer { + + @Override + public String encode(String value) { + if (value == null) { + return ""; + } + + return value; + } + + @Override + public String decode(String value) { + if (value == null) { + return ""; + } + + return value; + } +} diff --git a/src/main/java/org/cache/core/Cache.java b/src/main/java/org/cache/core/Cache.java index 845b83d..8d75d8c 100644 --- a/src/main/java/org/cache/core/Cache.java +++ b/src/main/java/org/cache/core/Cache.java @@ -4,11 +4,11 @@ import java.util.Optional; -public interface Cache { +public interface Cache { - void put(K key, V value, long ttlMillis); + void put(K key, byte[] value, ValueType type, long ttlMillis); - Optional get(K key); + Optional get(K key); void delete(K key); diff --git a/src/main/java/org/cache/core/CacheEntry.java b/src/main/java/org/cache/core/CacheEntry.java index 37d2b14..9d34da7 100644 --- a/src/main/java/org/cache/core/CacheEntry.java +++ b/src/main/java/org/cache/core/CacheEntry.java @@ -1,19 +1,27 @@ package org.cache.core; -public class CacheEntry { +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() { diff --git a/src/main/java/org/cache/core/LocalCache.java b/src/main/java/org/cache/core/LocalCache.java index 2129abd..9725c55 100644 --- a/src/main/java/org/cache/core/LocalCache.java +++ b/src/main/java/org/cache/core/LocalCache.java @@ -12,16 +12,16 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -public class LocalCache implements Cache, AutoCloseable { +public class LocalCache implements Cache, AutoCloseable { - private final ConcurrentHashMap> cache; + private final ConcurrentHashMap cache; private final int capacity; private final EvictionPolicy evictionPolicy; private final CacheMetrics metrics; private final Object evictionLock; private final int cleanupBatchSize; private final ScheduledExecutorService cleanupScheduler; - private Iterator>> cleanupIterator; + private Iterator> cleanupIterator; public LocalCache(int capacity, EvictionPolicy evictionPolicy) { this(capacity, evictionPolicy, 1_000, 100); @@ -56,9 +56,9 @@ public LocalCache( } @Override - public void put(K key, V value, long ttlMillis) { + public void put(K key, byte[] value, ValueType type, long ttlMillis) { - var newEntry = new CacheEntry<>(value, ttlMillis); + var newEntry = new CacheEntry(value, type, ttlMillis); synchronized (evictionLock) { cache.put(key, newEntry); @@ -76,7 +76,7 @@ public void put(K key, V value, long ttlMillis) { } @Override - public Optional get(K key) { + public Optional get(K key) { var entry = cache.get(key); if (entry == null) { @@ -103,7 +103,7 @@ public Optional get(K key) { evictionPolicy.onKeyAccessed(key); metrics.recordHit(); - return Optional.ofNullable(currentEntry.getValue()); + return Optional.of(currentEntry); } } diff --git a/src/main/java/org/cache/core/ValueType.java b/src/main/java/org/cache/core/ValueType.java new file mode 100644 index 0000000..49dd9ac --- /dev/null +++ b/src/main/java/org/cache/core/ValueType.java @@ -0,0 +1,7 @@ +package org.cache.core; + +public enum ValueType { + + STRING, + LIST +} diff --git a/src/main/java/org/cache/core/metrics/Snapshot.java b/src/main/java/org/cache/core/metrics/Snapshot.java index 30852b6..d7d351e 100644 --- a/src/main/java/org/cache/core/metrics/Snapshot.java +++ b/src/main/java/org/cache/core/metrics/Snapshot.java @@ -7,7 +7,7 @@ public class Snapshot { private final long expirations; private final double hitRate; - Snapshot(long hits, long misses, long evictions, long expirations, double hitRate) { + public Snapshot(long hits, long misses, long evictions, long expirations, double hitRate) { this.hits = hits; this.misses = misses; this.evictions = evictions; @@ -34,4 +34,4 @@ public long getExpirations() { public double getHitRate() { return hitRate; } -} \ No newline at end of file +} diff --git a/src/main/java/org/cache/network/ClientConnectionHandler.java b/src/main/java/org/cache/network/ClientConnectionHandler.java deleted file mode 100644 index b48a85b..0000000 --- a/src/main/java/org/cache/network/ClientConnectionHandler.java +++ /dev/null @@ -1,40 +0,0 @@ -package org.cache.network; - -import org.cache.protocol.CommandProcessor; - -import java.io.BufferedReader; -import java.io.IOException; -import java.io.InputStreamReader; -import java.io.PrintWriter; -import java.net.Socket; -import java.nio.charset.StandardCharsets; - -public class ClientConnectionHandler implements Runnable { - - private final Socket socket; - private final CommandProcessor commandProcessor; - - public ClientConnectionHandler(Socket socket, CommandProcessor commandProcessor) { - this.socket = socket; - this.commandProcessor = commandProcessor; - } - - @Override - public void run() { - try ( - socket; - var reader = new BufferedReader( - new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8) - ); - var writer = new PrintWriter(socket.getOutputStream(), true, StandardCharsets.UTF_8) - ) { - String command; - while ((command = reader.readLine()) != null) { - String response = commandProcessor.process(command); - writer.println(response); - } - } catch (IOException exception) { - System.err.println("Client connection failed: " + exception.getMessage()); - } - } -} diff --git a/src/main/java/org/cache/network/TcpCacheServer.java b/src/main/java/org/cache/network/TcpCacheServer.java index c805f83..8609ebe 100644 --- a/src/main/java/org/cache/network/TcpCacheServer.java +++ b/src/main/java/org/cache/network/TcpCacheServer.java @@ -1,5 +1,6 @@ package org.cache.network; +import org.cache.network.connection.ClientConnectionHandler; import org.cache.protocol.CommandProcessor; import java.io.IOException; @@ -11,16 +12,16 @@ public class TcpCacheServer implements AutoCloseable { private final int port; - private final CommandProcessor commandProcessor; + private final CommandProcessor commandProcessor; private final ExecutorService executor; private ServerSocket serverSocket; - private final static int NUMBER_OF_THREADS = 16; + private static final int THREAD_COUNT = 16; - public TcpCacheServer(int port, CommandProcessor commandProcessor) { + public TcpCacheServer(int port, CommandProcessor commandProcessor) { this.port = port; this.commandProcessor = commandProcessor; - this.executor = Executors.newFixedThreadPool(NUMBER_OF_THREADS); + this.executor = Executors.newFixedThreadPool(THREAD_COUNT); } public void start() throws IOException { diff --git a/src/main/java/org/cache/network/connection/ClientConnectionHandler.java b/src/main/java/org/cache/network/connection/ClientConnectionHandler.java new file mode 100644 index 0000000..d623dda --- /dev/null +++ b/src/main/java/org/cache/network/connection/ClientConnectionHandler.java @@ -0,0 +1,41 @@ +package org.cache.network.connection; + +import org.cache.protocol.CommandProcessor; + +import java.io.IOException; +import java.net.Socket; +import java.util.List; + +public class ClientConnectionHandler implements Runnable { + + private final Socket socket; + private final CommandProcessor commandProcessor; + + public ClientConnectionHandler(Socket socket, CommandProcessor commandProcessor) { + this.socket = socket; + this.commandProcessor = commandProcessor; + } + + @Override + public void run() { + try (socket) { + var connectionFactory = new ProtocolConnectionFactory(socket); + var connection = connectionFactory.create(); + + if (connection.isEmpty()) { + return; + } + + var protocolConnection = connection.get(); + + List command; + while ((command = protocolConnection.readCommand()) != null) { + String response = commandProcessor.process(command); + protocolConnection.write(response); + } + + } catch (IOException exception) { + System.err.println("Client connection failed: " + exception.getMessage()); + } + } +} diff --git a/src/main/java/org/cache/network/connection/LineProtocolConnection.java b/src/main/java/org/cache/network/connection/LineProtocolConnection.java new file mode 100644 index 0000000..1872c95 --- /dev/null +++ b/src/main/java/org/cache/network/connection/LineProtocolConnection.java @@ -0,0 +1,86 @@ +package org.cache.network.connection; + +import org.cache.protocol.ProtocolConstants; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.List; + +public class LineProtocolConnection implements ProtocolConnection, AutoCloseable { + + private final Socket socket; + private final InputStream input; + private final OutputStream output; + + public LineProtocolConnection(Socket socket) throws IOException { + this(socket, socket.getInputStream(), socket.getOutputStream()); + } + + LineProtocolConnection(Socket socket, InputStream input, OutputStream output) { + this.socket = socket; + this.input = input; + this.output = output; + } + + public List readCommand() throws IOException { + String line = readLine(); + + if (line == null) { + return null; + } + + if (line.isBlank()) { + return List.of(); + } + + return List.of(line.trim().split("\\s+")); + } + + @Override + public String send(String line) throws IOException { + write(line); + String response = readLine(); + + if (response == null) { + throw new IOException("Connection closed"); + } + + return response; + } + + public String readLine() throws IOException { + StringBuilder line = new StringBuilder(); + + while (true) { + int next = input.read(); + + if (next == -1) { + return line.isEmpty() ? null : line.toString(); + } + + if (next == ProtocolConstants.NEW_LINE) { + return line.toString(); + } + + if (next != ProtocolConstants.CARRIAGE_RETURN) { + line.append((char) next); + } + } + } + + @Override + public void write(String line) throws IOException { + output.write(line.getBytes(StandardCharsets.UTF_8)); + output.write(ProtocolConstants.CARRIAGE_RETURN); + output.write(ProtocolConstants.NEW_LINE); + output.flush(); + } + + @Override + public void close() throws IOException { + socket.close(); + } +} diff --git a/src/main/java/org/cache/network/connection/ProtocolConnection.java b/src/main/java/org/cache/network/connection/ProtocolConnection.java new file mode 100644 index 0000000..f8c5382 --- /dev/null +++ b/src/main/java/org/cache/network/connection/ProtocolConnection.java @@ -0,0 +1,13 @@ +package org.cache.network.connection; + +import java.io.IOException; +import java.util.List; + +public interface ProtocolConnection { + + List readCommand() throws IOException; + + void write(String value) throws IOException; + + String send(String value) throws IOException; +} diff --git a/src/main/java/org/cache/network/connection/ProtocolConnectionFactory.java b/src/main/java/org/cache/network/connection/ProtocolConnectionFactory.java new file mode 100644 index 0000000..71844db --- /dev/null +++ b/src/main/java/org/cache/network/connection/ProtocolConnectionFactory.java @@ -0,0 +1,35 @@ +package org.cache.network.connection; + +import org.cache.protocol.ProtocolConstants; + +import java.io.IOException; +import java.io.PushbackInputStream; +import java.net.Socket; +import java.util.Optional; + +public class ProtocolConnectionFactory { + + private final Socket socket; + + public ProtocolConnectionFactory(Socket socket) { + this.socket = socket; + } + + public Optional create() throws IOException { + var input = new PushbackInputStream(socket.getInputStream(), 1); + var output = socket.getOutputStream(); + int firstByte = input.read(); + + if (firstByte == -1) { + return Optional.empty(); + } + + input.unread(firstByte); + + if (firstByte == ProtocolConstants.ARRAY_PREFIX) { + return Optional.of(new RespConnection(socket, input, output)); + } + + return Optional.of(new LineProtocolConnection(socket, input, output)); + } +} diff --git a/src/main/java/org/cache/network/connection/RespConnection.java b/src/main/java/org/cache/network/connection/RespConnection.java new file mode 100644 index 0000000..4406436 --- /dev/null +++ b/src/main/java/org/cache/network/connection/RespConnection.java @@ -0,0 +1,223 @@ +package org.cache.network.connection; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.Socket; +import java.util.ArrayList; +import java.util.List; + +import static java.nio.charset.StandardCharsets.US_ASCII; +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.cache.protocol.ProtocolConstants.*; +import static org.cache.protocol.commands.ResponseConstants.ERROR; +import static org.cache.protocol.commands.ResponseConstants.METRICS; +import static org.cache.protocol.commands.ResponseConstants.NOT_FOUND; +import static org.cache.protocol.commands.ResponseConstants.OK; +import static org.cache.protocol.commands.ResponseConstants.SIZE; +import static org.cache.protocol.commands.ResponseConstants.VALUE; + +public class RespConnection implements ProtocolConnection, AutoCloseable { + + private final Socket socket; + private final InputStream input; + private final OutputStream output; + private final LineProtocolConnection lines; + + public RespConnection(Socket socket) throws IOException { + this(socket, socket.getInputStream(), socket.getOutputStream()); + } + + RespConnection(Socket socket, InputStream input, OutputStream output) throws IOException { + this.socket = socket; + this.input = input; + this.output = output; + this.lines = new LineProtocolConnection(socket, input, output); + } + + @Override + public List readCommand() throws IOException { + int type = input.read(); + + if (type == -1) { + return null; + } + + if (type != ARRAY_PREFIX) { + throw new IOException("Expected '" + ARRAY_PREFIX + "' but received '" + (char) type + "'"); + } + + int itemCount = parseInteger(lines.readLine(), "array length"); + List values = new ArrayList<>(itemCount); + + for (int i = 0; i < itemCount; i++) { + values.add(readBulkString()); + } + + return values; + } + + @Override + public void write(String value) throws IOException { + writeArray(responseParts(value)); + } + + @Override + public String send(String value) throws IOException { + return sendCommand(List.of(value.trim().split("\\s+"))); + } + + public String sendCommand(List commandParts) throws IOException { + List response = sendCommandForResponse(commandParts); + return String.join(" ", response); + } + + public List sendCommandForResponse(List commandParts) throws IOException { + writeArray(commandParts); + List response = readResponse(); + + if (response == null) { + throw new IOException("Connection closed"); + } + + return response; + } + + public void writeArray(List values) throws IOException { + writeAscii(ARRAY_PREFIX + Integer.toString(values.size()) + CRLF); + + for (String value : values) { + byte[] bytes = value.getBytes(UTF_8); + writeAscii(BULK_STRING_PREFIX + Integer.toString(bytes.length) + CRLF); + output.write(bytes); + writeAscii(CRLF); + } + + output.flush(); + } + + private String readString() throws IOException { + List response = readResponse(); + + if (response == null) { + return null; + } + + return String.join(" ", response); + } + + private List readResponse() throws IOException { + int type = input.read(); + + if (type == -1) { + return null; + } + + return switch (type) { + case ARRAY_PREFIX -> readArrayAfterType(); + case SIMPLE_STRING_PREFIX, INTEGER_PREFIX -> List.of(lines.readLine()); + case ERROR_PREFIX -> List.of(ERROR.name(), lines.readLine()); + case BULK_STRING_PREFIX -> List.of(readBulkStringAfterType()); + default -> throw new IOException("Unsupported RESP response type: " + (char) type); + }; + } + + private List readArrayAfterType() throws IOException { + int itemCount = parseInteger(lines.readLine(), "array length"); + List values = new ArrayList<>(itemCount); + + for (int i = 0; i < itemCount; i++) { + values.add(readBulkString()); + } + + return values; + } + + private String readBulkString() throws IOException { + expect(BULK_STRING_PREFIX); + return readBulkStringAfterType(); + } + + private String readBulkStringAfterType() throws IOException { + int length = parseInteger(lines.readLine(), "bulk string length"); + + if (length < 0) { + return null; + } + + byte[] bytes = input.readNBytes(length); + if (bytes.length != length) { + throw new IOException("Connection closed while reading bulk string"); + } + + expect(CARRIAGE_RETURN); + expect(NEW_LINE); + + return new String(bytes, UTF_8); + } + + private int parseInteger(String value, String name) throws IOException { + try { + return Integer.parseInt(value); + } catch (NumberFormatException exception) { + throw new IOException("Invalid RESP " + name + ": " + value, exception); + } + } + + private void expect(char expected) throws IOException { + int actual = input.read(); + + if (actual != expected) { + throw new IOException("Expected '" + expected + "' but received '" + (char) actual + "'"); + } + } + + private void writeAscii(String value) throws IOException { + output.write(value.getBytes(US_ASCII)); + } + + private List responseParts(String response) { + if (response.equals(OK.name()) || response.equals(NOT_FOUND.name())) { + return List.of(response); + } + + if (response.startsWith(ERROR.name() + " ")) { + return List.of(ERROR.name(), response.substring((ERROR.name() + " ").length())); + } + + if (response.startsWith(VALUE.name() + " ")) { + return List.of(VALUE.name(), response.substring((VALUE.name() + " ").length())); + } + + if (response.startsWith(SIZE.name() + " ")) { + return List.of(SIZE.name(), response.substring((SIZE.name() + " ").length())); + } + + if (response.startsWith(METRICS.name() + " ")) { + return metricParts(response); + } + + return List.of(response); + } + + private List metricParts(String response) { + List parts = new ArrayList<>(); + parts.add(METRICS.name()); + + String metrics = response.substring((METRICS.name() + " ").length()); + for (String metric : metrics.split("\\s+")) { + String[] nameAndValue = metric.split("=", 2); + if (nameAndValue.length == 2) { + parts.add(nameAndValue[0]); + parts.add(nameAndValue[1]); + } + } + + return parts; + } + + @Override + public void close() throws IOException { + socket.close(); + } +} diff --git a/src/main/java/org/cache/protocol/CommandParser.java b/src/main/java/org/cache/protocol/CommandParser.java index 239f6f5..15e2d82 100644 --- a/src/main/java/org/cache/protocol/CommandParser.java +++ b/src/main/java/org/cache/protocol/CommandParser.java @@ -1,35 +1,36 @@ package org.cache.protocol; -import org.cache.protocol.codec.Codec; +import org.cache.core.ValueType; +import org.cache.protocol.codec.KeyCodec; import org.cache.protocol.commands.*; -import java.util.concurrent.TimeUnit; +import java.util.List; -public class CommandParser { +public class CommandParser { - private final Codec keyCodec; - private final Codec valueCodec; + private final KeyCodec keyCodec; + private static final int COMMAND_PARTS = 1; + private static final int KEY_COMMAND_PARTS = 2; + private static final int VALUE_COMMAND_PARTS = 3; + private static final int TTL_COMMAND_PARTS = 4; - private final static int ONE = 1; - private final static int TWO = 2; - private final static int THREE = 3; - private final static int FOUR = 4; + private static final int COMMAND_INDEX = 0; + private static final int KEY_INDEX = 1; + private static final int VALUE_INDEX = 2; + private static final int TTL_INDEX = 3; - public CommandParser(Codec keyCodec, Codec valueCodec) { + public CommandParser(KeyCodec keyCodec) { this.keyCodec = keyCodec; - this.valueCodec = valueCodec; } - public CacheCommand parse(String rawCommand) { - if (rawCommand == null || rawCommand.isBlank()) { + public CacheCommand parse(List parts) { + if (parts == null || parts.isEmpty()) { return new UnknownCommand<>(); } - String[] parts = rawCommand.trim().split("\\s+"); - try { - CommandType type = CommandType.valueOf(parts[0].toUpperCase()); + CommandType type = CommandType.valueOf(parts.get(COMMAND_INDEX).toUpperCase()); return switch (type) { case PUT -> parsePut(parts); case GET -> parseGet(parts); @@ -37,6 +38,7 @@ public CacheCommand parse(String rawCommand) { case SIZE -> parseSize(parts); case CLEAR -> parseClear(parts); case METRICS -> parseMetrics(parts); + case PUSH -> parsePush(parts); case UNKNOWN -> new UnknownCommand<>(); }; } catch (IllegalArgumentException exception) { @@ -44,56 +46,72 @@ public CacheCommand parse(String rawCommand) { } } - private CacheCommand parsePut(String[] parts) { - if (parts.length != THREE && parts.length != FOUR) { - return new InvalidCommand<>("usage: PUT key value [ttlSeconds]"); + + private CacheCommand parsePut(List parts) { + if (parts.size() != VALUE_COMMAND_PARTS && parts.size() != TTL_COMMAND_PARTS) { + return new InvalidCommand<>("usage: PUT key value [ttlMillis]"); } try { + long ttlMillis = parts.size() == TTL_COMMAND_PARTS ? Long.parseLong(parts.get(TTL_INDEX)) : 0; + return new PutCommand<>( - keyCodec.decode(parts[ONE]), - valueCodec.decode(parts[TWO]), - Long.parseLong(parts[THREE]) + keyCodec.decode(parts.get(KEY_INDEX)), + parts.get(VALUE_INDEX), + ValueType.STRING, + ttlMillis ); } catch (NumberFormatException exception) { return new InvalidCommand<>("ttl must be a number"); } } - private CacheCommand parseGet(String[] parts) { - if (parts.length != TWO) { + private CacheCommand parsePush(List parts) { + if (parts.size() != VALUE_COMMAND_PARTS) { + return new InvalidCommand<>("usage: PUSH key value"); + } + + return new PushCommand<>( + keyCodec.decode(parts.get(KEY_INDEX)), + parts.get(VALUE_INDEX), + ValueType.LIST + ); + } + + private CacheCommand parseGet(List parts) { + if (parts.size() != KEY_COMMAND_PARTS) { return new InvalidCommand<>("usage: GET key"); } - return new GetCommand<>(keyCodec.decode(parts[1])); + return new GetCommand<>(keyCodec.decode(parts.get(KEY_INDEX))); } - private CacheCommand parseDelete(String[] parts) { - if (parts.length != TWO) { + private CacheCommand parseDelete(List parts) { + if (parts.size() != KEY_COMMAND_PARTS) { return new InvalidCommand<>("usage: DELETE key"); } - return new DeleteCommand<>(keyCodec.decode(parts[1])); + return new DeleteCommand<>(keyCodec.decode(parts.get(KEY_INDEX))); } - private CacheCommand parseSize(String[] parts) { - if (parts.length != ONE) { + private CacheCommand parseSize(List parts) { + if (parts.size() != COMMAND_PARTS) { return new InvalidCommand<>("usage: SIZE"); } return new SizeCommand<>(); } - private CacheCommand parseClear(String[] parts) { - if (parts.length != ONE) { + private CacheCommand parseClear(List parts) { + if (parts.size() != COMMAND_PARTS) { return new InvalidCommand<>("usage: CLEAR"); } return new ClearCommand<>(); } - private CacheCommand parseMetrics(String[] parts) { - if (parts.length != ONE) { + private CacheCommand parseMetrics(List parts) { + if (parts.size() != COMMAND_PARTS) { return new InvalidCommand<>("usage: METRICS"); } diff --git a/src/main/java/org/cache/protocol/CommandProcessor.java b/src/main/java/org/cache/protocol/CommandProcessor.java index 0bdce98..9688c6e 100644 --- a/src/main/java/org/cache/protocol/CommandProcessor.java +++ b/src/main/java/org/cache/protocol/CommandProcessor.java @@ -1,23 +1,25 @@ package org.cache.protocol; import org.cache.core.Cache; -import org.cache.protocol.codec.Codec; +import org.cache.protocol.codec.ValueCodecRegistry; import org.cache.protocol.commands.CacheCommand; -public class CommandProcessor { +import java.util.List; - private final Cache cache; - private final CommandParser parser; - private final Codec valueCodec; +public class CommandProcessor { - public CommandProcessor(Cache cache, CommandParser parser, Codec valueCodec) { + private final Cache cache; + private final CommandParser parser; + private final ValueCodecRegistry valueCodecs; + + public CommandProcessor(Cache cache, CommandParser parser, ValueCodecRegistry valueCodecs) { this.cache = cache; this.parser = parser; - this.valueCodec = valueCodec; + this.valueCodecs = valueCodecs; } - public String process(String rawCommand) { - CacheCommand command = parser.parse(rawCommand); - return command.process(cache, valueCodec); + public String process(List commandParts) { + CacheCommand command = parser.parse(commandParts); + return command.process(cache, valueCodecs); } } diff --git a/src/main/java/org/cache/protocol/ProtocolConstants.java b/src/main/java/org/cache/protocol/ProtocolConstants.java new file mode 100644 index 0000000..44f6794 --- /dev/null +++ b/src/main/java/org/cache/protocol/ProtocolConstants.java @@ -0,0 +1,16 @@ +package org.cache.protocol; + +public final class ProtocolConstants { + + public static final char SIMPLE_STRING_PREFIX = '+'; + public static final char ERROR_PREFIX = '-'; + public static final char INTEGER_PREFIX = ':'; + public static final char BULK_STRING_PREFIX = '$'; + public static final char ARRAY_PREFIX = '*'; + public static final char CARRIAGE_RETURN = '\r'; + public static final char NEW_LINE = '\n'; + public static final String CRLF = "\r\n"; + + private ProtocolConstants() { + } +} diff --git a/src/main/java/org/cache/protocol/codec/Codec.java b/src/main/java/org/cache/protocol/codec/Codec.java deleted file mode 100644 index e6fe5f9..0000000 --- a/src/main/java/org/cache/protocol/codec/Codec.java +++ /dev/null @@ -1,8 +0,0 @@ -package org.cache.protocol.codec; - -public interface Codec { - - T decode(String value); - - String encode(T value); -} diff --git a/src/main/java/org/cache/protocol/codec/CodecConversionException.java b/src/main/java/org/cache/protocol/codec/CodecConversionException.java new file mode 100644 index 0000000..72deb81 --- /dev/null +++ b/src/main/java/org/cache/protocol/codec/CodecConversionException.java @@ -0,0 +1,7 @@ +package org.cache.protocol.codec; + +public class CodecConversionException extends RuntimeException { + public CodecConversionException(String message) { + super(message); + } +} diff --git a/src/main/java/org/cache/protocol/codec/IntegerKeyCodec.java b/src/main/java/org/cache/protocol/codec/IntegerKeyCodec.java new file mode 100644 index 0000000..a229816 --- /dev/null +++ b/src/main/java/org/cache/protocol/codec/IntegerKeyCodec.java @@ -0,0 +1,26 @@ +package org.cache.protocol.codec; + +public class IntegerKeyCodec implements KeyCodec { + + @Override + public String encode(Integer value) { + if (value == null) { + return null; + } + + return String.valueOf(value); + } + + @Override + public Integer decode(String value) { + if (value == null || value.trim().isEmpty()) { + return null; + } + + try { + return Integer.parseInt(value); + } catch (NumberFormatException e) { + throw new IllegalArgumentException("Cannot encode non-numeric string: " + value, e); + } + } +} diff --git a/src/main/java/org/cache/protocol/codec/KeyCodec.java b/src/main/java/org/cache/protocol/codec/KeyCodec.java new file mode 100644 index 0000000..f661da5 --- /dev/null +++ b/src/main/java/org/cache/protocol/codec/KeyCodec.java @@ -0,0 +1,8 @@ +package org.cache.protocol.codec; + +public interface KeyCodec { + + K decode(String value); + + String encode(K value); +} diff --git a/src/main/java/org/cache/protocol/codec/ListValueCodec.java b/src/main/java/org/cache/protocol/codec/ListValueCodec.java new file mode 100644 index 0000000..dffb5b9 --- /dev/null +++ b/src/main/java/org/cache/protocol/codec/ListValueCodec.java @@ -0,0 +1,47 @@ +package org.cache.protocol.codec; + +import java.io.*; +import java.util.ArrayList; +import java.util.List; + +public class ListValueCodec implements ValueCodec> { + + @Override + public byte[] encode(List value) { + + try(ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) { + DataOutputStream dao = new DataOutputStream(byteArrayOutputStream); + + for(String element : value) { + dao.writeUTF(element); + } + + return byteArrayOutputStream.toByteArray(); + } catch (IOException e) { + throw new CodecConversionException("Something went wrong in list to byte encoding"); + } + } + + @Override + public List decode(byte[] value) { + List list = new ArrayList<>(); + + try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(value); + DataInputStream dis = new DataInputStream(byteArrayInputStream)) { + + while (dis.available() > 0) { + list.add(dis.readUTF()); + } + } catch (IOException e) { + throw new CodecConversionException("Something went wrong in byte to list decoding"); + } + return list; + } + + @Override + public String toString(byte[] value) { + List list = decode(value); + + return String.join(", ", list); + } +} diff --git a/src/main/java/org/cache/protocol/codec/StringCodec.java b/src/main/java/org/cache/protocol/codec/StringKeyCodec.java similarity index 77% rename from src/main/java/org/cache/protocol/codec/StringCodec.java rename to src/main/java/org/cache/protocol/codec/StringKeyCodec.java index c75d45a..cedab19 100644 --- a/src/main/java/org/cache/protocol/codec/StringCodec.java +++ b/src/main/java/org/cache/protocol/codec/StringKeyCodec.java @@ -1,6 +1,6 @@ package org.cache.protocol.codec; -public class StringCodec implements Codec { +public class StringKeyCodec implements KeyCodec { @Override public String decode(String value) { diff --git a/src/main/java/org/cache/protocol/codec/StringValueCodec.java b/src/main/java/org/cache/protocol/codec/StringValueCodec.java new file mode 100644 index 0000000..fc168c7 --- /dev/null +++ b/src/main/java/org/cache/protocol/codec/StringValueCodec.java @@ -0,0 +1,29 @@ +package org.cache.protocol.codec; + +import java.nio.charset.StandardCharsets; + +public class StringValueCodec implements ValueCodec { + + @Override + public byte[] encode(String value) { + if (value == null) { + throw new CodecConversionException("Value must not be null for encoding"); + } + + return value.getBytes(StandardCharsets.UTF_8); + } + + @Override + public String decode(byte[] value) { + + if (value == null) { + throw new CodecConversionException("Value must not be null for decoding"); + } + return new String(value, StandardCharsets.UTF_8); + } + + @Override + public String toString(byte[] value) { + return new String(value, StandardCharsets.UTF_8); + } +} diff --git a/src/main/java/org/cache/protocol/codec/ValueCodec.java b/src/main/java/org/cache/protocol/codec/ValueCodec.java new file mode 100644 index 0000000..df896dc --- /dev/null +++ b/src/main/java/org/cache/protocol/codec/ValueCodec.java @@ -0,0 +1,10 @@ +package org.cache.protocol.codec; + +public interface ValueCodec { + + byte[] encode(V value); + + V decode(byte[] value); + + String toString(byte[] value); +} diff --git a/src/main/java/org/cache/protocol/codec/ValueCodecRegistry.java b/src/main/java/org/cache/protocol/codec/ValueCodecRegistry.java new file mode 100644 index 0000000..14b1786 --- /dev/null +++ b/src/main/java/org/cache/protocol/codec/ValueCodecRegistry.java @@ -0,0 +1,25 @@ +package org.cache.protocol.codec; + +import org.cache.core.ValueType; + +import java.util.EnumMap; + +public class ValueCodecRegistry { + + private final EnumMap> codecs = new EnumMap<>(ValueType.class); + + public ValueCodecRegistry register(ValueType type, ValueCodec valueCodec) { + codecs.put(type, valueCodec); + return this; + } + + public ValueCodec get(ValueType type) { + ValueCodec valueCodec = codecs.get(type); + + if (valueCodec == null) { + throw new IllegalArgumentException("unsupported value type: " + type); + } + + return valueCodec; + } +} diff --git a/src/main/java/org/cache/protocol/commands/CacheCommand.java b/src/main/java/org/cache/protocol/commands/CacheCommand.java index 8e26a69..475f66a 100644 --- a/src/main/java/org/cache/protocol/commands/CacheCommand.java +++ b/src/main/java/org/cache/protocol/commands/CacheCommand.java @@ -1,9 +1,9 @@ package org.cache.protocol.commands; import org.cache.core.Cache; -import org.cache.protocol.codec.Codec; +import org.cache.protocol.codec.ValueCodecRegistry; -public interface CacheCommand { +public interface CacheCommand { - String process(Cache cache, Codec valueCodec); + String process(Cache cache, ValueCodecRegistry valueCodecs); } diff --git a/src/main/java/org/cache/protocol/commands/ClearCommand.java b/src/main/java/org/cache/protocol/commands/ClearCommand.java index 2e352ac..ce09795 100644 --- a/src/main/java/org/cache/protocol/commands/ClearCommand.java +++ b/src/main/java/org/cache/protocol/commands/ClearCommand.java @@ -1,14 +1,14 @@ package org.cache.protocol.commands; import org.cache.core.Cache; -import org.cache.protocol.codec.Codec; +import org.cache.protocol.codec.ValueCodecRegistry; import static org.cache.protocol.commands.ResponseConstants.OK; -public class ClearCommand implements CacheCommand { +public class ClearCommand implements CacheCommand { @Override - public String process(Cache cache, Codec valueCodec) { + public String process(Cache cache, ValueCodecRegistry valueCodecs) { cache.clear(); return OK.name(); } diff --git a/src/main/java/org/cache/protocol/commands/CommandType.java b/src/main/java/org/cache/protocol/commands/CommandType.java index a5c2440..bd6dc62 100644 --- a/src/main/java/org/cache/protocol/commands/CommandType.java +++ b/src/main/java/org/cache/protocol/commands/CommandType.java @@ -7,5 +7,6 @@ public enum CommandType { SIZE, CLEAR, METRICS, + PUSH, UNKNOWN } diff --git a/src/main/java/org/cache/protocol/commands/DeleteCommand.java b/src/main/java/org/cache/protocol/commands/DeleteCommand.java index f63cef7..a2651c6 100644 --- a/src/main/java/org/cache/protocol/commands/DeleteCommand.java +++ b/src/main/java/org/cache/protocol/commands/DeleteCommand.java @@ -1,11 +1,11 @@ package org.cache.protocol.commands; import org.cache.core.Cache; -import org.cache.protocol.codec.Codec; +import org.cache.protocol.codec.ValueCodecRegistry; import static org.cache.protocol.commands.ResponseConstants.OK; -public class DeleteCommand implements CacheCommand { +public class DeleteCommand implements CacheCommand { private final K key; @@ -14,7 +14,7 @@ public DeleteCommand(K key) { } @Override - public String process(Cache cache, Codec valueCodec) { + public String process(Cache cache, ValueCodecRegistry valueCodecs) { cache.delete(key); return OK.name(); } diff --git a/src/main/java/org/cache/protocol/commands/GetCommand.java b/src/main/java/org/cache/protocol/commands/GetCommand.java index 917417e..7d55d1e 100644 --- a/src/main/java/org/cache/protocol/commands/GetCommand.java +++ b/src/main/java/org/cache/protocol/commands/GetCommand.java @@ -1,12 +1,16 @@ package org.cache.protocol.commands; import org.cache.core.Cache; -import org.cache.protocol.codec.Codec; +import org.cache.core.CacheEntry; +import org.cache.protocol.codec.ValueCodec; +import org.cache.protocol.codec.ValueCodecRegistry; + +import java.util.Optional; import static org.cache.protocol.commands.ResponseConstants.NOT_FOUND; import static org.cache.protocol.commands.ResponseConstants.VALUE; -public class GetCommand implements CacheCommand { +public class GetCommand implements CacheCommand { private final K key; @@ -15,9 +19,15 @@ public GetCommand(K key) { } @Override - public String process(Cache cache, Codec valueCodec) { - return cache.get(key) - .map(value -> VALUE.name() + " " + valueCodec.encode(value)) - .orElse(NOT_FOUND.name()); + public String process(Cache cache, ValueCodecRegistry valueCodecs) { + Optional entry = cache.get(key); + + if (entry.isEmpty()) { + return NOT_FOUND.name(); + } + + CacheEntry cacheEntry = entry.get(); + ValueCodec codec = valueCodecs.get(cacheEntry.getType()); + return VALUE.name() + " " + codec.toString(cacheEntry.getValue()); } } diff --git a/src/main/java/org/cache/protocol/commands/InvalidCommand.java b/src/main/java/org/cache/protocol/commands/InvalidCommand.java index 8229d55..dddf3d9 100644 --- a/src/main/java/org/cache/protocol/commands/InvalidCommand.java +++ b/src/main/java/org/cache/protocol/commands/InvalidCommand.java @@ -1,11 +1,11 @@ package org.cache.protocol.commands; import org.cache.core.Cache; -import org.cache.protocol.codec.Codec; +import org.cache.protocol.codec.ValueCodecRegistry; import static org.cache.protocol.commands.ResponseConstants.ERROR; -public class InvalidCommand implements CacheCommand { +public class InvalidCommand implements CacheCommand { private final String message; @@ -14,7 +14,7 @@ public InvalidCommand(String message) { } @Override - public String process(Cache cache, Codec valueCodec) { + public String process(Cache cache, ValueCodecRegistry valueCodecs) { return ERROR.name() + " " + message; } } diff --git a/src/main/java/org/cache/protocol/commands/MetricsCommand.java b/src/main/java/org/cache/protocol/commands/MetricsCommand.java index 8509ff7..d537eb8 100644 --- a/src/main/java/org/cache/protocol/commands/MetricsCommand.java +++ b/src/main/java/org/cache/protocol/commands/MetricsCommand.java @@ -2,14 +2,14 @@ import org.cache.core.Cache; import org.cache.core.metrics.Snapshot; -import org.cache.protocol.codec.Codec; +import org.cache.protocol.codec.ValueCodecRegistry; import static org.cache.protocol.commands.ResponseConstants.METRICS; -public class MetricsCommand implements CacheCommand { +public class MetricsCommand implements CacheCommand { @Override - public String process(Cache cache, Codec valueCodec) { + public String process(Cache cache, ValueCodecRegistry valueCodecs) { Snapshot metrics = cache.metrics(); return METRICS.name() + " hits=" + metrics.getHits() + " misses=" + metrics.getMisses() diff --git a/src/main/java/org/cache/protocol/commands/PushCommand.java b/src/main/java/org/cache/protocol/commands/PushCommand.java new file mode 100644 index 0000000..e6bcc64 --- /dev/null +++ b/src/main/java/org/cache/protocol/commands/PushCommand.java @@ -0,0 +1,53 @@ +package org.cache.protocol.commands; + +import org.cache.core.Cache; +import org.cache.core.CacheEntry; +import org.cache.core.ValueType; +import org.cache.protocol.codec.ValueCodec; +import org.cache.protocol.codec.ValueCodecRegistry; + +import java.util.ArrayList; +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 PushCommand implements CacheCommand { + private final K key; + private final String value; + private final ValueType type; + + public PushCommand(K key, String value, ValueType type) { + this.key = key; + this.value = value; + this.type = type; + } + + @SuppressWarnings("unchecked") + @Override + public String process(Cache cache, ValueCodecRegistry valueCodecs) { + Optional existingList = cache.get(key); + ValueCodec> listCodec = (ValueCodec>) valueCodecs.get(ValueType.LIST); + + List list; + byte[] finalValue; + + if (existingList.isEmpty() || existingList.get().getValue() == null) { + list = new ArrayList<>(); + } else { + CacheEntry entry = existingList.get(); + if (entry.getType() != ValueType.LIST) { + return ERROR.name() + " key contains " + entry.getType().name().toLowerCase() + " value"; + } + + list = listCodec.decode(entry.getValue()); + } + + list.add(value); + finalValue = listCodec.encode(list); + + cache.put(key, finalValue, type, 0); + return OK.name(); + } +} diff --git a/src/main/java/org/cache/protocol/commands/PutCommand.java b/src/main/java/org/cache/protocol/commands/PutCommand.java index a3b7d41..b689f47 100644 --- a/src/main/java/org/cache/protocol/commands/PutCommand.java +++ b/src/main/java/org/cache/protocol/commands/PutCommand.java @@ -1,29 +1,36 @@ package org.cache.protocol.commands; import org.cache.core.Cache; -import org.cache.protocol.codec.Codec; +import org.cache.core.ValueType; +import org.cache.protocol.codec.ValueCodec; +import org.cache.protocol.codec.ValueCodecRegistry; + +import java.util.Arrays; import static org.cache.protocol.commands.ResponseConstants.OK; -public class PutCommand implements CacheCommand { +public class PutCommand implements CacheCommand { private final K key; - private final V value; + private final String value; + private final ValueType type; private final long ttlMillis; - public PutCommand(K key, V value, long ttlMillis) { + public PutCommand(K key, String value, ValueType type, long ttlMillis) { this.key = key; this.value = value; + this.type = type; this.ttlMillis = ttlMillis; } + @SuppressWarnings("unchecked") @Override - public String process(Cache cache, Codec valueCodec) { - cache.put(key, value, ttlMillis); - return OK.name(); - } + public String process(Cache cache, ValueCodecRegistry valueCodecs) { + ValueCodec stringCodec = (ValueCodec) valueCodecs.get(ValueType.STRING); - public V getValue() { - return value; + byte[] finalValue = stringCodec.encode(value); + cache.put(key, finalValue, type, ttlMillis); + + return OK.name(); } } diff --git a/src/main/java/org/cache/protocol/commands/SizeCommand.java b/src/main/java/org/cache/protocol/commands/SizeCommand.java index 7b06fb5..c246528 100644 --- a/src/main/java/org/cache/protocol/commands/SizeCommand.java +++ b/src/main/java/org/cache/protocol/commands/SizeCommand.java @@ -1,14 +1,14 @@ package org.cache.protocol.commands; import org.cache.core.Cache; -import org.cache.protocol.codec.Codec; +import org.cache.protocol.codec.ValueCodecRegistry; import static org.cache.protocol.commands.ResponseConstants.SIZE; -public class SizeCommand implements CacheCommand { +public class SizeCommand implements CacheCommand { @Override - public String process(Cache cache, Codec valueCodec) { + public String process(Cache cache, ValueCodecRegistry valueCodecs) { return SIZE.name() + " " + cache.size(); } } diff --git a/src/main/java/org/cache/protocol/commands/UnknownCommand.java b/src/main/java/org/cache/protocol/commands/UnknownCommand.java index f0e82cd..423424b 100644 --- a/src/main/java/org/cache/protocol/commands/UnknownCommand.java +++ b/src/main/java/org/cache/protocol/commands/UnknownCommand.java @@ -1,14 +1,14 @@ package org.cache.protocol.commands; import org.cache.core.Cache; -import org.cache.protocol.codec.Codec; +import org.cache.protocol.codec.ValueCodecRegistry; import static org.cache.protocol.commands.ResponseConstants.ERROR; -public class UnknownCommand implements CacheCommand { +public class UnknownCommand implements CacheCommand { @Override - public String process(Cache cache, Codec valueCodec) { + public String process(Cache cache, ValueCodecRegistry valueCodecs) { return ERROR.name() + " unknown command"; } } diff --git a/src/main/java/org/cache/util/ConverterUtil.java b/src/main/java/org/cache/util/ConverterUtil.java new file mode 100644 index 0000000..b063e51 --- /dev/null +++ b/src/main/java/org/cache/util/ConverterUtil.java @@ -0,0 +1,10 @@ +package org.cache.util; + +import java.nio.charset.StandardCharsets; + +public final class ConverterUtil { + + public static byte[] toByteArray(String value) { + return value.getBytes(StandardCharsets.UTF_8); + } +}