DVLP-011: Add ping command

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);
+    }
 }

src/main/java/org/cache/cluster/NodeStatus.java

@@ -0,0 +1,7 @@
+package org.cache.cluster;
+
+public enum NodeStatus {
+    HEALTHY,
+    SUSPECTED,
+    UNAVAILABLE
+}

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);

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<Object> cacheService;
     private final KeyCodec<Object> keyCodec;
 
@@ -84,6 +86,15 @@ public ResponseEntity<SizeResponseDto> size() {
         return ResponseEntity.ok(new SizeResponseDto(cacheService.size()));
     }
 
+    @GetMapping("/ping")
+    public ResponseEntity<GetResponseDto> 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<MetricsResponseDto> metrics() {
         Snapshot snapshot = cacheService.metrics();

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<List<String>, String> commandProcessor;
 
@@ -23,12 +26,24 @@ public void run() {
 
             List<String> 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<String> 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;
+        }
+    }
 }

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<K> keyCodec, CacheOperations<K> 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<String> commandParts) {

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;
 }

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<String> 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;
+    }
+}

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<List<String>, 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));
+    }
 }

src/test/java/org/cache/protocol/CommandProcessorTest.java

@@ -118,6 +118,16 @@ void processMetricsReturnsSnapshot() {
         }
     }
 
+    @Test
+    void processPingReturnsPongOrEchoValue() {
+        try (var cache = new LocalCache<String>(10, new LruEvictionPolicy<>())) {
+            CommandProcessor<String> 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<String>(10, new LruEvictionPolicy<>())) {

src/test/java/org/cache/protocol/handlers/CommandHandlerTest.java

@@ -25,6 +25,7 @@ class CommandHandlerTest {
     private GetHandler<String> getHandler;
     private LrangeHandler<String> lrangeHandler;
     private MetricsHandler metricsHandler;
+    private PingHandler pingHandler;
     private PushHandler<String> pushHandler;
     private PutHandler<String> 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())