diff --git a/src/main/java/org/cache/cluster/CacheNode.java b/src/main/java/org/cache/cluster/CacheNode.java index 5bad979..440c8f7 100644 --- a/src/main/java/org/cache/cluster/CacheNode.java +++ b/src/main/java/org/cache/cluster/CacheNode.java @@ -5,6 +5,11 @@ public record CacheNode( String host, int httpPort, int tcpPort, - int clusterPort + int clusterPort, + NodeStatus status ) { + + public CacheNode(String id, String host, int httpPort, int tcpPort, int clusterPort) { + this(id, host, httpPort, tcpPort, clusterPort, NodeStatus.HEALTHY); + } } diff --git a/src/main/java/org/cache/cluster/NodeStatus.java b/src/main/java/org/cache/cluster/NodeStatus.java new file mode 100644 index 0000000..6424c6e --- /dev/null +++ b/src/main/java/org/cache/cluster/NodeStatus.java @@ -0,0 +1,7 @@ +package org.cache.cluster; + +public enum NodeStatus { + HEALTHY, + SUSPECTED, + UNAVAILABLE +} diff --git a/src/main/java/org/cache/config/CacheConfigLoader.java b/src/main/java/org/cache/config/CacheConfigLoader.java index dba4a1d..de1787d 100644 --- a/src/main/java/org/cache/config/CacheConfigLoader.java +++ b/src/main/java/org/cache/config/CacheConfigLoader.java @@ -2,6 +2,7 @@ import org.cache.cluster.CacheNode; import org.cache.cluster.ClusterInfo; +import org.cache.cluster.NodeStatus; import org.cache.eviction.EvictionPolicy; import org.cache.eviction.EvictionPolicyType; import org.cache.eviction.LruEvictionPolicy; @@ -84,7 +85,8 @@ private CacheNode buildCacheNode(Properties properties) { getString(properties, ConfigKey.merge(ConfigKey.NODE, ConfigKey.HOST), DEFAULT_NODE_HOST), getInt(properties, ConfigKey.merge(ConfigKey.NODE, ConfigKey.HTTP_PORT), DEFAULT_HTTP_PORT), getInt(properties, ConfigKey.merge(ConfigKey.NODE, ConfigKey.TCP_PORT), DEFAULT_TCP_PORT), - getInt(properties, ConfigKey.merge(ConfigKey.NODE, ConfigKey.CLUSTER_PORT), DEFAULT_CLUSTER_PORT) + getInt(properties, ConfigKey.merge(ConfigKey.NODE, ConfigKey.CLUSTER_PORT), DEFAULT_CLUSTER_PORT), + NodeStatus.HEALTHY ); } @@ -112,7 +114,8 @@ private ClusterInfo buildClusterInfo(Properties properties) { getString(properties, clusterNodeKey(index, ConfigKey.HOST)), getInt(properties, clusterNodeKey(index, ConfigKey.HTTP_PORT)), getInt(properties, clusterNodeKey(index, ConfigKey.TCP_PORT)), - getInt(properties, clusterNodeKey(index, ConfigKey.CLUSTER_PORT)) + getInt(properties, clusterNodeKey(index, ConfigKey.CLUSTER_PORT)), + NodeStatus.HEALTHY ); nodes.add(node); diff --git a/src/main/java/org/cache/network/http/CacheController.java b/src/main/java/org/cache/network/http/CacheController.java index 7033b2e..7a743c3 100644 --- a/src/main/java/org/cache/network/http/CacheController.java +++ b/src/main/java/org/cache/network/http/CacheController.java @@ -26,6 +26,8 @@ @RequestMapping("/cache") public class CacheController { + private static final String PING_DEFAULT_RESPONSE = "PONG"; + private final CacheOperations cacheService; private final KeyCodec keyCodec; @@ -84,6 +86,15 @@ public ResponseEntity size() { return ResponseEntity.ok(new SizeResponseDto(cacheService.size())); } + @GetMapping("/ping") + public ResponseEntity ping(@RequestParam(required = false) String value) { + if (value == null || value.isBlank()) { + return ResponseEntity.ok(new GetResponseDto(PING_DEFAULT_RESPONSE)); + } + + return ResponseEntity.ok(new GetResponseDto(value)); + } + @GetMapping("/metrics") public ResponseEntity metrics() { Snapshot snapshot = cacheService.metrics(); diff --git a/src/main/java/org/cache/network/tcp/connection/ClientConnectionHandler.java b/src/main/java/org/cache/network/tcp/connection/ClientConnectionHandler.java index 55b64ff..94aa951 100644 --- a/src/main/java/org/cache/network/tcp/connection/ClientConnectionHandler.java +++ b/src/main/java/org/cache/network/tcp/connection/ClientConnectionHandler.java @@ -1,13 +1,16 @@ package org.cache.network.tcp.connection; -import java.util.function.Function; - import java.io.IOException; import java.net.Socket; import java.util.List; +import java.util.function.Function; + +import static org.cache.protocol.handlers.ResponseConstants.ERROR; public class ClientConnectionHandler implements Runnable { + private static final String DEFAULT_ERROR_MESSAGE = "internal server error"; + private final Socket socket; private final Function, String> commandProcessor; @@ -23,12 +26,24 @@ public void run() { List command; while ((command = protocolConnection.readCommand()) != null) { - String response = commandProcessor.apply(command); - protocolConnection.write(response); + protocolConnection.write(process(command)); } } catch (IOException exception) { System.err.println("Client connection failed: " + exception.getMessage()); } } + + private String process(List command) { + try { + return commandProcessor.apply(command); + } catch (RuntimeException exception) { + String message = exception.getMessage(); + if (message == null || message.isBlank()) { + message = DEFAULT_ERROR_MESSAGE; + } + + return ERROR.name() + " " + message; + } + } } diff --git a/src/main/java/org/cache/protocol/CommandProcessor.java b/src/main/java/org/cache/protocol/CommandProcessor.java index dc0d53c..7d854e0 100644 --- a/src/main/java/org/cache/protocol/CommandProcessor.java +++ b/src/main/java/org/cache/protocol/CommandProcessor.java @@ -1,17 +1,18 @@ package org.cache.protocol; import org.cache.protocol.codec.KeyCodec; -import org.cache.protocol.handlers.CommandType; +import org.cache.core.CacheOperations; import org.cache.protocol.handlers.ClearHandler; import org.cache.protocol.handlers.CommandHandler; +import org.cache.protocol.handlers.CommandType; import org.cache.protocol.handlers.DeleteHandler; import org.cache.protocol.handlers.GetHandler; import org.cache.protocol.handlers.LrangeHandler; import org.cache.protocol.handlers.MetricsHandler; +import org.cache.protocol.handlers.PingHandler; import org.cache.protocol.handlers.PushHandler; import org.cache.protocol.handlers.PutHandler; import org.cache.protocol.handlers.SizeHandler; -import org.cache.core.CacheOperations; import java.util.EnumMap; import java.util.List; @@ -33,6 +34,7 @@ public CommandProcessor(KeyCodec keyCodec, CacheOperations cacheService) { handlers.put(CommandType.METRICS, new MetricsHandler(cacheService)); handlers.put(CommandType.PUSH, new PushHandler<>(keyCodec, cacheService)); handlers.put(CommandType.LRANGE, new LrangeHandler<>(keyCodec, cacheService)); + handlers.put(CommandType.PING, new PingHandler()); } public String process(List commandParts) { diff --git a/src/main/java/org/cache/protocol/handlers/CommandType.java b/src/main/java/org/cache/protocol/handlers/CommandType.java index 08aae94..c24434b 100644 --- a/src/main/java/org/cache/protocol/handlers/CommandType.java +++ b/src/main/java/org/cache/protocol/handlers/CommandType.java @@ -1,23 +1,14 @@ package org.cache.protocol.handlers; public enum CommandType { - PUT(true), - GET(true), - DELETE(true), - SIZE(false), - CLEAR(false), - METRICS(false), - PUSH(true), - LRANGE(true), - UNKNOWN(false); - - private final boolean isKeyCommand; - - CommandType(boolean isKeyCommand) { - this.isKeyCommand = isKeyCommand; - } - - public boolean isKeyCommand() { - return isKeyCommand; - } + PUT, + GET, + DELETE, + SIZE, + CLEAR, + METRICS, + PUSH, + LRANGE, + PING, + UNKNOWN; } diff --git a/src/main/java/org/cache/protocol/handlers/PingHandler.java b/src/main/java/org/cache/protocol/handlers/PingHandler.java new file mode 100644 index 0000000..9c28ffe --- /dev/null +++ b/src/main/java/org/cache/protocol/handlers/PingHandler.java @@ -0,0 +1,24 @@ +package org.cache.protocol.handlers; + +import java.util.List; + +public class PingHandler implements CommandHandler { + + private static final int COMMAND_PARTS_SIZE = 1; + private static final int VALUE_COMMAND_PARTS_SIZE = 2; + private static final int VALUE_INDEX = 1; + private static final String DEFAULT_RESPONSE = "PONG"; + + @Override + public String handle(List parts) { + if (parts.size() != COMMAND_PARTS_SIZE && parts.size() != VALUE_COMMAND_PARTS_SIZE) { + return TcpResponseSupport.error("usage: PING [value]"); + } + + if (parts.size() == VALUE_COMMAND_PARTS_SIZE) { + return parts.get(VALUE_INDEX); + } + + return DEFAULT_RESPONSE; + } +} diff --git a/src/test/java/org/cache/network/connection/ClientConnectionHandlerTest.java b/src/test/java/org/cache/network/connection/ClientConnectionHandlerTest.java index 494bd98..c308766 100644 --- a/src/test/java/org/cache/network/connection/ClientConnectionHandlerTest.java +++ b/src/test/java/org/cache/network/connection/ClientConnectionHandlerTest.java @@ -34,4 +34,24 @@ void runProcessesCommandAndWritesResponse() throws Exception { verify(socket).close(); assertEquals("VALUE apple\r\n", output.toString(UTF_8)); } + + @Test + void runWritesErrorAndKeepsConnectionOpenWhenCommandProcessingFails() throws Exception { + Socket socket = mock(Socket.class); + Function, String> commandProcessor = mock(Function.class); + var input = new ByteArrayInputStream("PUT fruit apple\r\nPING\r\n".getBytes(UTF_8)); + var output = new ByteArrayOutputStream(); + + when(socket.getInputStream()).thenReturn(input); + when(socket.getOutputStream()).thenReturn(output); + when(commandProcessor.apply(List.of("PUT", "fruit", "apple"))).thenThrow(new RuntimeException("replica unavailable")); + when(commandProcessor.apply(List.of("PING"))).thenReturn("PONG"); + + new ClientConnectionHandler(socket, commandProcessor).run(); + + verify(commandProcessor).apply(List.of("PUT", "fruit", "apple")); + verify(commandProcessor).apply(List.of("PING")); + verify(socket).close(); + assertEquals("ERROR replica unavailable\r\nPONG\r\n", output.toString(UTF_8)); + } } diff --git a/src/test/java/org/cache/protocol/CommandProcessorTest.java b/src/test/java/org/cache/protocol/CommandProcessorTest.java index 0ed655e..6d626ec 100644 --- a/src/test/java/org/cache/protocol/CommandProcessorTest.java +++ b/src/test/java/org/cache/protocol/CommandProcessorTest.java @@ -118,6 +118,16 @@ void processMetricsReturnsSnapshot() { } } + @Test + void processPingReturnsPongOrEchoValue() { + try (var cache = new LocalCache(10, new LruEvictionPolicy<>())) { + CommandProcessor processor = processor(cache); + + assertEquals("PONG", processor.process(List.of("PING"))); + assertEquals("hello", processor.process(List.of("PING", "hello"))); + } + } + @Test void processReturnsInvalidUsageErrors() { try (var cache = new LocalCache(10, new LruEvictionPolicy<>())) { diff --git a/src/test/java/org/cache/protocol/handlers/CommandHandlerTest.java b/src/test/java/org/cache/protocol/handlers/CommandHandlerTest.java index a8ba68a..0314939 100644 --- a/src/test/java/org/cache/protocol/handlers/CommandHandlerTest.java +++ b/src/test/java/org/cache/protocol/handlers/CommandHandlerTest.java @@ -25,6 +25,7 @@ class CommandHandlerTest { private GetHandler getHandler; private LrangeHandler lrangeHandler; private MetricsHandler metricsHandler; + private PingHandler pingHandler; private PushHandler pushHandler; private PutHandler putHandler; private SizeHandler sizeHandler; @@ -39,6 +40,7 @@ void setUp() { getHandler = new GetHandler<>(keyCodec, cacheService); lrangeHandler = new LrangeHandler<>(keyCodec, cacheService); metricsHandler = new MetricsHandler(cacheService); + pingHandler = new PingHandler(); pushHandler = new PushHandler<>(keyCodec, cacheService); putHandler = new PutHandler<>(keyCodec, cacheService); sizeHandler = new SizeHandler(cacheService); @@ -147,6 +149,13 @@ void metricsReturnsSnapshotAndRejectsInvalidUsage() { assertEquals("ERROR usage: METRICS", metricsHandler.handle(List.of("METRICS", "extra"))); } + @Test + void pingReturnsPongEchoValueAndRejectsInvalidUsage() { + assertEquals("PONG", pingHandler.handle(List.of("PING"))); + assertEquals("hello", pingHandler.handle(List.of("PING", "hello"))); + assertEquals("ERROR usage: PING [value]", pingHandler.handle(List.of("PING", "hello", "again"))); + } + private static ValueCodecRegistry valueCodecs() { return new ValueCodecRegistry() .register(ValueType.STRING, new StringValueCodec())