diff --git a/src/main/java/org/cache/Main.java b/src/main/java/org/cache/Main.java index f40d83b..b3e6728 100644 --- a/src/main/java/org/cache/Main.java +++ b/src/main/java/org/cache/Main.java @@ -1,14 +1,18 @@ package org.cache; import org.cache.cluster.CacheNode; +import org.cache.cluster.routing.ClusterForwardingClient; +import org.cache.cluster.routing.RoutedCacheService; import org.cache.config.CacheConfig; import org.cache.config.CacheConfigLoader; import org.cache.core.Cache; +import org.cache.core.CacheOperations; import org.cache.core.CacheService; import org.cache.core.LocalCache; import org.cache.core.ValueType; import org.cache.eviction.EvictionPolicy; import org.cache.network.tcp.TcpCacheServer; +import org.cache.network.tcp.TcpCacheServerLifecycle; import org.cache.network.tcp.connection.ClientConnectionHandler; import org.cache.protocol.CommandProcessor; import org.cache.protocol.codec.KeyCodec; @@ -17,7 +21,9 @@ import org.cache.protocol.codec.ValueCodecRegistry; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Primary; import java.io.IOException; import java.net.ServerSocket; @@ -86,25 +92,78 @@ public CacheService cacheService(Cache cache, ValueCodecRegistry } @Bean - public CommandProcessor commandProcessor(KeyCodec keyCodec, CacheService cacheService) { + @Primary + public CacheOperations routedCacheService( + CacheService cacheService, + CacheNode cacheNode, + CacheConfig cacheConfig, + ClusterForwardingClient forwardingClient, + KeyCodec keyCodec + ) { + return new RoutedCacheService<>(cacheService, cacheNode, cacheConfig.clusterInfo(), forwardingClient, keyCodec); + } + + @Bean + public CommandProcessor commandProcessor(KeyCodec keyCodec, CacheOperations cacheService) { + return new CommandProcessor<>(keyCodec, cacheService); + } + + @Bean + public CommandProcessor clusterCommandProcessor( + KeyCodec keyCodec, + CacheService cacheService + ) { return new CommandProcessor<>(keyCodec, cacheService); } + @Bean + public ClusterForwardingClient clusterForwardingClient() { + return new ClusterForwardingClient(); + } + @Bean(destroyMethod = "shutdownNow") public ExecutorService tcpClientExecutor() { return Executors.newFixedThreadPool(SERVER_THREAD_COUNT); } + @Bean(destroyMethod = "shutdownNow") + public ExecutorService clusterClientExecutor() { + return Executors.newFixedThreadPool(SERVER_THREAD_COUNT); + } + @Bean - public TcpCacheServer tcpCacheServer( + public TcpCacheServer clientTcpCacheServer( CacheNode cacheNode, - ExecutorService tcpClientExecutor, - CommandProcessor commandProcessor + @Qualifier("tcpClientExecutor") ExecutorService tcpClientExecutor, + @Qualifier("commandProcessor") CommandProcessor commandProcessor ) throws IOException { return new TcpCacheServer( new ServerSocket(cacheNode.tcpPort()), tcpClientExecutor, - socket -> new ClientConnectionHandler(socket, commandProcessor) + socket -> new ClientConnectionHandler(socket, commandProcessor::process) + ); + } + + @Bean + public TcpCacheServer clusterTcpCacheServer( + CacheNode cacheNode, + @Qualifier("clusterClientExecutor") ExecutorService clusterClientExecutor, + @Qualifier("clusterCommandProcessor") CommandProcessor clusterCommandProcessor + ) throws IOException { + return new TcpCacheServer( + new ServerSocket(cacheNode.clusterPort()), + clusterClientExecutor, + socket -> new ClientConnectionHandler(socket, clusterCommandProcessor::process) ); } + + @Bean + public TcpCacheServerLifecycle clientTcpCacheServerLifecycle(@Qualifier("clientTcpCacheServer") TcpCacheServer server) { + return new TcpCacheServerLifecycle(server, "tcp-cache-server"); + } + + @Bean + public TcpCacheServerLifecycle clusterTcpCacheServerLifecycle(@Qualifier("clusterTcpCacheServer") TcpCacheServer server) { + return new TcpCacheServerLifecycle(server, "cluster-cache-server"); + } } diff --git a/src/main/java/org/cache/client/TcpCacheClient.java b/src/main/java/org/cache/client/TcpCacheClient.java index 141ac16..1e39646 100644 --- a/src/main/java/org/cache/client/TcpCacheClient.java +++ b/src/main/java/org/cache/client/TcpCacheClient.java @@ -2,54 +2,54 @@ import org.cache.client.serializer.Serializer; import org.cache.core.metrics.Snapshot; -import org.cache.network.tcp.connection.RespConnection; +import org.cache.network.tcp.connection.CacheResponseParser; +import org.cache.network.tcp.connection.RespCommandClient; import org.cache.protocol.codec.KeyCodec; import org.cache.protocol.handlers.ResponseConstants; import java.io.IOException; -import java.net.InetSocketAddress; -import java.net.Socket; import java.util.List; import java.util.Optional; -import static org.cache.protocol.handlers.ResponseConstants.ERROR; -import static org.cache.protocol.handlers.ResponseConstants.OK; +public class TcpCacheClient implements CacheClient { -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 static final int SINGLE_PART_RESPONSE_SIZE = 1; private static final int TWO_PART_RESPONSE_SIZE = 2; private static final int FIRST_VALUE_INDEX = 1; private static final int METRIC_NAME_VALUE_START_INDEX = 1; private static final int METRIC_NAME_VALUE_PAIR_SIZE = 2; private static final int DEFAULT_LRANGE_FROM_INDEX = 0; - private final RespConnection connection; + private final RespCommandClient commandClient; + private final String address; + private final int port; private final KeyCodec keyCodec; + private final CacheResponseParser responseParser; private final Serializer valueSerializer; - - public TcpCacheClient( + TcpCacheClient( + RespCommandClient commandClient, String address, int port, KeyCodec keyCodec, Serializer valueSerializer ) { - try { - this.connection = new RespConnection(connect(address, port)); - this.keyCodec = keyCodec; - this.valueSerializer = valueSerializer; - } catch (IOException exception) { - throw new CacheClientException("failed to connect to cache server", exception); - } + this(commandClient, address, port, keyCodec, valueSerializer, new CacheResponseParser()); } - TcpCacheClient(RespConnection connection, KeyCodec keyCodec, Serializer valueSerializer) { - this.connection = connection; + TcpCacheClient( + RespCommandClient commandClient, + String address, + int port, + KeyCodec keyCodec, + Serializer valueSerializer, + CacheResponseParser responseParser + ) { + this.commandClient = commandClient; + this.address = address; + this.port = port; this.keyCodec = keyCodec; this.valueSerializer = valueSerializer; + this.responseParser = responseParser; } @Override @@ -66,15 +66,15 @@ public void put(K key, V value) { public Optional get(K key) { List response = send(List.of("GET", keyCodec.encode(key))); - if (isResponse(response, ResponseConstants.NOT_FOUND.name(), SINGLE_PART_RESPONSE_SIZE)) { + if (responseParser.isNotFound(response)) { return Optional.empty(); } - if (isResponse(response, ResponseConstants.VALUE.name(), TWO_PART_RESPONSE_SIZE)) { - return Optional.of(valueSerializer.decode(response.get(FIRST_VALUE_INDEX))); - } - - throw new CacheClientException("unexpected cache server response: " + response); + return responseParser.value(response) + .map(valueSerializer::decode) + .or(() -> { + throw unexpectedResponse(response); + }); } @Override @@ -122,8 +122,8 @@ public Snapshot metrics() { public int size() { List response = send(List.of("SIZE")); - if (!isResponse(response, ResponseConstants.SIZE.name(), TWO_PART_RESPONSE_SIZE)) { - throw new CacheClientException("unexpected cache server response: " + response); + if (!responseParser.isResponse(response, ResponseConstants.SIZE, TWO_PART_RESPONSE_SIZE)) { + throw unexpectedResponse(response); } return Integer.parseInt(response.get(FIRST_VALUE_INDEX)); @@ -153,7 +153,7 @@ public List lrange(K key, int from, int to) { Integer.toString(to) )); - if (isResponse(response, ResponseConstants.NOT_FOUND.name(), SINGLE_PART_RESPONSE_SIZE)) { + if (responseParser.isNotFound(response)) { return List.of(); } @@ -162,29 +162,20 @@ public List lrange(K key, int from, int to) { .toList(); } - @Override - public void close() { - try { - connection.close(); - } catch (IOException exception) { - throw new CacheClientException("failed to close cache client", exception); - } - } - private void expectOk(List command) { List response = send(command); - if (!isResponse(response, OK.name(), SINGLE_PART_RESPONSE_SIZE)) { - throw new CacheClientException("unexpected cache server response: " + response); + if (!responseParser.isOk(response)) { + throw unexpectedResponse(response); } } private List send(List command) { try { - List response = connection.sendCommandForResponse(command); + List response = commandClient.send(address, port, command); - if (!response.isEmpty() && response.getFirst().equals(ERROR.name())) { - throw new CacheClientException("unexpected cache server response: " + response); + if (responseParser.isError(response)) { + throw unexpectedResponse(response); } return response; @@ -193,22 +184,8 @@ private List send(List command) { } } - private boolean isResponse(List response, String type, int size) { - return response.size() == size && response.getFirst().equals(type); + private CacheClientException unexpectedResponse(List response) { + return new CacheClientException("unexpected cache server response: " + response); } - private Socket connect( - String address, - int port - ) throws IOException { - Socket socket = new Socket(); - try { - socket.connect(new InetSocketAddress(address, port), DEFAULT_CONNECT_TIMEOUT_MILLIS); - socket.setSoTimeout(DEFAULT_READ_TIMEOUT_MILLIS); - return socket; - } catch (IOException exception) { - socket.close(); - throw exception; - } - } } diff --git a/src/main/java/org/cache/cluster/routing/ClusterForwardingClient.java b/src/main/java/org/cache/cluster/routing/ClusterForwardingClient.java new file mode 100644 index 0000000..529882d --- /dev/null +++ b/src/main/java/org/cache/cluster/routing/ClusterForwardingClient.java @@ -0,0 +1,28 @@ +package org.cache.cluster.routing; + +import org.cache.cluster.CacheNode; +import org.cache.network.tcp.connection.RespCommandClient; + +import java.io.IOException; +import java.util.List; + +public class ClusterForwardingClient { + + private final RespCommandClient commandClient; + + public ClusterForwardingClient() { + this(new RespCommandClient()); + } + + ClusterForwardingClient(RespCommandClient commandClient) { + this.commandClient = commandClient; + } + + public List forward(CacheNode targetNode, List commandParts) { + try { + return commandClient.send(targetNode.host(), targetNode.clusterPort(), commandParts); + } catch (IOException exception) { + throw new ClusterForwardingException("Failed to forward request to node: " + targetNode.id(), exception); + } + } +} diff --git a/src/main/java/org/cache/cluster/routing/ClusterForwardingException.java b/src/main/java/org/cache/cluster/routing/ClusterForwardingException.java new file mode 100644 index 0000000..b4d97b9 --- /dev/null +++ b/src/main/java/org/cache/cluster/routing/ClusterForwardingException.java @@ -0,0 +1,12 @@ +package org.cache.cluster.routing; + +public class ClusterForwardingException extends RuntimeException { + + public ClusterForwardingException(String message) { + super(message); + } + + public ClusterForwardingException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/src/main/java/org/cache/cluster/routing/RoutedCacheService.java b/src/main/java/org/cache/cluster/routing/RoutedCacheService.java new file mode 100644 index 0000000..ce598e6 --- /dev/null +++ b/src/main/java/org/cache/cluster/routing/RoutedCacheService.java @@ -0,0 +1,164 @@ +package org.cache.cluster.routing; + +import org.cache.cluster.CacheNode; +import org.cache.cluster.ClusterInfo; +import org.cache.cluster.hashing.ConsistentHashRing; +import org.cache.core.CacheOperations; +import org.cache.core.metrics.Snapshot; +import org.cache.network.tcp.connection.CacheResponseParser; +import org.cache.protocol.codec.KeyCodec; + +import java.util.List; +import java.util.Optional; + +public class RoutedCacheService implements CacheOperations { + + private final CacheOperations localService; + private final CacheNode currentNode; + private final ClusterForwardingClient forwardingClient; + private final ConsistentHashRing hashRing; + private final KeyCodec keyCodec; + private final CacheResponseParser responseParser; + + public RoutedCacheService( + CacheOperations localService, + CacheNode currentNode, + ClusterInfo clusterInfo, + ClusterForwardingClient forwardingClient, + KeyCodec keyCodec + ) { + this(localService, currentNode, clusterInfo, forwardingClient, keyCodec, new CacheResponseParser()); + } + + RoutedCacheService( + CacheOperations localService, + CacheNode currentNode, + ClusterInfo clusterInfo, + ClusterForwardingClient forwardingClient, + KeyCodec keyCodec, + CacheResponseParser responseParser + ) { + this.localService = localService; + this.currentNode = currentNode; + this.forwardingClient = forwardingClient; + this.hashRing = clusterInfo == null ? null : new ConsistentHashRing(clusterInfo); + this.keyCodec = keyCodec; + this.responseParser = responseParser; + } + + @Override + public void putString(K key, String value, long ttlMillis) { + CacheNode owner = ownerFor(key); + if (isLocal(owner)) { + localService.putString(key, value, ttlMillis); + return; + } + + expectOk(forwardingClient.forward(owner, List.of( + "PUT", + keyCodec.encode(key), + value, + Long.toString(ttlMillis) + ))); + } + + @Override + public Optional getString(K key) { + CacheNode owner = ownerFor(key); + if (isLocal(owner)) { + return localService.getString(key); + } + + List response = forwardingClient.forward(owner, List.of("GET", keyCodec.encode(key))); + if (responseParser.isNotFound(response)) { + return Optional.empty(); + } + + return responseParser.value(response) + .or(() -> { + throw new ClusterForwardingException("Unexpected cluster response: " + response); + }); + } + + @Override + public void push(K key, String value) { + CacheNode owner = ownerFor(key); + if (isLocal(owner)) { + localService.push(key, value); + return; + } + + expectOk(forwardingClient.forward(owner, List.of("PUSH", keyCodec.encode(key), value))); + } + + @Override + public Optional> lrange(K key, int from, int to) { + CacheNode owner = ownerFor(key); + if (isLocal(owner)) { + return localService.lrange(key, from, to); + } + + List response = forwardingClient.forward(owner, List.of( + "LRANGE", + keyCodec.encode(key), + Integer.toString(from), + Integer.toString(to) + )); + + if (responseParser.isNotFound(response)) { + return Optional.empty(); + } + + if (response.isEmpty()) { + return Optional.of(List.of()); + } + + if (!responseParser.isKnownResponse(response)) { + return Optional.of(List.copyOf(response)); + } + + throw new ClusterForwardingException("Unexpected cluster response: " + response); + } + + @Override + public void delete(K key) { + CacheNode owner = ownerFor(key); + if (isLocal(owner)) { + localService.delete(key); + return; + } + + expectOk(forwardingClient.forward(owner, List.of("DELETE", keyCodec.encode(key)))); + } + + @Override + public int size() { + return localService.size(); + } + + @Override + public void clear() { + localService.clear(); + } + + @Override + public Snapshot metrics() { + return localService.metrics(); + } + + private CacheNode ownerFor(K key) { + return hashRing == null ? currentNode : hashRing.nodeFor(keyCodec.encode(key)); + } + + private boolean isLocal(CacheNode owner) { + return owner.id().equals(currentNode.id()); + } + + private void expectOk(List response) { + if (responseParser.isOk(response)) { + return; + } + + throw new ClusterForwardingException("Unexpected cluster response: " + response); + } +} diff --git a/src/main/java/org/cache/config/CacheConfigLoader.java b/src/main/java/org/cache/config/CacheConfigLoader.java index da2f0ce..dba4a1d 100644 --- a/src/main/java/org/cache/config/CacheConfigLoader.java +++ b/src/main/java/org/cache/config/CacheConfigLoader.java @@ -23,6 +23,7 @@ public final class CacheConfigLoader { private static final String DEFAULT_CONFIG_FILE = "config.yml"; + private static final String CONFIG_FILE_PROPERTY = "cache.config"; private static final int DEFAULT_CAPACITY = 1_000; private static final long DEFAULT_TTL_MILLIS = 0L; private static final String DEFAULT_KEY_TYPE = "string"; @@ -37,7 +38,7 @@ public final class CacheConfigLoader { private final String configFile; public CacheConfigLoader() { - this(DEFAULT_CONFIG_FILE); + this(System.getProperty(CONFIG_FILE_PROPERTY, DEFAULT_CONFIG_FILE)); } public CacheConfigLoader(String configFile) { diff --git a/src/main/java/org/cache/core/CacheOperations.java b/src/main/java/org/cache/core/CacheOperations.java new file mode 100644 index 0000000..75bce73 --- /dev/null +++ b/src/main/java/org/cache/core/CacheOperations.java @@ -0,0 +1,25 @@ +package org.cache.core; + +import org.cache.core.metrics.Snapshot; + +import java.util.List; +import java.util.Optional; + +public interface CacheOperations { + + void putString(K key, String value, long ttlMillis); + + Optional getString(K key); + + void push(K key, String value); + + Optional> lrange(K key, int from, int to); + + void delete(K key); + + int size(); + + void clear(); + + Snapshot metrics(); +} diff --git a/src/main/java/org/cache/core/CacheService.java b/src/main/java/org/cache/core/CacheService.java index 5b31495..7e1a601 100644 --- a/src/main/java/org/cache/core/CacheService.java +++ b/src/main/java/org/cache/core/CacheService.java @@ -9,7 +9,7 @@ import java.util.List; import java.util.Optional; -public class CacheService { +public class CacheService implements CacheOperations { private final Cache cache; private final ValueCodecRegistry valueCodecs; diff --git a/src/main/java/org/cache/network/http/CacheController.java b/src/main/java/org/cache/network/http/CacheController.java index be1b519..7033b2e 100644 --- a/src/main/java/org/cache/network/http/CacheController.java +++ b/src/main/java/org/cache/network/http/CacheController.java @@ -1,6 +1,7 @@ package org.cache.network.http; -import org.cache.core.CacheService; +import org.cache.cluster.routing.ClusterForwardingException; +import org.cache.core.CacheOperations; import org.cache.core.metrics.Snapshot; import org.cache.network.http.dto.GetResponseDto; import org.cache.network.http.dto.ListResponseDto; @@ -17,18 +18,18 @@ import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.PutMapping; import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; -import org.springframework.web.bind.annotation.RequestMapping; @RestController @RequestMapping("/cache") public class CacheController { - private final CacheService cacheService; + private final CacheOperations cacheService; private final KeyCodec keyCodec; - public CacheController(CacheService cacheService, KeyCodec keyCodec) { + public CacheController(CacheOperations cacheService, KeyCodec keyCodec) { this.cacheService = cacheService; this.keyCodec = keyCodec; } @@ -40,6 +41,8 @@ public ResponseEntity put(@PathVariable String key, @RequestBody ValueRequ return ResponseEntity.noContent().build(); } catch (IllegalArgumentException exception) { return ResponseEntity.badRequest().build(); + } catch (ClusterForwardingException exception) { + return ResponseEntity.status(HttpStatus.BAD_GATEWAY).build(); } } @@ -53,6 +56,8 @@ public ResponseEntity get(@PathVariable String key) { return ResponseEntity.badRequest().build(); } catch (WrongValueTypeException exception) { return ResponseEntity.status(HttpStatus.CONFLICT).build(); + } catch (ClusterForwardingException exception) { + return ResponseEntity.status(HttpStatus.BAD_GATEWAY).build(); } } @@ -63,6 +68,8 @@ public ResponseEntity delete(@PathVariable String key) { return ResponseEntity.noContent().build(); } catch (IllegalArgumentException exception) { return ResponseEntity.badRequest().build(); + } catch (ClusterForwardingException exception) { + return ResponseEntity.status(HttpStatus.BAD_GATEWAY).build(); } } @@ -98,6 +105,8 @@ public ResponseEntity push(@PathVariable String key, @RequestBody ValueReq return ResponseEntity.badRequest().build(); } catch (WrongValueTypeException exception) { return ResponseEntity.status(HttpStatus.CONFLICT).build(); + } catch (ClusterForwardingException exception) { + return ResponseEntity.status(HttpStatus.BAD_GATEWAY).build(); } } @@ -115,6 +124,8 @@ public ResponseEntity lrange( return ResponseEntity.badRequest().build(); } catch (WrongValueTypeException exception) { return ResponseEntity.status(HttpStatus.CONFLICT).build(); + } catch (ClusterForwardingException exception) { + return ResponseEntity.status(HttpStatus.BAD_GATEWAY).build(); } } diff --git a/src/main/java/org/cache/network/tcp/TcpCacheServerLifecycle.java b/src/main/java/org/cache/network/tcp/TcpCacheServerLifecycle.java index d76e7b2..ca70903 100644 --- a/src/main/java/org/cache/network/tcp/TcpCacheServerLifecycle.java +++ b/src/main/java/org/cache/network/tcp/TcpCacheServerLifecycle.java @@ -1,19 +1,23 @@ package org.cache.network.tcp; import org.springframework.context.SmartLifecycle; -import org.springframework.stereotype.Component; import java.io.IOException; -@Component public class TcpCacheServerLifecycle implements SmartLifecycle { private final TcpCacheServer server; + private final String threadName; private volatile boolean running; private Thread serverThread; public TcpCacheServerLifecycle(TcpCacheServer server) { + this(server, "tcp-cache-server"); + } + + public TcpCacheServerLifecycle(TcpCacheServer server, String threadName) { this.server = server; + this.threadName = threadName; } @Override @@ -23,7 +27,7 @@ public void start() { } running = true; - serverThread = new Thread(this::runServer, "tcp-cache-server"); + serverThread = new Thread(this::runServer, threadName); serverThread.start(); } diff --git a/src/main/java/org/cache/network/tcp/connection/CacheResponseParser.java b/src/main/java/org/cache/network/tcp/connection/CacheResponseParser.java new file mode 100644 index 0000000..e4b3f57 --- /dev/null +++ b/src/main/java/org/cache/network/tcp/connection/CacheResponseParser.java @@ -0,0 +1,64 @@ +package org.cache.network.tcp.connection; + +import org.cache.protocol.handlers.ResponseConstants; + +import java.util.List; +import java.util.Optional; + +public class CacheResponseParser { + + private static final int RESPONSE_TYPE_INDEX = 0; + private static final int RESPONSE_VALUE_INDEX = 1; + private static final int SINGLE_PART_RESPONSE_SIZE = 1; + private static final int TWO_PART_RESPONSE_SIZE = 2; + + public boolean isOk(List response) { + return isResponse(response, ResponseConstants.OK, SINGLE_PART_RESPONSE_SIZE); + } + + public boolean isError(List response) { + return isResponse(response, ResponseConstants.ERROR); + } + + public boolean isNotFound(List response) { + return isResponse(response, ResponseConstants.NOT_FOUND, SINGLE_PART_RESPONSE_SIZE); + } + + public Optional value(List response) { + if (isResponse(response, ResponseConstants.VALUE, TWO_PART_RESPONSE_SIZE)) { + return Optional.of(response.get(RESPONSE_VALUE_INDEX)); + } + + return Optional.empty(); + } + + public Optional errorMessage(List response) { + if (isError(response) && response.size() > RESPONSE_VALUE_INDEX) { + return Optional.of(response.get(RESPONSE_VALUE_INDEX)); + } + + return Optional.empty(); + } + + public boolean isKnownResponse(List response) { + return !response.isEmpty() && isKnownResponse(response.getFirst()); + } + + public boolean isKnownResponse(String value) { + for (ResponseConstants responseType : ResponseConstants.values()) { + if (responseType.name().equals(value)) { + return true; + } + } + + return false; + } + + public boolean isResponse(List response, ResponseConstants responseType) { + return !response.isEmpty() && response.get(RESPONSE_TYPE_INDEX).equals(responseType.name()); + } + + public boolean isResponse(List response, ResponseConstants responseType, int size) { + return response.size() == size && response.get(RESPONSE_TYPE_INDEX).equals(responseType.name()); + } +} 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 a2e0ec0..55b64ff 100644 --- a/src/main/java/org/cache/network/tcp/connection/ClientConnectionHandler.java +++ b/src/main/java/org/cache/network/tcp/connection/ClientConnectionHandler.java @@ -1,6 +1,6 @@ package org.cache.network.tcp.connection; -import org.cache.protocol.CommandProcessor; +import java.util.function.Function; import java.io.IOException; import java.net.Socket; @@ -9,9 +9,9 @@ public class ClientConnectionHandler implements Runnable { private final Socket socket; - private final CommandProcessor commandProcessor; + private final Function, String> commandProcessor; - public ClientConnectionHandler(Socket socket, CommandProcessor commandProcessor) { + public ClientConnectionHandler(Socket socket, Function, String> commandProcessor) { this.socket = socket; this.commandProcessor = commandProcessor; } @@ -23,7 +23,7 @@ public void run() { List command; while ((command = protocolConnection.readCommand()) != null) { - String response = commandProcessor.process(command); + String response = commandProcessor.apply(command); protocolConnection.write(response); } diff --git a/src/main/java/org/cache/network/tcp/connection/RespCommandClient.java b/src/main/java/org/cache/network/tcp/connection/RespCommandClient.java new file mode 100644 index 0000000..3eabf9b --- /dev/null +++ b/src/main/java/org/cache/network/tcp/connection/RespCommandClient.java @@ -0,0 +1,31 @@ +package org.cache.network.tcp.connection; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.util.List; + +public class RespCommandClient { + + private static final int DEFAULT_CONNECT_TIMEOUT_MILLIS = 2_000; + private static final int DEFAULT_READ_TIMEOUT_MILLIS = 5_000; + + public List send(String host, int port, List commandParts) throws IOException { + try (Socket socket = connect(host, port); + RespConnection connection = new RespConnection(socket)) { + return connection.sendCommandForResponse(commandParts); + } + } + + private Socket connect(String host, int port) throws IOException { + Socket socket = new Socket(); + try { + socket.connect(new InetSocketAddress(host, port), DEFAULT_CONNECT_TIMEOUT_MILLIS); + socket.setSoTimeout(DEFAULT_READ_TIMEOUT_MILLIS); + return socket; + } catch (IOException exception) { + socket.close(); + throw exception; + } + } +} diff --git a/src/main/java/org/cache/protocol/CommandProcessor.java b/src/main/java/org/cache/protocol/CommandProcessor.java index 759c2c8..dc0d53c 100644 --- a/src/main/java/org/cache/protocol/CommandProcessor.java +++ b/src/main/java/org/cache/protocol/CommandProcessor.java @@ -11,7 +11,7 @@ import org.cache.protocol.handlers.PushHandler; import org.cache.protocol.handlers.PutHandler; import org.cache.protocol.handlers.SizeHandler; -import org.cache.core.CacheService; +import org.cache.core.CacheOperations; import java.util.EnumMap; import java.util.List; @@ -23,7 +23,7 @@ public class CommandProcessor { private final Map handlers; - public CommandProcessor(KeyCodec keyCodec, CacheService cacheService) { + public CommandProcessor(KeyCodec keyCodec, CacheOperations cacheService) { this.handlers = new EnumMap<>(CommandType.class); handlers.put(CommandType.PUT, new PutHandler<>(keyCodec, cacheService)); handlers.put(CommandType.GET, new GetHandler<>(keyCodec, cacheService)); diff --git a/src/main/java/org/cache/protocol/handlers/ClearHandler.java b/src/main/java/org/cache/protocol/handlers/ClearHandler.java index a27aaf2..f58a83e 100644 --- a/src/main/java/org/cache/protocol/handlers/ClearHandler.java +++ b/src/main/java/org/cache/protocol/handlers/ClearHandler.java @@ -1,6 +1,6 @@ package org.cache.protocol.handlers; -import org.cache.core.CacheService; +import org.cache.core.CacheOperations; import java.util.List; @@ -8,9 +8,9 @@ public class ClearHandler implements CommandHandler { private static final int COMMAND_PARTS = 1; - private final CacheService cacheService; + private final CacheOperations cacheService; - public ClearHandler(CacheService cacheService) { + public ClearHandler(CacheOperations cacheService) { this.cacheService = cacheService; } diff --git a/src/main/java/org/cache/protocol/handlers/CommandType.java b/src/main/java/org/cache/protocol/handlers/CommandType.java index 09f06cc..08aae94 100644 --- a/src/main/java/org/cache/protocol/handlers/CommandType.java +++ b/src/main/java/org/cache/protocol/handlers/CommandType.java @@ -1,13 +1,23 @@ package org.cache.protocol.handlers; public enum CommandType { - PUT, - GET, - DELETE, - SIZE, - CLEAR, - METRICS, - PUSH, - LRANGE, - UNKNOWN + 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; + } } diff --git a/src/main/java/org/cache/protocol/handlers/DeleteHandler.java b/src/main/java/org/cache/protocol/handlers/DeleteHandler.java index bb6a7db..46e32f7 100644 --- a/src/main/java/org/cache/protocol/handlers/DeleteHandler.java +++ b/src/main/java/org/cache/protocol/handlers/DeleteHandler.java @@ -1,7 +1,7 @@ package org.cache.protocol.handlers; import org.cache.protocol.codec.KeyCodec; -import org.cache.core.CacheService; +import org.cache.core.CacheOperations; import org.cache.protocol.codec.KeyCodecException; import java.util.List; @@ -12,9 +12,9 @@ public class DeleteHandler implements CommandHandler { private static final int KEY_INDEX = 1; private final KeyCodec keyCodec; - private final CacheService cacheService; + private final CacheOperations cacheService; - public DeleteHandler(KeyCodec keyCodec, CacheService cacheService) { + public DeleteHandler(KeyCodec keyCodec, CacheOperations cacheService) { this.keyCodec = keyCodec; this.cacheService = cacheService; } diff --git a/src/main/java/org/cache/protocol/handlers/GetHandler.java b/src/main/java/org/cache/protocol/handlers/GetHandler.java index ce5393c..960300d 100644 --- a/src/main/java/org/cache/protocol/handlers/GetHandler.java +++ b/src/main/java/org/cache/protocol/handlers/GetHandler.java @@ -1,7 +1,7 @@ package org.cache.protocol.handlers; import org.cache.protocol.codec.KeyCodec; -import org.cache.core.CacheService; +import org.cache.core.CacheOperations; import org.cache.protocol.codec.KeyCodecException; import java.util.List; @@ -12,9 +12,9 @@ public class GetHandler implements CommandHandler { private static final int KEY_INDEX = 1; private final KeyCodec keyCodec; - private final CacheService cacheService; + private final CacheOperations cacheService; - public GetHandler(KeyCodec keyCodec, CacheService cacheService) { + public GetHandler(KeyCodec keyCodec, CacheOperations cacheService) { this.keyCodec = keyCodec; this.cacheService = cacheService; } diff --git a/src/main/java/org/cache/protocol/handlers/LrangeHandler.java b/src/main/java/org/cache/protocol/handlers/LrangeHandler.java index ffe46a3..17d4f2a 100644 --- a/src/main/java/org/cache/protocol/handlers/LrangeHandler.java +++ b/src/main/java/org/cache/protocol/handlers/LrangeHandler.java @@ -1,7 +1,7 @@ package org.cache.protocol.handlers; import org.cache.protocol.codec.KeyCodec; -import org.cache.core.CacheService; +import org.cache.core.CacheOperations; import org.cache.protocol.codec.KeyCodecException; import java.util.List; @@ -16,9 +16,9 @@ public class LrangeHandler implements CommandHandler { private static final int DEFAULT_FROM_INDEX = 0; private final KeyCodec keyCodec; - private final CacheService cacheService; + private final CacheOperations cacheService; - public LrangeHandler(KeyCodec keyCodec, CacheService cacheService) { + public LrangeHandler(KeyCodec keyCodec, CacheOperations cacheService) { this.keyCodec = keyCodec; this.cacheService = cacheService; } diff --git a/src/main/java/org/cache/protocol/handlers/MetricsHandler.java b/src/main/java/org/cache/protocol/handlers/MetricsHandler.java index 7816b35..6d917e9 100644 --- a/src/main/java/org/cache/protocol/handlers/MetricsHandler.java +++ b/src/main/java/org/cache/protocol/handlers/MetricsHandler.java @@ -1,7 +1,7 @@ package org.cache.protocol.handlers; import org.cache.core.metrics.Snapshot; -import org.cache.core.CacheService; +import org.cache.core.CacheOperations; import java.util.List; @@ -9,9 +9,9 @@ public class MetricsHandler implements CommandHandler { private static final int COMMAND_PARTS = 1; - private final CacheService cacheService; + private final CacheOperations cacheService; - public MetricsHandler(CacheService cacheService) { + public MetricsHandler(CacheOperations cacheService) { this.cacheService = cacheService; } diff --git a/src/main/java/org/cache/protocol/handlers/PushHandler.java b/src/main/java/org/cache/protocol/handlers/PushHandler.java index 45b7c3c..cb7066d 100644 --- a/src/main/java/org/cache/protocol/handlers/PushHandler.java +++ b/src/main/java/org/cache/protocol/handlers/PushHandler.java @@ -1,7 +1,7 @@ package org.cache.protocol.handlers; import org.cache.protocol.codec.KeyCodec; -import org.cache.core.CacheService; +import org.cache.core.CacheOperations; import org.cache.protocol.codec.KeyCodecException; import java.util.List; @@ -13,9 +13,9 @@ public class PushHandler implements CommandHandler { private static final int VALUE_INDEX = 2; private final KeyCodec keyCodec; - private final CacheService cacheService; + private final CacheOperations cacheService; - public PushHandler(KeyCodec keyCodec, CacheService cacheService) { + public PushHandler(KeyCodec keyCodec, CacheOperations cacheService) { this.keyCodec = keyCodec; this.cacheService = cacheService; } diff --git a/src/main/java/org/cache/protocol/handlers/PutHandler.java b/src/main/java/org/cache/protocol/handlers/PutHandler.java index 9db5ec9..808e0ba 100644 --- a/src/main/java/org/cache/protocol/handlers/PutHandler.java +++ b/src/main/java/org/cache/protocol/handlers/PutHandler.java @@ -1,7 +1,7 @@ package org.cache.protocol.handlers; import org.cache.protocol.codec.KeyCodec; -import org.cache.core.CacheService; +import org.cache.core.CacheOperations; import org.cache.protocol.codec.KeyCodecException; import java.util.List; @@ -15,9 +15,9 @@ public class PutHandler implements CommandHandler { private static final int TTL_INDEX = 3; private final KeyCodec keyCodec; - private final CacheService cacheService; + private final CacheOperations cacheService; - public PutHandler(KeyCodec keyCodec, CacheService cacheService) { + public PutHandler(KeyCodec keyCodec, CacheOperations cacheService) { this.keyCodec = keyCodec; this.cacheService = cacheService; } diff --git a/src/main/java/org/cache/protocol/handlers/SizeHandler.java b/src/main/java/org/cache/protocol/handlers/SizeHandler.java index 890dbeb..a4e1bd4 100644 --- a/src/main/java/org/cache/protocol/handlers/SizeHandler.java +++ b/src/main/java/org/cache/protocol/handlers/SizeHandler.java @@ -1,6 +1,6 @@ package org.cache.protocol.handlers; -import org.cache.core.CacheService; +import org.cache.core.CacheOperations; import java.util.List; @@ -8,9 +8,9 @@ public class SizeHandler implements CommandHandler { private static final int COMMAND_PARTS = 1; - private final CacheService cacheService; + private final CacheOperations cacheService; - public SizeHandler(CacheService cacheService) { + public SizeHandler(CacheOperations cacheService) { this.cacheService = cacheService; } diff --git a/src/test/java/org/cache/client/TcpCacheClientTest.java b/src/test/java/org/cache/client/TcpCacheClientTest.java index f5a3e3e..92b31aa 100644 --- a/src/test/java/org/cache/client/TcpCacheClientTest.java +++ b/src/test/java/org/cache/client/TcpCacheClientTest.java @@ -2,7 +2,7 @@ import org.cache.client.serializer.StringSerializer; import org.cache.core.metrics.Snapshot; -import org.cache.network.tcp.connection.RespConnection; +import org.cache.network.tcp.connection.RespCommandClient; import org.cache.protocol.codec.StringKeyCodec; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -19,36 +19,36 @@ class TcpCacheClientTest { - private RespConnection connection; + private RespCommandClient commandClient; private TcpCacheClient client; @BeforeEach void setUp() { - connection = mock(RespConnection.class); - client = new TcpCacheClient<>(connection, new StringKeyCodec(), new StringSerializer()); + commandClient = mock(RespCommandClient.class); + client = new TcpCacheClient<>(commandClient, "localhost", 2020, new StringKeyCodec(), new StringSerializer()); } @Test void putSendsPutCommand() throws Exception { - when(connection.sendCommandForResponse(List.of("PUT", "fruit", "apple"))).thenReturn(List.of("OK")); + when(commandClient.send("localhost", 2020, List.of("PUT", "fruit", "apple"))).thenReturn(List.of("OK")); client.put("fruit", "apple"); - verify(connection).sendCommandForResponse(List.of("PUT", "fruit", "apple")); + verify(commandClient).send("localhost", 2020, List.of("PUT", "fruit", "apple")); } @Test void putWithTtlSendsPutCommandWithTtl() throws Exception { - when(connection.sendCommandForResponse(List.of("PUT", "fruit", "apple", "1000"))).thenReturn(List.of("OK")); + when(commandClient.send("localhost", 2020, List.of("PUT", "fruit", "apple", "1000"))).thenReturn(List.of("OK")); client.put("fruit", "apple", 1_000); - verify(connection).sendCommandForResponse(List.of("PUT", "fruit", "apple", "1000")); + verify(commandClient).send("localhost", 2020, List.of("PUT", "fruit", "apple", "1000")); } @Test void getReturnsValueWhenServerReturnsValue() throws Exception { - when(connection.sendCommandForResponse(List.of("GET", "fruit"))).thenReturn(List.of("VALUE", "apple")); + when(commandClient.send("localhost", 2020, List.of("GET", "fruit"))).thenReturn(List.of("VALUE", "apple")); Optional value = client.get("fruit"); @@ -57,7 +57,7 @@ void getReturnsValueWhenServerReturnsValue() throws Exception { @Test void getReturnsEmptyWhenServerReturnsNotFound() throws Exception { - when(connection.sendCommandForResponse(List.of("GET", "missing"))).thenReturn(List.of("NOT_FOUND")); + when(commandClient.send("localhost", 2020, List.of("GET", "missing"))).thenReturn(List.of("NOT_FOUND")); Optional value = client.get("missing"); @@ -66,16 +66,16 @@ void getReturnsEmptyWhenServerReturnsNotFound() throws Exception { @Test void deleteSendsDeleteCommand() throws Exception { - when(connection.sendCommandForResponse(List.of("DELETE", "fruit"))).thenReturn(List.of("OK")); + when(commandClient.send("localhost", 2020, List.of("DELETE", "fruit"))).thenReturn(List.of("OK")); client.delete("fruit"); - verify(connection).sendCommandForResponse(List.of("DELETE", "fruit")); + verify(commandClient).send("localhost", 2020, List.of("DELETE", "fruit")); } @Test void sizeReturnsParsedSize() throws Exception { - when(connection.sendCommandForResponse(List.of("SIZE"))).thenReturn(List.of("SIZE", "3")); + when(commandClient.send("localhost", 2020, List.of("SIZE"))).thenReturn(List.of("SIZE", "3")); int size = client.size(); @@ -84,25 +84,25 @@ void sizeReturnsParsedSize() throws Exception { @Test void clearSendsClearCommand() throws Exception { - when(connection.sendCommandForResponse(List.of("CLEAR"))).thenReturn(List.of("OK")); + when(commandClient.send("localhost", 2020, List.of("CLEAR"))).thenReturn(List.of("OK")); client.clear(); - verify(connection).sendCommandForResponse(List.of("CLEAR")); + verify(commandClient).send("localhost", 2020, List.of("CLEAR")); } @Test void pushSendsPushCommand() throws Exception { - when(connection.sendCommandForResponse(List.of("PUSH", "fruits", "apple"))).thenReturn(List.of("OK")); + when(commandClient.send("localhost", 2020, List.of("PUSH", "fruits", "apple"))).thenReturn(List.of("OK")); client.push("fruits", "apple"); - verify(connection).sendCommandForResponse(List.of("PUSH", "fruits", "apple")); + verify(commandClient).send("localhost", 2020, List.of("PUSH", "fruits", "apple")); } @Test void lrangeSendsRangeCommandAndReturnsValues() throws Exception { - when(connection.sendCommandForResponse(List.of("LRANGE", "fruits", "0", "2"))) + when(commandClient.send("localhost", 2020, List.of("LRANGE", "fruits", "0", "2"))) .thenReturn(List.of("apple", "banana")); List values = client.lrange("fruits", 0, 2); @@ -112,7 +112,7 @@ void lrangeSendsRangeCommandAndReturnsValues() throws Exception { @Test void lrangeWithToUsesZeroAsFromIndex() throws Exception { - when(connection.sendCommandForResponse(List.of("LRANGE", "fruits", "0", "2"))) + when(commandClient.send("localhost", 2020, List.of("LRANGE", "fruits", "0", "2"))) .thenReturn(List.of("apple", "banana")); List values = client.lrange("fruits", 2); @@ -122,7 +122,7 @@ void lrangeWithToUsesZeroAsFromIndex() throws Exception { @Test void lrangeReturnsEmptyListWhenServerReturnsNotFound() throws Exception { - when(connection.sendCommandForResponse(List.of("LRANGE", "missing", "0", "2"))) + when(commandClient.send("localhost", 2020, List.of("LRANGE", "missing", "0", "2"))) .thenReturn(List.of("NOT_FOUND")); List values = client.lrange("missing", 0, 2); @@ -132,7 +132,7 @@ void lrangeReturnsEmptyListWhenServerReturnsNotFound() throws Exception { @Test void metricsReturnsParsedSnapshot() throws Exception { - when(connection.sendCommandForResponse(List.of("METRICS"))).thenReturn(List.of( + when(commandClient.send("localhost", 2020, List.of("METRICS"))).thenReturn(List.of( "METRICS", "hits", "1", "misses", "2", @@ -152,22 +152,15 @@ void metricsReturnsParsedSnapshot() throws Exception { @Test void commandThrowsWhenServerReturnsError() throws Exception { - when(connection.sendCommandForResponse(List.of("GET", "fruit"))).thenReturn(List.of("ERROR", "broken")); + when(commandClient.send("localhost", 2020, List.of("GET", "fruit"))).thenReturn(List.of("ERROR", "broken")); assertThrows(CacheClientException.class, () -> client.get("fruit")); } @Test void commandThrowsWhenConnectionFails() throws Exception { - when(connection.sendCommandForResponse(List.of("GET", "fruit"))).thenThrow(new IOException("closed")); + when(commandClient.send("localhost", 2020, List.of("GET", "fruit"))).thenThrow(new IOException("closed")); assertThrows(CacheClientException.class, () -> client.get("fruit")); } - - @Test - void closeClosesConnection() throws Exception { - client.close(); - - verify(connection).close(); - } } diff --git a/src/test/java/org/cache/cluster/routing/ClusterForwardingClientTest.java b/src/test/java/org/cache/cluster/routing/ClusterForwardingClientTest.java new file mode 100644 index 0000000..5fdf01c --- /dev/null +++ b/src/test/java/org/cache/cluster/routing/ClusterForwardingClientTest.java @@ -0,0 +1,40 @@ +package org.cache.cluster.routing; + +import org.cache.cluster.CacheNode; +import org.cache.network.tcp.connection.RespCommandClient; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class ClusterForwardingClientTest { + + private final CacheNode node = new CacheNode("node-a", "localhost", 8080, 2020, 10001); + + @Test + void forwardSendsCommandToClusterPort() throws Exception { + RespCommandClient commandClient = mock(RespCommandClient.class); + ClusterForwardingClient forwardingClient = new ClusterForwardingClient(commandClient); + + when(commandClient.send("localhost", 10001, List.of("GET", "fruit"))).thenReturn(List.of("VALUE", "apple")); + + assertEquals(List.of("VALUE", "apple"), forwardingClient.forward(node, List.of("GET", "fruit"))); + verify(commandClient).send("localhost", 10001, List.of("GET", "fruit")); + } + + @Test + void forwardWrapsIoFailure() throws Exception { + RespCommandClient commandClient = mock(RespCommandClient.class); + ClusterForwardingClient forwardingClient = new ClusterForwardingClient(commandClient); + + when(commandClient.send("localhost", 10001, List.of("GET", "fruit"))).thenThrow(new IOException("closed")); + + assertThrows(ClusterForwardingException.class, () -> forwardingClient.forward(node, List.of("GET", "fruit"))); + } +} diff --git a/src/test/java/org/cache/cluster/routing/RoutedCacheServiceTest.java b/src/test/java/org/cache/cluster/routing/RoutedCacheServiceTest.java new file mode 100644 index 0000000..a9bf795 --- /dev/null +++ b/src/test/java/org/cache/cluster/routing/RoutedCacheServiceTest.java @@ -0,0 +1,120 @@ +package org.cache.cluster.routing; + +import org.cache.cluster.CacheNode; +import org.cache.cluster.ClusterInfo; +import org.cache.cluster.hashing.ConsistentHashRing; +import org.cache.core.CacheOperations; +import org.cache.protocol.codec.StringKeyCodec; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Optional; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +class RoutedCacheServiceTest { + + private final CacheNode nodeA = node("node-a", 8080, 2020, 10001); + private final CacheNode nodeB = node("node-b", 8081, 2021, 10002); + private final ClusterInfo clusterInfo = new ClusterInfo(1, List.of(nodeA, nodeB)); + private final StringKeyCodec keyCodec = new StringKeyCodec(); + + @Test + void getStringUsesLocalServiceWhenCurrentNodeOwnsKey() { + CacheOperations localService = localService(); + ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); + String key = keyOwnedBy(nodeA); + RoutedCacheService service = new RoutedCacheService<>( + localService, + nodeA, + clusterInfo, + forwardingClient, + keyCodec + ); + + when(localService.getString(key)).thenReturn(Optional.of("apple")); + + assertEquals(Optional.of("apple"), service.getString(key)); + verify(localService).getString(key); + verifyNoInteractions(forwardingClient); + } + + @Test + void getStringForwardsWhenAnotherNodeOwnsKey() { + CacheOperations localService = localService(); + ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); + String key = keyOwnedBy(nodeB); + RoutedCacheService service = new RoutedCacheService<>( + localService, + nodeA, + clusterInfo, + forwardingClient, + keyCodec + ); + + when(forwardingClient.forward(nodeB, List.of("GET", key))).thenReturn(List.of("VALUE", "apple")); + + assertEquals(Optional.of("apple"), service.getString(key)); + verify(forwardingClient).forward(nodeB, List.of("GET", key)); + verifyNoInteractions(localService); + } + + @Test + void lrangeForwardsAndReturnsListValues() { + CacheOperations localService = localService(); + ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); + String key = keyOwnedBy(nodeB); + RoutedCacheService service = new RoutedCacheService<>( + localService, + nodeA, + clusterInfo, + forwardingClient, + keyCodec + ); + + when(forwardingClient.forward(nodeB, List.of("LRANGE", key, "0", "2"))).thenReturn(List.of("one", "two")); + + assertEquals(Optional.of(List.of("one", "two")), service.lrange(key, 0, 2)); + verify(forwardingClient).forward(nodeB, List.of("LRANGE", key, "0", "2")); + verifyNoInteractions(localService); + } + + @Test + void clusterDisabledUsesLocalService() { + CacheOperations localService = localService(); + ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); + RoutedCacheService service = new RoutedCacheService<>(localService, nodeA, null, forwardingClient, keyCodec); + + when(localService.getString("fruit")).thenReturn(Optional.of("apple")); + + assertEquals(Optional.of("apple"), service.getString("fruit")); + verify(localService).getString("fruit"); + verifyNoInteractions(forwardingClient); + } + + @SuppressWarnings("unchecked") + private CacheOperations localService() { + return mock(CacheOperations.class); + } + + private String keyOwnedBy(CacheNode owner) { + ConsistentHashRing ring = new ConsistentHashRing(clusterInfo); + + for (int index = 0; index < 10_000; index++) { + String key = "key-" + index; + if (ring.nodeFor(key).id().equals(owner.id())) { + return key; + } + } + + throw new IllegalStateException("Could not find key owned by node: " + owner.id()); + } + + private CacheNode node(String id, int httpPort, int tcpPort, int clusterPort) { + return new CacheNode(id, "localhost", httpPort, tcpPort, clusterPort); + } +} diff --git a/src/test/java/org/cache/config/CacheConfigLoaderTest.java b/src/test/java/org/cache/config/CacheConfigLoaderTest.java index 0d3f308..92c7eca 100644 --- a/src/test/java/org/cache/config/CacheConfigLoaderTest.java +++ b/src/test/java/org/cache/config/CacheConfigLoaderTest.java @@ -14,6 +14,25 @@ class CacheConfigLoaderTest { + @Test + void loadUsesConfigFileFromSystemProperty() { + String previousConfigFile = System.getProperty("cache.config"); + System.setProperty("cache.config", "config/full-config.yml"); + + try { + CacheConfig config = new CacheConfigLoader().load(); + + assertEquals("node-test", config.cacheNode().id()); + assertEquals(250, config.capacity()); + } finally { + if (previousConfigFile == null) { + System.clearProperty("cache.config"); + } else { + System.setProperty("cache.config", previousConfigFile); + } + } + } + @Test void loadUsesDefaultsWhenConfigFileDoesNotExist() { CacheConfig config = new CacheConfigLoader("missing-config.yml").load(); diff --git a/src/test/java/org/cache/network/connection/ClientConnectionHandlerTest.java b/src/test/java/org/cache/network/connection/ClientConnectionHandlerTest.java index 1f76b15..494bd98 100644 --- a/src/test/java/org/cache/network/connection/ClientConnectionHandlerTest.java +++ b/src/test/java/org/cache/network/connection/ClientConnectionHandlerTest.java @@ -1,13 +1,13 @@ package org.cache.network.connection; import org.cache.network.tcp.connection.ClientConnectionHandler; -import org.cache.protocol.CommandProcessor; import org.junit.jupiter.api.Test; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.net.Socket; import java.util.List; +import java.util.function.Function; import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -20,17 +20,17 @@ class ClientConnectionHandlerTest { @Test void runProcessesCommandAndWritesResponse() throws Exception { Socket socket = mock(Socket.class); - CommandProcessor commandProcessor = mock(CommandProcessor.class); + Function, String> commandProcessor = mock(Function.class); var input = new ByteArrayInputStream("GET fruit\r\n".getBytes(UTF_8)); var output = new ByteArrayOutputStream(); when(socket.getInputStream()).thenReturn(input); when(socket.getOutputStream()).thenReturn(output); - when(commandProcessor.process(List.of("GET", "fruit"))).thenReturn("VALUE apple"); + when(commandProcessor.apply(List.of("GET", "fruit"))).thenReturn("VALUE apple"); new ClientConnectionHandler(socket, commandProcessor).run(); - verify(commandProcessor).process(List.of("GET", "fruit")); + verify(commandProcessor).apply(List.of("GET", "fruit")); verify(socket).close(); assertEquals("VALUE apple\r\n", output.toString(UTF_8)); } diff --git a/src/test/java/org/cache/network/http/CacheControllerTest.java b/src/test/java/org/cache/network/http/CacheControllerTest.java index 157649e..8810c74 100644 --- a/src/test/java/org/cache/network/http/CacheControllerTest.java +++ b/src/test/java/org/cache/network/http/CacheControllerTest.java @@ -1,6 +1,6 @@ package org.cache.network.http; -import org.cache.core.CacheService; +import org.cache.core.CacheOperations; import org.cache.core.ValueType; import org.cache.core.metrics.Snapshot; import org.cache.network.http.dto.GetResponseDto; @@ -27,13 +27,13 @@ class CacheControllerTest { - private CacheService cacheService; + private CacheOperations cacheService; private CacheController controller; @BeforeEach @SuppressWarnings("unchecked") void setUp() { - cacheService = mock(CacheService.class); + cacheService = mock(CacheOperations.class); controller = new CacheController(cacheService, (KeyCodec) (KeyCodec) new StringKeyCodec()); }