From 085b1757d7a798b62915ef6af28cc2df6921a0f5 Mon Sep 17 00:00:00 2001 From: ferid333 <135500346+ferid333@users.noreply.github.com> Date: Mon, 27 Jul 2026 01:06:08 +1000 Subject: [PATCH 1/2] Phase-5: Adding TCP client and different connection types --- build.gradle | 3 +- .../java/org/cache/client/CacheClient.java | 22 +++ .../cache/client/CacheClientException.java | 12 ++ .../java/org/cache/client/TcpCacheClient.java | 161 ++++++++++++++++++ .../java/org/cache/core/metrics/Snapshot.java | 4 +- .../org/cache/network/TcpCacheServer.java | 1 + .../ClientConnectionHandler.java | 31 ++-- .../connection/LineProtocolConnection.java | 86 ++++++++++ .../connection/ProtocolConnection.java | 13 ++ .../connection/ProtocolConnectionFactory.java | 35 ++++ .../network/connection/RespConnection.java | 154 +++++++++++++++++ .../org/cache/protocol/CommandParser.java | 50 +++--- .../org/cache/protocol/CommandProcessor.java | 7 + .../org/cache/protocol/ProtocolConstants.java | 16 ++ 14 files changed, 556 insertions(+), 39 deletions(-) create mode 100644 src/main/java/org/cache/client/CacheClient.java create mode 100644 src/main/java/org/cache/client/CacheClientException.java create mode 100644 src/main/java/org/cache/client/TcpCacheClient.java rename src/main/java/org/cache/network/{ => connection}/ClientConnectionHandler.java (53%) create mode 100644 src/main/java/org/cache/network/connection/LineProtocolConnection.java create mode 100644 src/main/java/org/cache/network/connection/ProtocolConnection.java create mode 100644 src/main/java/org/cache/network/connection/ProtocolConnectionFactory.java create mode 100644 src/main/java/org/cache/network/connection/RespConnection.java create mode 100644 src/main/java/org/cache/protocol/ProtocolConstants.java diff --git a/build.gradle b/build.gradle index 51276ba..7fc491e 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 { @@ -25,4 +26,4 @@ tasks.withType(JavaCompile).configureEach { test { useJUnitPlatform() -} \ 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..479e99f --- /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); + + boolean 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..1f62a45 --- /dev/null +++ b/src/main/java/org/cache/client/TcpCacheClient.java @@ -0,0 +1,161 @@ +package org.cache.client; + +import org.cache.core.metrics.Snapshot; +import org.cache.network.connection.RespConnection; +import org.cache.protocol.codec.Codec; +import org.cache.protocol.codec.StringCodec; +import org.cache.protocol.commands.ResponseConstants; + +import java.io.IOException; +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 final RespConnection connection; + private final Codec keyCodec; + private final Codec valueCodec; + + @SuppressWarnings("unchecked") + public TcpCacheClient(String address, int port) { + this( + address, + port, + (Codec) new StringCodec(), + (Codec) new StringCodec() + ); + } + + public TcpCacheClient(String address, int port, Codec keyCodec, Codec valueCodec) { + try { + this.connection = new RespConnection(new Socket(address, port)); + this.keyCodec = keyCodec; + this.valueCodec = valueCodec; + } 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), valueCodec.encode(value), Long.toString(ttl))); + } + + @Override + public void put(K key, V value) { + expectOk(List.of("PUT", keyCodec.encode(key), valueCodec.encode(value))); + } + + @Override + public Optional get(K key) { + String response = send(List.of("GET", keyCodec.encode(key))); + + if (response.equals(ResponseConstants.NOT_FOUND.name())) { + return Optional.empty(); + } + + String prefix = ResponseConstants.VALUE.name() + " "; + if (response.startsWith(prefix)) { + return Optional.of(valueCodec.decode(response.substring(prefix.length()))); + } + + throw protocolException(response); + } + + @Override + public boolean delete(K key) { + expectOk(List.of("DELETE", keyCodec.encode(key))); + return true; + } + + @Override + public Snapshot metrics() { + String response = send(List.of("METRICS")); + String prefix = ResponseConstants.METRICS.name() + " "; + + if (!response.startsWith(prefix)) { + throw protocolException(response); + } + + long hits = 0; + long misses = 0; + long evictions = 0; + long expirations = 0; + double hitRate = 0; + + String[] metrics = response.substring(prefix.length()).split("\\s+"); + for (String metric : metrics) { + String[] parts = metric.split("=", 2); + if (parts.length != 2) { + throw protocolException(response); + } + + switch (parts[0]) { + case "hits" -> hits = Long.parseLong(parts[1]); + case "misses" -> misses = Long.parseLong(parts[1]); + case "evictions" -> evictions = Long.parseLong(parts[1]); + case "expirations" -> expirations = Long.parseLong(parts[1]); + case "hitRate" -> hitRate = Double.parseDouble(parts[1]); + default -> throw protocolException(response); + } + } + + return new Snapshot(hits, misses, evictions, expirations, hitRate); + } + + @Override + public int size() { + String response = send(List.of("SIZE")); + String prefix = ResponseConstants.SIZE.name() + " "; + + if (!response.startsWith(prefix)) { + throw protocolException(response); + } + + return Integer.parseInt(response.substring(prefix.length())); + } + + @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) { + String response = send(command); + + if (!response.equals(OK.name())) { + throw protocolException(response); + } + } + + private String send(List command) { + try { + String response = connection.sendCommand(command); + + if (response.startsWith(ERROR.name() + " ")) { + throw protocolException(response); + } + + return response; + } catch (IOException exception) { + throw new CacheClientException("failed to read response from cache server", exception); + } + } + + private CacheClientException protocolException(String response) { + return new CacheClientException("unexpected cache server response: " + response); + } +} 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/TcpCacheServer.java b/src/main/java/org/cache/network/TcpCacheServer.java index c805f83..431b4af 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; diff --git a/src/main/java/org/cache/network/ClientConnectionHandler.java b/src/main/java/org/cache/network/connection/ClientConnectionHandler.java similarity index 53% rename from src/main/java/org/cache/network/ClientConnectionHandler.java rename to src/main/java/org/cache/network/connection/ClientConnectionHandler.java index b48a85b..47b5349 100644 --- a/src/main/java/org/cache/network/ClientConnectionHandler.java +++ b/src/main/java/org/cache/network/connection/ClientConnectionHandler.java @@ -1,13 +1,10 @@ -package org.cache.network; +package org.cache.network.connection; 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; +import java.util.List; public class ClientConnectionHandler implements Runnable { @@ -21,18 +18,22 @@ public ClientConnectionHandler(Socket socket, CommandProcessor commandProc @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) { + 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); - writer.println(response); + 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..c76fc1e --- /dev/null +++ b/src/main/java/org/cache/network/connection/RespConnection.java @@ -0,0 +1,154 @@ +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.*; + +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 { + byte[] bytes = value.getBytes(UTF_8); + writeAscii(BULK_STRING_PREFIX + Integer.toString(bytes.length) + CRLF); + output.write(bytes); + writeAscii(CRLF); + output.flush(); + } + + @Override + public String send(String value) throws IOException { + return sendCommand(List.of(value.trim().split("\\s+"))); + } + + public String sendCommand(List commandParts) throws IOException { + writeArray(commandParts); + String response = readString(); + + 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 { + int type = input.read(); + + if (type == -1) { + return null; + } + + return switch (type) { + case SIMPLE_STRING_PREFIX, INTEGER_PREFIX -> lines.readLine(); + case ERROR_PREFIX -> "ERROR " + lines.readLine(); + case BULK_STRING_PREFIX -> readBulkStringAfterType(); + default -> throw new IOException("Unsupported RESP response type: " + (char) type); + }; + } + + 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)); + } + + @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..cfec5f8 100644 --- a/src/main/java/org/cache/protocol/CommandParser.java +++ b/src/main/java/org/cache/protocol/CommandParser.java @@ -3,7 +3,7 @@ import org.cache.protocol.codec.Codec; import org.cache.protocol.commands.*; -import java.util.concurrent.TimeUnit; +import java.util.List; public class CommandParser { @@ -26,10 +26,16 @@ public CacheCommand parse(String rawCommand) { return new UnknownCommand<>(); } - String[] parts = rawCommand.trim().split("\\s+"); + return parse(List.of(rawCommand.trim().split("\\s+"))); + } + + public CacheCommand parse(List parts) { + if (parts == null || parts.isEmpty()) { + return new UnknownCommand<>(); + } try { - CommandType type = CommandType.valueOf(parts[0].toUpperCase()); + CommandType type = CommandType.valueOf(parts.get(0).toUpperCase()); return switch (type) { case PUT -> parsePut(parts); case GET -> parseGet(parts); @@ -44,56 +50,58 @@ 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() != THREE && parts.size() != FOUR) { + return new InvalidCommand<>("usage: PUT key value [ttlMillis]"); } try { + long ttlMillis = parts.size() == FOUR ? Long.parseLong(parts.get(THREE)) : 0; + return new PutCommand<>( - keyCodec.decode(parts[ONE]), - valueCodec.decode(parts[TWO]), - Long.parseLong(parts[THREE]) + keyCodec.decode(parts.get(ONE)), + valueCodec.decode(parts.get(TWO)), + ttlMillis ); } catch (NumberFormatException exception) { return new InvalidCommand<>("ttl must be a number"); } } - private CacheCommand parseGet(String[] parts) { - if (parts.length != TWO) { + private CacheCommand parseGet(List parts) { + if (parts.size() != TWO) { return new InvalidCommand<>("usage: GET key"); } - return new GetCommand<>(keyCodec.decode(parts[1])); + return new GetCommand<>(keyCodec.decode(parts.get(ONE))); } - private CacheCommand parseDelete(String[] parts) { - if (parts.length != TWO) { + private CacheCommand parseDelete(List parts) { + if (parts.size() != TWO) { return new InvalidCommand<>("usage: DELETE key"); } - return new DeleteCommand<>(keyCodec.decode(parts[1])); + return new DeleteCommand<>(keyCodec.decode(parts.get(ONE))); } - private CacheCommand parseSize(String[] parts) { - if (parts.length != ONE) { + private CacheCommand parseSize(List parts) { + if (parts.size() != ONE) { 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() != ONE) { 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() != ONE) { 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..7217dc4 100644 --- a/src/main/java/org/cache/protocol/CommandProcessor.java +++ b/src/main/java/org/cache/protocol/CommandProcessor.java @@ -4,6 +4,8 @@ import org.cache.protocol.codec.Codec; import org.cache.protocol.commands.CacheCommand; +import java.util.List; + public class CommandProcessor { private final Cache cache; @@ -20,4 +22,9 @@ 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, valueCodec); + } } 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() { + } +} From 0eb9eda3b18594eba741d1f9684e3ac47b6f65c9 Mon Sep 17 00:00:00 2001 From: ferid333 <135500346+ferid333@users.noreply.github.com> Date: Thu, 30 Jul 2026 20:22:09 +1000 Subject: [PATCH 2/2] Phase-5: Adding TCP client and different connection types --- build.gradle | 4 +- src/main/java/org/cache/Main.java | 19 ++- .../java/org/cache/client/CacheClient.java | 2 +- .../java/org/cache/client/TcpCacheClient.java | 138 ++++++++++-------- .../cache/client/serializer/Serializer.java | 8 + .../client/serializer/StringSerializer.java | 22 +++ src/main/java/org/cache/core/Cache.java | 6 +- src/main/java/org/cache/core/CacheEntry.java | 20 ++- src/main/java/org/cache/core/LocalCache.java | 14 +- src/main/java/org/cache/core/ValueType.java | 7 + .../org/cache/network/TcpCacheServer.java | 8 +- .../connection/ClientConnectionHandler.java | 4 +- .../network/connection/RespConnection.java | 87 +++++++++-- .../org/cache/protocol/CommandParser.java | 84 ++++++----- .../org/cache/protocol/CommandProcessor.java | 23 ++- .../java/org/cache/protocol/codec/Codec.java | 8 - .../codec/CodecConversionException.java | 7 + .../cache/protocol/codec/IntegerKeyCodec.java | 26 ++++ .../org/cache/protocol/codec/KeyCodec.java | 8 + .../cache/protocol/codec/ListValueCodec.java | 47 ++++++ .../{StringCodec.java => StringKeyCodec.java} | 2 +- .../protocol/codec/StringValueCodec.java | 29 ++++ .../org/cache/protocol/codec/ValueCodec.java | 10 ++ .../protocol/codec/ValueCodecRegistry.java | 25 ++++ .../cache/protocol/commands/CacheCommand.java | 6 +- .../cache/protocol/commands/ClearCommand.java | 6 +- .../cache/protocol/commands/CommandType.java | 1 + .../protocol/commands/DeleteCommand.java | 6 +- .../cache/protocol/commands/GetCommand.java | 22 ++- .../protocol/commands/InvalidCommand.java | 6 +- .../protocol/commands/MetricsCommand.java | 6 +- .../cache/protocol/commands/PushCommand.java | 53 +++++++ .../cache/protocol/commands/PutCommand.java | 27 ++-- .../cache/protocol/commands/SizeCommand.java | 6 +- .../protocol/commands/UnknownCommand.java | 6 +- .../java/org/cache/util/ConverterUtil.java | 10 ++ 36 files changed, 568 insertions(+), 195 deletions(-) create mode 100644 src/main/java/org/cache/client/serializer/Serializer.java create mode 100644 src/main/java/org/cache/client/serializer/StringSerializer.java create mode 100644 src/main/java/org/cache/core/ValueType.java delete mode 100644 src/main/java/org/cache/protocol/codec/Codec.java create mode 100644 src/main/java/org/cache/protocol/codec/CodecConversionException.java create mode 100644 src/main/java/org/cache/protocol/codec/IntegerKeyCodec.java create mode 100644 src/main/java/org/cache/protocol/codec/KeyCodec.java create mode 100644 src/main/java/org/cache/protocol/codec/ListValueCodec.java rename src/main/java/org/cache/protocol/codec/{StringCodec.java => StringKeyCodec.java} (77%) create mode 100644 src/main/java/org/cache/protocol/codec/StringValueCodec.java create mode 100644 src/main/java/org/cache/protocol/codec/ValueCodec.java create mode 100644 src/main/java/org/cache/protocol/codec/ValueCodecRegistry.java create mode 100644 src/main/java/org/cache/protocol/commands/PushCommand.java create mode 100644 src/main/java/org/cache/util/ConverterUtil.java diff --git a/build.gradle b/build.gradle index 7fc491e..eb4ae53 100644 --- a/build.gradle +++ b/build.gradle @@ -21,9 +21,9 @@ application { } tasks.withType(JavaCompile).configureEach { - options.release = 17 + options.release = 21 } test { useJUnitPlatform() -} +} \ No newline at end of file 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 index 479e99f..8912bae 100644 --- a/src/main/java/org/cache/client/CacheClient.java +++ b/src/main/java/org/cache/client/CacheClient.java @@ -12,7 +12,7 @@ public interface CacheClient { Optional get(K key); - boolean delete(K key); + void delete(K key); Snapshot metrics(); diff --git a/src/main/java/org/cache/client/TcpCacheClient.java b/src/main/java/org/cache/client/TcpCacheClient.java index 1f62a45..fdc3b01 100644 --- a/src/main/java/org/cache/client/TcpCacheClient.java +++ b/src/main/java/org/cache/client/TcpCacheClient.java @@ -1,12 +1,13 @@ 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.Codec; -import org.cache.protocol.codec.StringCodec; +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; @@ -16,25 +17,30 @@ 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 Codec keyCodec; - private final Codec valueCodec; - - @SuppressWarnings("unchecked") - public TcpCacheClient(String address, int port) { - this( - address, - port, - (Codec) new StringCodec(), - (Codec) new StringCodec() - ); + 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, Codec keyCodec, Codec valueCodec) { + public TcpCacheClient( + String address, + int port, + KeyCodec keyCodec, + Serializer valueSerializer, + int connectTimeoutMillis, + int readTimeoutMillis + ) { try { - this.connection = new RespConnection(new Socket(address, port)); + this.connection = new RespConnection(connect(address, port, connectTimeoutMillis, readTimeoutMillis)); this.keyCodec = keyCodec; - this.valueCodec = valueCodec; + this.valueSerializer = valueSerializer; } catch (IOException exception) { throw new CacheClientException("failed to connect to cache server", exception); } @@ -42,43 +48,40 @@ public TcpCacheClient(String address, int port, Codec keyCodec, Codec valu @Override public void put(K key, V value, long ttl) { - expectOk(List.of("PUT", keyCodec.encode(key), valueCodec.encode(value), Long.toString(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), valueCodec.encode(value))); + expectOk(List.of("PUT", keyCodec.encode(key), valueSerializer.encode(value))); } @Override public Optional get(K key) { - String response = send(List.of("GET", keyCodec.encode(key))); + List response = send(List.of("GET", keyCodec.encode(key))); - if (response.equals(ResponseConstants.NOT_FOUND.name())) { + if (isResponse(response, ResponseConstants.NOT_FOUND.name(), 1)) { return Optional.empty(); } - String prefix = ResponseConstants.VALUE.name() + " "; - if (response.startsWith(prefix)) { - return Optional.of(valueCodec.decode(response.substring(prefix.length()))); + if (isResponse(response, ResponseConstants.VALUE.name(), 2)) { + return Optional.of(valueSerializer.decode(response.get(1))); } - throw protocolException(response); + throw new CacheClientException("unexpected cache server response: " + response); } @Override - public boolean delete(K key) { + public void delete(K key) { expectOk(List.of("DELETE", keyCodec.encode(key))); - return true; } @Override public Snapshot metrics() { - String response = send(List.of("METRICS")); - String prefix = ResponseConstants.METRICS.name() + " "; + List response = send(List.of("METRICS")); - if (!response.startsWith(prefix)) { - throw protocolException(response); + if (response.isEmpty() || !response.get(0).equals(ResponseConstants.METRICS.name())) { + throw new CacheClientException("unexpected cache server response: " + response); } long hits = 0; @@ -87,20 +90,25 @@ public Snapshot metrics() { long expirations = 0; double hitRate = 0; - String[] metrics = response.substring(prefix.length()).split("\\s+"); - for (String metric : metrics) { - String[] parts = metric.split("=", 2); - if (parts.length != 2) { - throw protocolException(response); - } + if ((response.size() - 1) % 2 != 0) { + throw new CacheClientException("unexpected cache server response: " + response); + } - switch (parts[0]) { - case "hits" -> hits = Long.parseLong(parts[1]); - case "misses" -> misses = Long.parseLong(parts[1]); - case "evictions" -> evictions = Long.parseLong(parts[1]); - case "expirations" -> expirations = Long.parseLong(parts[1]); - case "hitRate" -> hitRate = Double.parseDouble(parts[1]); - default -> throw protocolException(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); } } @@ -109,14 +117,13 @@ public Snapshot metrics() { @Override public int size() { - String response = send(List.of("SIZE")); - String prefix = ResponseConstants.SIZE.name() + " "; + List response = send(List.of("SIZE")); - if (!response.startsWith(prefix)) { - throw protocolException(response); + if (!isResponse(response, ResponseConstants.SIZE.name(), 2)) { + throw new CacheClientException("unexpected cache server response: " + response); } - return Integer.parseInt(response.substring(prefix.length())); + return Integer.parseInt(response.get(1)); } @Override @@ -134,19 +141,19 @@ public void close() { } private void expectOk(List command) { - String response = send(command); + List response = send(command); - if (!response.equals(OK.name())) { - throw protocolException(response); + if (!isResponse(response, OK.name(), 1)) { + throw new CacheClientException("unexpected cache server response: " + response); } } - private String send(List command) { + private List send(List command) { try { - String response = connection.sendCommand(command); + List response = connection.sendCommandForResponse(command); - if (response.startsWith(ERROR.name() + " ")) { - throw protocolException(response); + if (!response.isEmpty() && response.getFirst().equals(ERROR.name())) { + throw new CacheClientException("unexpected cache server response: " + response); } return response; @@ -155,7 +162,24 @@ private String send(List command) { } } - private CacheClientException protocolException(String response) { - return new CacheClientException("unexpected cache server response: " + response); + 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/network/TcpCacheServer.java b/src/main/java/org/cache/network/TcpCacheServer.java index 431b4af..8609ebe 100644 --- a/src/main/java/org/cache/network/TcpCacheServer.java +++ b/src/main/java/org/cache/network/TcpCacheServer.java @@ -12,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 index 47b5349..d623dda 100644 --- a/src/main/java/org/cache/network/connection/ClientConnectionHandler.java +++ b/src/main/java/org/cache/network/connection/ClientConnectionHandler.java @@ -9,9 +9,9 @@ public class ClientConnectionHandler implements Runnable { private final Socket socket; - private final CommandProcessor commandProcessor; + private final CommandProcessor commandProcessor; - public ClientConnectionHandler(Socket socket, CommandProcessor commandProcessor) { + public ClientConnectionHandler(Socket socket, CommandProcessor commandProcessor) { this.socket = socket; this.commandProcessor = commandProcessor; } diff --git a/src/main/java/org/cache/network/connection/RespConnection.java b/src/main/java/org/cache/network/connection/RespConnection.java index c76fc1e..4406436 100644 --- a/src/main/java/org/cache/network/connection/RespConnection.java +++ b/src/main/java/org/cache/network/connection/RespConnection.java @@ -10,6 +10,12 @@ 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 { @@ -53,11 +59,7 @@ public List readCommand() throws IOException { @Override public void write(String value) throws IOException { - byte[] bytes = value.getBytes(UTF_8); - writeAscii(BULK_STRING_PREFIX + Integer.toString(bytes.length) + CRLF); - output.write(bytes); - writeAscii(CRLF); - output.flush(); + writeArray(responseParts(value)); } @Override @@ -66,8 +68,13 @@ public String send(String value) throws IOException { } public String sendCommand(List commandParts) throws IOException { + List response = sendCommandForResponse(commandParts); + return String.join(" ", response); + } + + public List sendCommandForResponse(List commandParts) throws IOException { writeArray(commandParts); - String response = readString(); + List response = readResponse(); if (response == null) { throw new IOException("Connection closed"); @@ -90,6 +97,16 @@ public void writeArray(List values) throws IOException { } 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) { @@ -97,13 +114,25 @@ private String readString() throws IOException { } return switch (type) { - case SIMPLE_STRING_PREFIX, INTEGER_PREFIX -> lines.readLine(); - case ERROR_PREFIX -> "ERROR " + lines.readLine(); - case BULK_STRING_PREFIX -> readBulkStringAfterType(); + 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(); @@ -147,6 +176,46 @@ 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 cfec5f8..15e2d82 100644 --- a/src/main/java/org/cache/protocol/CommandParser.java +++ b/src/main/java/org/cache/protocol/CommandParser.java @@ -1,41 +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.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()) { - return new UnknownCommand<>(); - } - - return parse(List.of(rawCommand.trim().split("\\s+"))); - } - - public CacheCommand parse(List parts) { + public CacheCommand parse(List parts) { if (parts == null || parts.isEmpty()) { return new UnknownCommand<>(); } try { - CommandType type = CommandType.valueOf(parts.get(0).toUpperCase()); + CommandType type = CommandType.valueOf(parts.get(COMMAND_INDEX).toUpperCase()); return switch (type) { case PUT -> parsePut(parts); case GET -> parseGet(parts); @@ -43,6 +38,7 @@ public CacheCommand parse(List parts) { case SIZE -> parseSize(parts); case CLEAR -> parseClear(parts); case METRICS -> parseMetrics(parts); + case PUSH -> parsePush(parts); case UNKNOWN -> new UnknownCommand<>(); }; } catch (IllegalArgumentException exception) { @@ -50,17 +46,19 @@ public CacheCommand parse(List parts) { } } - private CacheCommand parsePut(List parts) { - if (parts.size() != THREE && parts.size() != FOUR) { + + 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() == FOUR ? Long.parseLong(parts.get(THREE)) : 0; + long ttlMillis = parts.size() == TTL_COMMAND_PARTS ? Long.parseLong(parts.get(TTL_INDEX)) : 0; return new PutCommand<>( - keyCodec.decode(parts.get(ONE)), - valueCodec.decode(parts.get(TWO)), + keyCodec.decode(parts.get(KEY_INDEX)), + parts.get(VALUE_INDEX), + ValueType.STRING, ttlMillis ); } catch (NumberFormatException exception) { @@ -68,40 +66,52 @@ private CacheCommand parsePut(List parts) { } } - private CacheCommand parseGet(List parts) { - if (parts.size() != 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.get(ONE))); + return new GetCommand<>(keyCodec.decode(parts.get(KEY_INDEX))); } - private CacheCommand parseDelete(List parts) { - if (parts.size() != TWO) { + private CacheCommand parseDelete(List parts) { + if (parts.size() != KEY_COMMAND_PARTS) { return new InvalidCommand<>("usage: DELETE key"); } - return new DeleteCommand<>(keyCodec.decode(parts.get(ONE))); + return new DeleteCommand<>(keyCodec.decode(parts.get(KEY_INDEX))); } - private CacheCommand parseSize(List parts) { - if (parts.size() != ONE) { + private CacheCommand parseSize(List parts) { + if (parts.size() != COMMAND_PARTS) { return new InvalidCommand<>("usage: SIZE"); } return new SizeCommand<>(); } - private CacheCommand parseClear(List parts) { - if (parts.size() != ONE) { + private CacheCommand parseClear(List parts) { + if (parts.size() != COMMAND_PARTS) { return new InvalidCommand<>("usage: CLEAR"); } return new ClearCommand<>(); } - private CacheCommand parseMetrics(List parts) { - if (parts.size() != 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 7217dc4..9688c6e 100644 --- a/src/main/java/org/cache/protocol/CommandProcessor.java +++ b/src/main/java/org/cache/protocol/CommandProcessor.java @@ -1,30 +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; import java.util.List; -public class CommandProcessor { +public class CommandProcessor { - private final Cache cache; - private final CommandParser parser; - private final Codec valueCodec; + private final Cache cache; + private final CommandParser parser; + private final ValueCodecRegistry valueCodecs; - public CommandProcessor(Cache cache, CommandParser parser, Codec valueCodec) { + public CommandProcessor(Cache cache, CommandParser parser, ValueCodecRegistry valueCodecs) { this.cache = cache; this.parser = parser; - this.valueCodec = valueCodec; - } - - public String process(String rawCommand) { - CacheCommand command = parser.parse(rawCommand); - return command.process(cache, valueCodec); + this.valueCodecs = valueCodecs; } public String process(List commandParts) { - CacheCommand command = parser.parse(commandParts); - return command.process(cache, valueCodec); + CacheCommand command = parser.parse(commandParts); + return command.process(cache, valueCodecs); } } 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); + } +}