From ddf0724fb21016422dfeee39f58697f06c3cb533 Mon Sep 17 00:00:00 2001 From: unikdahal Date: Wed, 28 Jan 2026 04:55:17 +0530 Subject: [PATCH] Added Comments For Readability --- .../com/redis/commands/CommandRegistry.java | 91 +++++--- .../redis/commands/stream/XAddCommand.java | 60 ++++- .../redis/commands/stream/XRangeCommand.java | 58 ++++- .../redis/commands/stream/XReadCommand.java | 105 ++++++--- .../java/com/redis/config/RedisConfig.java | 4 +- .../com/redis/server/RedisCommandHandler.java | 182 ++++++++++----- .../java/com/redis/storage/ExpiryManager.java | 4 +- .../java/com/redis/storage/RedisDatabase.java | 219 +++++++++-------- .../java/com/redis/storage/RedisValue.java | 220 +++++++++++++----- src/main/java/com/redis/util/ExpiryTask.java | 23 +- src/main/java/com/redis/util/StreamId.java | 55 ++++- 11 files changed, 695 insertions(+), 326 deletions(-) diff --git a/src/main/java/com/redis/commands/CommandRegistry.java b/src/main/java/com/redis/commands/CommandRegistry.java index 5b1c329..b2b8eb1 100644 --- a/src/main/java/com/redis/commands/CommandRegistry.java +++ b/src/main/java/com/redis/commands/CommandRegistry.java @@ -7,48 +7,78 @@ import java.util.concurrent.ConcurrentHashMap; /** - * Registry for all available Redis commands. - * Supports dynamic command registration and lookup. - * This is a singleton initialized with built-in commands via ServiceLoader. - * - * Optimizations: - * - Case-insensitive lookups with single toUpperCase() call - * - ConcurrentHashMap with capacity hints for faster lookup - * - Dynamic discovery via ServiceLoader for easy extensibility + * The Central Command Registry. + *

+ * Role: Acts as the directory for all executable commands. It maps a string (e.g., "SET") + * to the actual Java object capable of executing that logic. + *

+ * Design Pattern: Singleton + * We use a Singleton because the list of commands is static for the application's lifetime. + * Re-scanning for commands for every client connection would be incredibly slow. + *

+ * Design Pattern: Strategy / Command + * This registry enables the "Command Pattern". The network layer doesn't need to know how "SET" works; + * it just retrieves the command object and calls .execute(). */ public class CommandRegistry { - private static CommandRegistry INSTANCE; + // The single instance of this class (Volatile is implied by the memory model of synchronized, + // but usually good practice to mark volatile in double-checked locking to prevent instruction reordering). + private static volatile CommandRegistry INSTANCE; + + // The actual storage. + // Key = Command Name (UPPERCASE), Value = Command Object. + // We use ConcurrentHashMap because the registry might be read by multiple Netty threads simultaneously. + // While writes happen mostly at startup, safe reads are critical. private final Map registry = new ConcurrentHashMap<>(32); /** - * Initialize the singleton registry and register all commands found via ServiceLoader. + * Private constructor to enforce Singleton usage. + *

+ * Mechanism: ServiceLoader (SPI) + * Instead of hardcoding "new SetCommand()", we ask Java to look at the classpath. + * Java looks for a file: META-INF/services/com.redis.commands.ICommand + * It reads the class names listed there and instantiates them. */ private CommandRegistry() { + // Step 1: Initialize the loader for the ICommand interface ServiceLoader loader = ServiceLoader.load(ICommand.class); + + // Step 2: Iterate through found implementations. + // The ServiceLoader lazily instantiates the classes as we iterate. for (ICommand cmd : loader) { register(cmd); } + + // Logging is helpful to verify that your META-INF file is set up correctly. System.out.println("[Redis] Registered " + registry.size() + " commands"); } /** - * Get the singleton instance of CommandRegistry. + * The Holder Class. + *

+ * 1. Lazy: This class is NOT loaded when CommandRegistry is loaded. + * It is only loaded when getInstance() is called for the first time. + * 2. Thread-Safe: The JVM guarantees that static field initialization + * (INSTANCE = new ...) happens atomically. No synchronized keyword needed. + */ + private static class RegistryHolder { + private static final CommandRegistry INSTANCE = new CommandRegistry(); + } + + /** + * Get the singleton instance. + * Triggers the loading of RegistryHolder and the creation of INSTANCE. */ public static CommandRegistry getInstance() { - if (INSTANCE == null) { - synchronized (CommandRegistry.class) { - if (INSTANCE == null) { - INSTANCE = new CommandRegistry(); - } - } - } - return INSTANCE; + return RegistryHolder.INSTANCE; } /** - * Register a command in the registry. - * Command names are stored in uppercase for case-insensitive lookup. + * Registers a command into the map. + *

+ * Normalization: We store all keys in UPPERCASE. This ensures that + * "set", "Set", and "SET" all resolve to the same entry. */ public void register(ICommand cmd) { String cmdName = cmd.name().toUpperCase(); @@ -56,34 +86,37 @@ public void register(ICommand cmd) { } /** - * Look up a command by name (case-insensitive). - * Returns the command instance or null if not found. - * Optimization: Single toUpperCase() call, direct HashMap lookup + * Retrieves the command object for a given name. + *

+ * Performance: This is a "Hot Path" method called for every single request. + * It must be O(1) and very fast. + * + * @param name The command name (e.g., "set") + * @return The ICommand instance, or null if not found. */ public ICommand get(String name) { if (name == null || name.isEmpty()) { return null; } + // Convert input to uppercase to match the storage key format. return registry.get(name.toUpperCase()); } /** - * Check if a command is registered. + * Utility to check command existence without retrieving it. */ public boolean exists(String name) { return get(name) != null; } /** - * Get all registered command names (unmodifiable). + * Returns a Read-Only view of all available commands. + * Useful for the "COMMAND" command in Redis which lists capabilities. */ public Set getRegisteredCommands() { return Collections.unmodifiableSet(registry.keySet()); } - /** - * Get the number of registered commands. - */ public int size() { return registry.size(); } diff --git a/src/main/java/com/redis/commands/stream/XAddCommand.java b/src/main/java/com/redis/commands/stream/XAddCommand.java index 6dfbb36..e656cc7 100644 --- a/src/main/java/com/redis/commands/stream/XAddCommand.java +++ b/src/main/java/com/redis/commands/stream/XAddCommand.java @@ -13,11 +13,20 @@ import java.util.concurrent.atomic.AtomicReference; /** - * XADD key ID field value [field value ...] - * Appends the specified stream entry to the stream at the specified key. - * If the key does not exist, as a side effect the stream is created. + * Implementation of the XADD command. + *

+ * Syntax: XADD key ID field value [field value ...] + *

+ * Role: Appends a new entry to a stream. This is the "Writer" command for streams. + *

+ * Concurrency Strategy: + * This command relies heavily on {@link RedisDatabase#compute} to ensure atomic ID generation. + * Since Stream IDs must be strictly monotonic (always increasing), we must lock the stream + * while we calculate the next ID to prevent two clients from generating the same ID simultaneously. */ public class XAddCommand implements ICommand { + + // Standard Redis Error Messages private static final String ERR_WRONG_ARGS = "-ERR wrong number of arguments for 'XADD' command\r\n"; private static final String ERR_WRONG_TYPE = "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n"; private static final String ERR_ID_TOO_SMALL = "-ERR The ID specified in XADD is equal or smaller than the target stream top item\r\n"; @@ -25,6 +34,9 @@ public class XAddCommand implements ICommand { @Override public String execute(List args, ChannelHandlerContext ctx) { + // Minimum args: key, ID, field, value (4 args -> size 4). + // Note: The List args usually includes key at 0. + // Format: XADD ... if (args.size() < 3) { return ERR_WRONG_ARGS; } @@ -32,81 +44,117 @@ public String execute(List args, ChannelHandlerContext ctx) { String key = args.get(0); String idArg = args.get(1); + // Validation: Fields and Values must come in pairs int fieldStart = 2; - // Basic argument validation for field-value pairs if ((args.size() - fieldStart) % 2 != 0) { - return ERR_WRONG_ARGS; + return ERR_WRONG_ARGS; } + // Parse fields into a LinkedHashMap to preserve insertion order (Redis convention) Map fields = new LinkedHashMap<>(); for (int i = fieldStart; i < args.size(); i += 2) { fields.put(args.get(i), args.get(i + 1)); } RedisDatabase db = RedisDatabase.getInstance(); + + // We use AtomicReferences to extract results/errors from inside the lambda AtomicReference error = new AtomicReference<>(null); AtomicReference addedId = new AtomicReference<>(null); + // CRITICAL SECTION: Atomic Read-Modify-Write + // We lock the key to ensure no one else inserts while we determine the next ID. db.compute(key, existing -> { ConcurrentSkipListMap> streamMap; + + // 1. Initialization / Type Check if (existing == null) { + // New Key: Create a new SkipList (Ordered Thread-Safe Map) streamMap = new ConcurrentSkipListMap<>(); } else if (existing.getType() != RedisValue.Type.STREAM) { + // Wrong Type: Cannot append stream data to a String/List error.set(ERR_WRONG_TYPE); return existing; } else { + // Existing Key: Cast the raw data @SuppressWarnings("unchecked") var data = (Map>) existing.getData(); + // We know it's a SkipList because we created it that way in RedisValue factory streamMap = (ConcurrentSkipListMap>) data; } + // 2. Determine Context (What is the last ID?) + // If stream is empty, assume "0-0" is the predecessor StreamId lastId = streamMap.isEmpty() ? new StreamId(0, 0) : streamMap.lastKey(); StreamId newId; try { + // 3. ID Generation Logic if (idArg.equals("*")) { + // AUTO-GENERATE: "*" long now = System.currentTimeMillis(); if (now > lastId.time()) { + // Standard case: New millisecond, reset sequence to 0 newId = new StreamId(now, 0); } else { + // Collision or Clock Skew: Time is the same (or older) than the last entry. + // We must increment the sequence number of the LAST entry's time. + // This handles high-throughput bursts within the same ms. newId = new StreamId(lastId.time(), lastId.sequence() + 1); } } else if (idArg.endsWith("-*")) { + // PARTIAL ID: "123456-*" long ms = Long.parseLong(idArg.substring(0, idArg.length() - 2)); + if (ms < lastId.time()) { error.set(ERR_ID_TOO_SMALL); return existing; } + // If time is the same, increment sequence. If time is new, seq is 0. long seq = (ms == lastId.time()) ? lastId.sequence() + 1 : 0; newId = new StreamId(ms, seq); } else { + // EXPLICIT ID: "123456-0" newId = StreamId.parse(idArg); } + // 4. Validation Rules + // Rule A: ID must be > 0-0 if (newId.time() == 0 && newId.sequence() == 0) { error.set(ERR_ID_ZERO); return existing; } + // Rule B: ID must be strictly greater than the last ID + // Note: If stream is empty, lastId is 0-0, so any valid ID passes. if (!streamMap.isEmpty() && !newId.isGreaterThan(lastId)) { error.set(ERR_ID_TOO_SMALL); return existing; } + // 5. Execution: Insert into map streamMap.put(newId, fields); addedId.set(newId); + + // Return wrapped value to update DB (or keep existing reference) return RedisValue.stream(streamMap); + } catch (Exception e) { + // Handle parsing errors (e.g., malformed ID string) error.set("-ERR " + e.getMessage() + "\r\n"); return existing; } }); + // 6. Response Handling + // If the lambda set an error, return it. if (error.get() != null) { return error.get(); } + // Otherwise return the ID we just generated/inserted. String idStr = addedId.get().toString(); + // Redis Bulk String format: $\r\n\r\n return "$" + idStr.length() + "\r\n" + idStr + "\r\n"; } @@ -114,4 +162,4 @@ public String execute(List args, ChannelHandlerContext ctx) { public String name() { return "XADD"; } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/commands/stream/XRangeCommand.java b/src/main/java/com/redis/commands/stream/XRangeCommand.java index 9dada69..779d430 100644 --- a/src/main/java/com/redis/commands/stream/XRangeCommand.java +++ b/src/main/java/com/redis/commands/stream/XRangeCommand.java @@ -11,15 +11,22 @@ import java.util.NavigableMap; /** - * XRANGE key start end [COUNT count] - * Returns the stream entries matching a given range of IDs. + * Implementation of the XRANGE command. + *

+ * Syntax: XRANGE key start end [COUNT count] + *

+ * Role: Performs efficient range queries on the Stream. + * Because the Stream is backed by a {@link java.util.concurrent.ConcurrentSkipListMap}, + * range lookups are O(log N) rather than O(N). */ public class XRangeCommand implements ICommand { + private static final String ERR_WRONG_ARGS = "-ERR wrong number of arguments for 'XRANGE' command\r\n"; private static final String ERR_WRONG_TYPE = "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n"; @Override public String execute(List args, ChannelHandlerContext ctx) { + // Syntax: XRANGE [COUNT ] if (args.size() < 3) { return ERR_WRONG_ARGS; } @@ -28,6 +35,7 @@ public String execute(List args, ChannelHandlerContext ctx) { String startArg = args.get(1); String endArg = args.get(2); + // Optional COUNT argument handling int count = Integer.MAX_VALUE; if (args.size() >= 5 && args.get(3).equalsIgnoreCase("COUNT")) { try { @@ -40,42 +48,55 @@ public String execute(List args, ChannelHandlerContext ctx) { RedisDatabase db = RedisDatabase.getInstance(); RedisValue value = db.getValue(key); + // Case 1: Stream does not exist -> Return Empty Array (Standard Redis behavior) if (value == null) { return "*0\r\n"; } + // Case 2: Key exists but is not a Stream -> Error if (value.getType() != RedisValue.Type.STREAM) { return ERR_WRONG_TYPE; } + // Retrieve the sorted map @SuppressWarnings("unchecked") NavigableMap> streamMap = (NavigableMap>) value.getData(); + // Parse bounds with context awareness (Start vs End) StreamId start = parseBound(startArg, true); StreamId end = parseBound(endArg, false); if (start == null || end == null) { - return "-ERR Invalid stream ID specified as range part\r\n"; + return "-ERR Invalid stream ID specified as range part\r\n"; } + // CORE LOGIC: Get a view of the map for the requested range. + // true, true means inclusive boundaries [start, end]. + // This is efficient (O(log N)) and does not copy data. NavigableMap> range = streamMap.subMap(start, true, end, true); + // Build RESP Array Response StringBuilder sb = new StringBuilder(); + + // Calculate exact size (considering COUNT limit) int entriesToReturn = Math.min(count, range.size()); sb.append("*").append(entriesToReturn).append("\r\n"); int i = 0; for (Map.Entry> entry : range.entrySet()) { if (i++ >= count) break; - + + // Start Entry Array sb.append("*2\r\n"); - // Entry ID + + // 1. The ID String idStr = entry.getKey().toString(); sb.append("$").append(idStr.length()).append("\r\n").append(idStr).append("\r\n"); - - // Entry Fields + + // 2. The Field-Value pairs Map fields = entry.getValue(); sb.append("*").append(fields.size() * 2).append("\r\n"); + for (Map.Entry field : fields.entrySet()) { String k = field.getKey(); String v = field.getValue(); @@ -87,19 +108,32 @@ public String execute(List args, ChannelHandlerContext ctx) { return sb.toString(); } + /** + * Parses user input for Range bounds. + * Handles special characters ('-', '+') and incomplete IDs ('1000'). + * + * @param bound The string argument (e.g., "1500", "1500-1", "-", "+") + * @param start True if this is the Start bound, False if End bound. + */ private StreamId parseBound(String bound, boolean start) { - if (bound.equals("-")) return StreamId.MIN; - if (bound.equals("+")) return StreamId.MAX; - + // Special Bounds + if (bound.equals("-")) return StreamId.MIN; // 0-0 + if (bound.equals("+")) return StreamId.MAX; // MaxLong-MaxLong + + // Case: Incomplete ID (e.g., "1000") if (!bound.contains("-")) { try { long ms = Long.parseLong(bound); + // Context Aware Logic: + // Start: "1000" -> 1000-0 (Beginning of ms) + // End: "1000" -> 1000-MAX (End of ms) return start ? new StreamId(ms, 0) : new StreamId(ms, Long.MAX_VALUE); } catch (NumberFormatException e) { return null; } } - + + // Case: Explicit ID (e.g., "1000-1") try { return StreamId.parse(bound); } catch (IllegalArgumentException e) { @@ -111,4 +145,4 @@ private StreamId parseBound(String bound, boolean start) { public String name() { return "XRANGE"; } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/commands/stream/XReadCommand.java b/src/main/java/com/redis/commands/stream/XReadCommand.java index 6d21f86..e190de4 100644 --- a/src/main/java/com/redis/commands/stream/XReadCommand.java +++ b/src/main/java/com/redis/commands/stream/XReadCommand.java @@ -15,23 +15,29 @@ import java.util.concurrent.TimeUnit; /** - * XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...] - * Read data from one or multiple streams, only returning entries with an ID greater - * than the last received ID reported by the caller. + * Implementation of XREAD (Blocking Stream Read). + *

+ * Syntax: XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...] + *

+ * Architecture: Asynchronous Polling + * This command demonstrates how to handle "Blocking" operations in a Non-Blocking framework (Netty). + * We cannot block the thread (Thread.sleep) because it would freeze the server. + * Instead, if no data is found, we schedule a background task to check again later (Polling). */ public class XReadCommand implements ICommand { private static final String ERR_WRONG_ARGS = "-ERR wrong number of arguments for 'XREAD' command\r\n"; private static final String ERR_WRONG_TYPE = "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n"; - private static final String RESP_NIL_ARRAY = "*-1\r\n"; - private static final long POLL_INTERVAL_MS = 50; + private static final String RESP_NIL_ARRAY = "*-1\r\n"; // Standard Redis "Null Array" response + private static final long POLL_INTERVAL_MS = 50; // How often to check for new data during BLOCK @Override public String execute(List args, ChannelHandlerContext ctx) { if (args.size() < 3) return ERR_WRONG_ARGS; - int count = Integer.MAX_VALUE; - long blockMs = -1; - int streamsIdx = -1; + // --- Step 1: Parse Options (COUNT, BLOCK) --- + int count = Integer.MAX_VALUE; // Default: Return all available + long blockMs = -1; // Default: Non-blocking + int streamsIdx = -1; // Index where "STREAMS" keyword appears try { for (int i = 0; i < args.size(); i++) { @@ -44,21 +50,26 @@ public String execute(List args, ChannelHandlerContext ctx) { blockMs = Long.parseLong(args.get(++i)); } else if (arg.equals("STREAMS")) { streamsIdx = i + 1; - break; + break; // Parsing done, everything after this is Keys/IDs } } } catch (NumberFormatException e) { return "-ERR value is not an integer or out of range\r\n"; } + // Validation: Must have STREAMS keyword and arguments following it if (streamsIdx == -1 || streamsIdx >= args.size()) return ERR_WRONG_ARGS; + // The remaining arguments must be even (Key1, Key2 ... ID1, ID2) int remainingArgs = args.size() - streamsIdx; if (remainingArgs <= 0 || remainingArgs % 2 != 0) return ERR_WRONG_ARGS; int numStreams = remainingArgs / 2; List keys = new ArrayList<>(numStreams); List ids = new ArrayList<>(numStreams); + + // Split arguments into Keys and IDs lists + // Example: STREAMS s1 s2 0-0 0-0 -> Keys:[s1, s2], IDs:[0-0, 0-0] for (int i = 0; i < numStreams; i++) { keys.add(args.get(streamsIdx + i)); ids.add(args.get(streamsIdx + numStreams + i)); @@ -66,72 +77,98 @@ public String execute(List args, ChannelHandlerContext ctx) { RedisDatabase db = RedisDatabase.getInstance(); - // Immediate read + // --- Step 2: Attempt Immediate Read --- + // If data is already there, return immediately. No need to block. String result = tryRead(db, keys, ids, count); if (result != null) return result; - if (blockMs < 0) return "*-1\r\n"; // Non-blocking and no data: Redis returns nil for XREAD? - // Wait, standard Redis XREAD (without BLOCK) returns (nil) if no data. + // --- Step 3: Handle Blocking --- + + // Case A: No BLOCK option requested. + // Standard Redis behavior: If no data found in non-blocking mode, return Nil. + if (blockMs < 0) return "*-1\r\n"; - // Blocking read + // Case B: BLOCK requested. + // Calculate when we should stop waiting. (0 means block forever). long deadline = blockMs == 0 ? Long.MAX_VALUE : System.currentTimeMillis() + blockMs; if (ctx.executor() != null) { + // ASYNC PATH (Correct for Netty): + // We return null to tell the main handler "Don't send a response yet". + // We schedule a background task to keep checking. schedulePolling(ctx, keys, ids, count, deadline, db); return null; } else { + // SYNC PATH (Fallback / Testing): + // Should be avoided in production as it blocks the thread. return pollSynchronously(keys, ids, count, deadline, db); } } + /** + * core logic to check the database for matching entries. + * Returns a RESP-formatted string if data is found, or null if nothing is found. + */ private String tryRead(RedisDatabase db, List keys, List idArgs, int count) { List streamResponses = new ArrayList<>(); for (int i = 0; i < keys.size(); i++) { String key = keys.get(i); String idArg = idArgs.get(i); - + RedisValue value = db.getValue(key); - if (value == null) continue; - if (value.getType() != RedisValue.Type.STREAM) { - // In Redis, if one key is not a stream, it might error or skip. - // Usually it errors. - continue; - } + if (value == null) continue; // Stream doesn't exist + if (value.getType() != RedisValue.Type.STREAM) continue; // Skip wrong types + // Access the underlying SkipList @SuppressWarnings("unchecked") NavigableMap> streamMap = (NavigableMap>) value.getData(); - + + // Resolve the ID to start reading from StreamId lastId; if (idArg.equals("$")) { + // Special ID "$": Means "Only new messages". + // So we start strictly AFTER the current last key. lastId = streamMap.isEmpty() ? new StreamId(0, 0) : streamMap.lastKey(); } else { try { lastId = StreamId.parse(idArg); } catch (IllegalArgumentException e) { - continue; // Skip invalid ID for now + continue; // Skip invalid IDs } } + // CRITICAL: tailMap(lastId, false) + // 'false' means EXCLUSIVE (strictly greater than lastId). + // This gives us O(log N) access to the new entries. NavigableMap> tail = streamMap.tailMap(lastId, false); if (tail.isEmpty()) continue; + // --- Build RESP Response for this Stream --- StringBuilder streamSb = new StringBuilder(); + + // 1. Array Header for this stream (Name + Entries) streamSb.append("*2\r\n"); - // Stream name + + // 2. Stream Name streamSb.append("$").append(key.length()).append("\r\n").append(key).append("\r\n"); - - // Entries + + // 3. Array of Entries int entriesToReturn = Math.min(count, tail.size()); streamSb.append("*").append(entriesToReturn).append("\r\n"); - + int j = 0; for (Map.Entry> entry : tail.entrySet()) { if (j++ >= count) break; + + // Entry Structure: [ID, [Field, Value, ...]] streamSb.append("*2\r\n"); + + // ID String sid = entry.getKey().toString(); streamSb.append("$").append(sid.length()).append("\r\n").append(sid).append("\r\n"); - + + // Field-Value Array Map fields = entry.getValue(); streamSb.append("*").append(fields.size() * 2).append("\r\n"); for (Map.Entry f : fields.entrySet()) { @@ -146,29 +183,41 @@ private String tryRead(RedisDatabase db, List keys, List idArgs, if (streamResponses.isEmpty()) return null; + // Wrap all stream responses in one outer array StringBuilder sb = new StringBuilder(); sb.append("*").append(streamResponses.size()).append("\r\n"); for (String s : streamResponses) sb.append(s); return sb.toString(); } + /** + * Async Polling Logic. + * Schedules itself to run repeatedly on the EventLoop until data is found or timeout occurs. + */ private void schedulePolling(ChannelHandlerContext ctx, List keys, List ids, int count, long deadline, RedisDatabase db) { ctx.executor().schedule(() -> { + // 1. Check Timeout if (System.currentTimeMillis() >= deadline) { writeResponse(ctx, RESP_NIL_ARRAY); return; } + // 2. Check Data String result = tryRead(db, keys, ids, count); if (result != null) { writeResponse(ctx, result); return; } + // 3. Reschedule (Recursive step, but async so no stack overflow) schedulePolling(ctx, keys, ids, count, deadline, db); }, POLL_INTERVAL_MS, TimeUnit.MILLISECONDS); } + /** + * Fallback for contexts without an executor (e.g. unit tests). + * WARNING: Blocks the calling thread. + */ private String pollSynchronously(List keys, List ids, int count, long deadline, RedisDatabase db) { while (System.currentTimeMillis() < deadline) { String result = tryRead(db, keys, ids, count); @@ -187,4 +236,4 @@ private void writeResponse(ChannelHandlerContext ctx, String response) { public String name() { return "XREAD"; } -} +} \ No newline at end of file diff --git a/src/main/java/com/redis/config/RedisConfig.java b/src/main/java/com/redis/config/RedisConfig.java index e47ad01..d6de2c9 100644 --- a/src/main/java/com/redis/config/RedisConfig.java +++ b/src/main/java/com/redis/config/RedisConfig.java @@ -6,11 +6,11 @@ /** * Central configuration for the Redis server. - * Loads defaults and can be overridden by application.properties or environment variables. + * Loads default and can be overridden by application.properties or environment variables. */ public class RedisConfig { private static RedisConfig instance; - private Properties properties = new Properties(); + private final Properties properties = new Properties(); // Default configuration values private static final int DEFAULT_PORT = 6379; diff --git a/src/main/java/com/redis/server/RedisCommandHandler.java b/src/main/java/com/redis/server/RedisCommandHandler.java index 8b87241..193f903 100644 --- a/src/main/java/com/redis/server/RedisCommandHandler.java +++ b/src/main/java/com/redis/server/RedisCommandHandler.java @@ -12,62 +12,95 @@ import java.util.List; /** - * Netty channel handler for processing incoming Redis commands. - * Parses RESP protocol using ByteToMessageDecoder for fragmentation support. - * - * IMPORTANT: This handler is instantiated per-channel (see NettyRedisServer.initChannel()), - * ensuring thread-safety for the argsBuffer field. Each channel has its own handler instance. - * - * Optimizations: - * - Reuses ArrayList for argument parsing - * - Fast integer parsing without object allocation - * - Robust handling of pipelined commands + * Netty channel handler responsible for the "Framing" phase of the Redis protocol. + *

+ * Role in Pipeline: + * This handler sits between the raw network bytes and the command execution logic. + * Its job is to turn a stream of fragmented TCP bytes into discrete, actionable Redis commands. + *

+ * Thread Safety & Lifecycle: + * IMPORTANT: This class is stateful (it holds `argsBuffer`). It must be instantiated + * per-channel (one instance per connected client). It is NOT @ChannelHandler.Sharable. + *

+ * Optimizations: + * 1. Zero-Allocation Integer Parsing: Avoids creating String objects just to read lengths. + * 2. Object Pooling: Reuses a single ArrayList {@code argsBuffer} for the lifetime of the connection + * to reduce Garbage Collection pressure during high-throughput bursts. */ public class RedisCommandHandler extends ByteToMessageDecoder { - // Preallocate list to avoid allocations for small commands + + // Initial capacity for the arguments list. Most Redis commands have fewer than 16 arguments. private static final int INITIAL_ARGS_CAPACITY = 16; + + /** + * A reusable buffer for command arguments. + * We clear and refill this list for every command instead of allocating a new ArrayList each time. + * This significantly reduces "Object Churn" in the JVM. + */ private final List argsBuffer = new ArrayList<>(INITIAL_ARGS_CAPACITY); /** - * Parses and processes as many complete RESP array commands as are available in the input buffer. - * - *

Reads RESP arrays from {@code in}, reusing the handler's argument buffer to collect the command - * name and arguments. If a full command is not yet available the reader index is reset and decoding - * stops so more data can arrive. For each complete command, resolves the command implementation, - * executes it with a view of the argument list, and writes the RESP response to the channel. Unknown - * commands result in the Redis error reply "-ERR unknown command 'name'". This method supports - * pipelined commands by looping until the input buffer has no more complete commands.

+ * Sentinel value indicating that the buffer does not yet contain a complete integer ending in CRLF. + */ + private static final int INCOMPLETE = Integer.MIN_VALUE; + + /** + * The main entry point called by Netty whenever new data arrives from the network. + * * @param ctx Context to interact with the channel pipeline (e.g., writing responses). * - * @param ctx the Netty channel handler context used to write responses - * @param in the inbound byte buffer containing RESP data; reader index is advanced for consumed data - * @param out the list to which decoded messages would normally be added (not used; responses are written directly) + * @param in The input ByteBuf containing raw bytes received from the OS. + * @param out (Unused) We write responses directly to ctx, rather than passing objects up the pipeline. */ @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) { + // PIPELINING SUPPORT: + // A client might send 5 commands in a single TCP packet. We must loop to process + // all of them before returning control to the Netty event loop. while (in.readableBytes() > 0) { + + // CHECKPOINTING: + // We mark the current reader index. If we start parsing a command but realize + // the packet is fragmented (incomplete data), we will reset the reader index + // back to this mark so we can try again later when the rest of the data arrives. in.markReaderIndex(); + + // Reset the reusable argument buffer for the new command argsBuffer.clear(); + // Attempt to parse one complete RESP command if (!parseRespArray(in, argsBuffer)) { + // FAILURE CASE (FRAGMENTATION): + // We ran out of data in the middle of a command. + // Reset the read pointer to the start of this command and return. + // Netty will call decode() again when more bytes arrive. in.resetReaderIndex(); - return; // Incomplete command, wait for more data + return; } + // Edge case: Parser returned true but found an empty array (unlikely in valid RESP but possible) if (argsBuffer.isEmpty()) { continue; } - // Extract command name and arguments + // COMMAND RESOLUTION: + // The first element of the array is always the command name (e.g., "SET", "GET"). String commandName = argsBuffer.get(0); ICommand cmd = CommandRegistry.getInstance().get(commandName); if (cmd == null) { + // Protocol requires reporting unknown commands to the client writeResponse(ctx, "-ERR unknown command '" + commandName + "'\r\n"); } else { - // Create a sublist for command args (view only) + // Create a view of the arguments (skipping the command name). + // subList is a lightweight view, not a copy. List commandArgs = argsBuffer.size() > 1 ? argsBuffer.subList(1, argsBuffer.size()) : List.of(); + + // EXECUTION: + // Run the actual logic (e.g., modifying the KeyValue store). String resp = cmd.execute(commandArgs, ctx); - // null response means async handling (e.g., BLPOP) + + // If response is not null, write it immediately. + // Null responses implies the command handles its own writing asynchronously (e.g., blocking ops). if (resp != null) { writeResponse(ctx, resp); } @@ -75,101 +108,127 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) { } } - private static final int INCOMPLETE = Integer.MIN_VALUE; - /** - * Parses a RESP array from the given ByteBuf and appends its elements to the provided list. - * - * Each bulk string element is added as a UTF-8 Java String; nil bulk strings are represented as `null`. + * Parses a RESP Array (the standard format for Redis commands). + *

+ * Protocol Format: {@code *\r\n$\r\n\r\n...} + *

+ * All-or-Nothing Strategy: + * This method returns {@code false} immediately if the buffer is missing ANY part of the command. + * It ensures we never partially consume the buffer, which simplifies state management. * - * @param buf the ByteBuf containing RESP-encoded data (reader index will be advanced as bytes are consumed) - * @param result the list to populate with parsed array elements; existing contents are not cleared by this method - * @return `true` if a complete RESP array was parsed and its elements appended to `result`, `false` if more data is required or the input is malformed + * @param buf The network buffer. + * @param result The list to populate with parsed strings. + * @return {@code true} if a full command was parsed, {@code false} if data is incomplete. */ private boolean parseRespArray(ByteBuf buf, List result) { try { + // Need at least 1 byte to check for the '*' marker if (buf.readableBytes() < 1) return false; - - // Check for Array Start (*) + + // VALIDATION: RESP Arrays must start with '*' if (buf.readByte() != '*') { - return false; + return false; // Malformed request or wrong protocol } - // Read Array Length + // Step 1: Read the number of arguments in the array int numArgs = readInteger(buf); - if (numArgs == INCOMPLETE) return false; - if (numArgs < 0) return true; // Nil array (*) or empty + if (numArgs == INCOMPLETE) return false; // Waiting for length delimiter + + if (numArgs < 0) return true; // Handle null array (rare in requests, common in responses) - // Read All Arguments + // Step 2: Loop through each argument for (int i = 0; i < numArgs; i++) { + // Need at least 1 byte for '$' if (buf.readableBytes() < 1) return false; - - // Check for Bulk String Start ($) + + // VALIDATION: Bulk Strings must start with '$' byte marker = buf.readByte(); if (marker != '$') return false; - // Read String Length + // Step 3: Read the length of the string int strLen = readInteger(buf); - if (strLen == INCOMPLETE) return false; + if (strLen == INCOMPLETE) return false; // Waiting for string length + + // Handle special case: Null Bulk String ($-1) if (strLen < 0) { result.add(null); continue; } + // Step 4: BOUNDS CHECKING (Crucial) + // We verify we have the FULL string payload + the 2 trailing bytes (\r\n) + // BEFORE we attempt to read. This prevents reading half a string. if (buf.readableBytes() < strLen + 2) return false; - // Read the actual String data + // Step 5: Materialize the String + // readCharSequence reads bytes and advances the readerIndex CharSequence arg = buf.readCharSequence(strLen, StandardCharsets.UTF_8); result.add(arg.toString()); - // Skip trailing \r\n + // Consume the trailing CRLF (\r\n) required by RESP buf.skipBytes(2); } - return true; + return true; // Success: Full command parsed } catch (Exception e) { + // In production, you might log this. Returning false triggers a resetReaderIndex. return false; } } /** - * Parse a RESP-style integer directly from the given ByteBuf and advance the reader index past its terminating CRLF. - * - * The method accepts an optional leading '-' for negative values and consumes the trailing "\r\n" when a full - * integer is available. If the buffer does not contain a complete integer line (no '\r' found or not enough bytes - * for the terminating CRLF), the reader index is not advanced and the method returns the INCOMPLETE sentinel. + * Custom integer parser optimized for the RESP protocol. + *

+ * Why not Integer.parseInt? + * Standard parsing requires extracting a substring (allocation), creating a String object (allocation), + * and then parsing it. This method reads raw bytes and computes the integer mathematically, + * generating zero garbage. * - * @param buf the buffer to read the integer from - * @return the parsed integer (negative if prefixed with '-'), or INCOMPLETE if more data is required + * @param buf The buffer to read from. + * @return The parsed integer, or {@code INCOMPLETE} if the delimiter (\r) isn't found yet. */ private int readInteger(ByteBuf buf) { + // Step 1: Scan for the end of the line ('\r') + // We look from current readerIndex up to writerIndex (available data) int rIndex = buf.indexOf(buf.readerIndex(), buf.writerIndex(), (byte) '\r'); + + // If no '\r' found, we don't have the full number yet. if (rIndex == -1) return INCOMPLETE; - if (buf.readableBytes() < (rIndex - buf.readerIndex() + 2)) return INCOMPLETE; // Need \r\n + + // Ensure we also have the '\n' following the '\r' + if (buf.readableBytes() < (rIndex - buf.readerIndex() + 2)) return INCOMPLETE; int value = 0; boolean negative = false; + + // Step 2: Read first byte to check for sign byte b = buf.readByte(); if (b == '-') { negative = true; } else if (b >= '0' && b <= '9') { - value = b - '0'; + value = b - '0'; // '0' is 48 in ASCII. '5' - '0' = 5. } + // Step 3: ASCII Arithmetic Loop + // Iterate until we hit the '\r' we found earlier while (buf.readerIndex() <= rIndex) { b = buf.readByte(); - if (b == '\r') break; + if (b == '\r') break; // Stop at delimiter if (b >= '0' && b <= '9') { + // Shift existing value left (x10) and add new digit value = value * 10 + (b - '0'); } } - buf.skipBytes(1); // Skip \n + + // Step 4: Skip the '\n' byte (we already processed '\r') + buf.skipBytes(1); return negative ? -value : value; } /** - * Write a RESP response to the client efficiently. - * Uses Netty's Unpooled buffer for small responses. + * Helper to write a response string back to the client. + * Uses an unpooled buffer because response strings are typically short lived. */ private void writeResponse(ChannelHandlerContext ctx, String response) { ctx.writeAndFlush(Unpooled.copiedBuffer(response, StandardCharsets.UTF_8)); @@ -177,6 +236,7 @@ private void writeResponse(ChannelHandlerContext ctx, String response) { @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + // Standard Netty error handling: log and close connection on fatal errors System.err.println("[RedisCommandHandler] Exception: " + cause.getMessage()); ctx.close(); } diff --git a/src/main/java/com/redis/storage/ExpiryManager.java b/src/main/java/com/redis/storage/ExpiryManager.java index f9cedc5..ee4f58d 100644 --- a/src/main/java/com/redis/storage/ExpiryManager.java +++ b/src/main/java/com/redis/storage/ExpiryManager.java @@ -54,8 +54,8 @@ private void runExpiryCleaner(Consumer onExpire) { if (!running) break; // Fast shutdown check - String key = task.getKey(); - long expectedExpiryTime = task.getExpiryTimeMillis(); + String key = task.key(); + long expectedExpiryTime = task.expiryTimeMillis(); // Deduplication: only process if this is still the latest expiry for this key Long currentExpiry = keyExpiryMap.get(key); diff --git a/src/main/java/com/redis/storage/RedisDatabase.java b/src/main/java/com/redis/storage/RedisDatabase.java index d7cc985..51fc076 100644 --- a/src/main/java/com/redis/storage/RedisDatabase.java +++ b/src/main/java/com/redis/storage/RedisDatabase.java @@ -5,107 +5,128 @@ import java.util.concurrent.atomic.AtomicBoolean; /** - * In-memory key-value store with expiry support using DelayQueue-based cleanup. - * This is the core data storage for the Redis server. - * - * Supports multiple value types via RedisValue wrapper: - * - STRING: Simple string values - * - LIST: Ordered list of strings - * - SET: Unordered collection of unique strings - * - HASH: Map of field-value pairs - * - SORTED_SET: Set with scores for ordering - * - * Expiry is managed by ExpiryManager (zero-polling approach with DelayQueue). + * The Central In-Memory Storage Engine. + *

+ * Role: This singleton acts as the "Heap" for the Redis server. It manages the lifecycle + * of all keys, including storage, retrieval, and expiration. + *

+ * Concurrency Model: + * Uses {@link ConcurrentHashMap} for high-throughput, thread-safe storage. + * Reads are generally lock-free. Writes are locked per-bucket. + * Atomic operations (like 'compute') allow for safe read-modify-write cycles needed for commands like INCR or APPEND. + *

+ * Expiration Strategy (Hybrid): + * 1. Lazy (Passive): Every read operation (get, exists) checks if the key is expired. If yes, it's deleted immediately. + * 2. Active: The {@link ExpiryManager} (background component) removes keys when their TTL hits zero. */ public class RedisDatabase { - private static RedisDatabase INSTANCE; + + // Singleton Instance + private static volatile RedisDatabase INSTANCE; /** - * Internal entry that wraps a RedisValue with optional expiry time. + * Internal wrapper. Bundles the data payload with its metadata (expiry). + *

+ * Why a Record? Immutable data carrier. + * Why absolute time? Storing 'expiryMillis' (Epoch) is cheaper to check than 'ttl' (Duration). + * Just compare {@code entry.expiry < System.currentTimeMillis()}. */ private record ValueEntry(RedisValue value, long expiryMillis) { } + // The core storage map. + // Key = Redis Key (String) + // Value = Wrapper containing Data + Expiry private final ConcurrentHashMap map = new ConcurrentHashMap<>(); + + // Helper component to handle background cleanup of expired keys. private final ExpiryManager expiryManager; private RedisDatabase() { + // Initialize the manager with a callback to remove keys from this map this.expiryManager = new ExpiryManager(this::removeKey); } /** - * Provide the lazily initialized, thread-safe singleton instance of RedisDatabase. - * - * @return the singleton RedisDatabase instance + * Holder Class Pattern + * */ + public static class RedisDatabaseHolder { + private static final RedisDatabase INSTANCE = new RedisDatabase(); + } + + /** + * Singleton accessor. + * @return the singleton instance */ public static RedisDatabase getInstance() { - if (INSTANCE == null) { - synchronized (RedisDatabase.class) { - if (INSTANCE == null) { - INSTANCE = new RedisDatabase(); - } - } - } - return INSTANCE; + return RedisDatabaseHolder.INSTANCE; } + // ==================== Expiry Management ==================== + /** - * Retrieve the absolute expiry time (epoch milliseconds) for the given key. + * checks the Time-To-Live (TTL) of a key. + *

+ * Lazy Expiration Logic: + * If we find the key is expired during this check, we delete it immediately. + * This prevents the user from seeing "ghost" keys that exist but are technically dead. * - * @param key the key to query - * @return -1 if the key does not exist or is expired, Long.MAX_VALUE if the key exists without expiry, otherwise the absolute expiry time in milliseconds + * @return Absolute epoch time in ms, or -1 if no expiry/not exists. */ public long getExpiryTime(String key) { var entry = map.get(key); if (entry == null) return -1; + + // LAZY CHECK: Is it dead? if (isExpired(entry)) { - map.remove(key, entry); + map.remove(key, entry); // Clean up trash return -1; } return entry.expiryMillis(); } /** - * Update the absolute expiry time for an existing key. - * - * If `expiryTimeMillis` is `Long.MAX_VALUE` the key's expiry is cleared (made persistent). - * If the key does not exist or is already expired the entry is removed and no update is performed. - * - * @param key the key whose expiry to update - * @param expiryTimeMillis the new absolute expiry time in milliseconds since epoch, or `Long.MAX_VALUE` to remove expiry - * @return `true` if an existing non-expired key's expiry was updated, `false` if the key did not exist or was expired + * Updates the expiry of an existing key. + *

+ * Atomicity: Uses `computeIfPresent` to ensure we don't resurrect a key + * that was deleted by another thread milliseconds ago. + * * @param expiryTimeMillis Absolute Epoch time. Long.MAX_VALUE means "Persistent" (No expiry). */ public boolean setExpiryTime(String key, long expiryTimeMillis) { AtomicBoolean updated = new AtomicBoolean(false); + map.computeIfPresent(key, (k, existing) -> { + // Edge case: It expired just before we got the lock if (isExpired(existing)) return null; + updated.set(true); + + // Notify the background manager if (expiryTimeMillis == Long.MAX_VALUE) { - expiryManager.clearExpiry(key); + expiryManager.clearExpiry(key); // Remove from delay queue } else { - expiryManager.scheduleExpiry(key, expiryTimeMillis); + expiryManager.scheduleExpiry(key, expiryTimeMillis); // Add/Update delay queue } + + // Return a new entry with updated time but same value return new ValueEntry(existing.value(), expiryTimeMillis); }); + return updated.get(); } - // ==================== Generic RedisValue Methods ==================== + // ==================== Storage Operations ==================== /** - * Store a RedisValue under the given key with no expiry. - * - * Any existing expiry associated with the key is cleared. - * - * @param key the key to store the value under - * @param value the value to store - */ + * Basic "SET" operation. + * Clears any previous expiry because a generic SET removes the TTL in Redis protocol. + */ public void put(String key, RedisValue value) { map.put(key, new ValueEntry(value, Long.MAX_VALUE)); expiryManager.clearExpiry(key); } /** - * Store a RedisValue with a time-to-live in milliseconds. + * "SETEX" operation (Set with Expiry). */ public void put(String key, RedisValue value, long ttlMillis) { if (ttlMillis <= 0) { @@ -119,24 +140,33 @@ public void put(String key, RedisValue value, long ttlMillis) { } /** - * Retrieve a RedisValue. Returns null if key doesn't exist or has expired. + * Core "GET" operation. + *

+ * Crucial: This is the primary point of Lazy Expiration. + * Every generic read goes through here. */ public RedisValue getValue(String key) { var entry = map.get(key); if (entry == null) return null; + if (isExpired(entry)) { - map.remove(key, entry); + map.remove(key, entry); // Lazy Delete return null; } return entry.value(); } /** - * Get value with expected type. Returns null if key doesn't exist, expired, or type mismatch. + * Type-Safe Retrieval. + * Used by specific commands (e.g., LPUSH needs a List, not a String). + * + * @param expectedType The internal enum type we expect (LIST, HASH, etc.) + * @return The raw Java object (List, Map) cast to T, or null on mismatch. */ @SuppressWarnings("unchecked") public T getTyped(String key, RedisValue.Type expectedType) { RedisValue value = getValue(key); + // Validates existence and type compatibility if (value == null || value.getType() != expectedType) { return null; } @@ -144,35 +174,24 @@ public T getTyped(String key, RedisValue.Type expectedType) { } /** - * Get the type of a key's value. Returns null if key doesn't exist. + * Lightweight check for "TYPE" command. */ public RedisValue.Type getType(String key) { RedisValue value = getValue(key); return value != null ? value.getType() : null; } - // ==================== String Convenience Methods (Backward Compatible) ==================== + // ==================== String Helpers ==================== + // Convenience wrappers to avoid manually creating RedisValue.StringValue every time. - /** - * Store a string value without expiry. - * Convenience method for STRING type. - */ public void put(String key, String value) { put(key, RedisValue.string(value)); } - /** - * Store a string value with TTL. - * Convenience method for STRING type. - */ public void put(String key, String value, long ttlMillis) { put(key, RedisValue.string(value), ttlMillis); } - /** - * Retrieve a string value. Returns null if key doesn't exist, expired, or not a STRING. - * Convenience method for STRING type. - */ public String get(String key) { RedisValue value = getValue(key); if (value == null || value.getType() != RedisValue.Type.STRING) { @@ -181,11 +200,8 @@ public String get(String key) { return value.asString(); } - // ==================== Key Operations ==================== + // ==================== Key Management ==================== - /** - * Check if a key exists (not expired). - */ public boolean exists(String key) { var entry = map.get(key); if (entry == null) return false; @@ -196,23 +212,15 @@ public boolean exists(String key) { return true; } - /** - * Remove the mapping for the given key and cancel any scheduled expiry. - * - * @param key the key to remove - * @return `true` if a mapping was removed, `false` otherwise - */ public boolean remove(String key) { + // remove() returns the previous value or null. boolean removed = map.remove(key) != null; if (removed) { - expiryManager.clearExpiry(key); + expiryManager.clearExpiry(key); // Remember to clean up the background task! } return removed; } - /** - * Remove multiple keys and return the count of removed keys. - */ public int removeAll(Collection keys) { int count = 0; for (String k : keys) { @@ -221,27 +229,30 @@ public int removeAll(Collection keys) { return count; } - // ==================== Utility Methods ==================== + // ==================== Internal Logic ==================== + /** + * Checks if the entry has passed its expiration timestamp. + * returns false if expiry is Long.MAX_VALUE (persistent). + */ private boolean isExpired(ValueEntry entry) { return entry.expiryMillis() != Long.MAX_VALUE && - entry.expiryMillis() <= System.currentTimeMillis(); + entry.expiryMillis() <= System.currentTimeMillis(); } + /** + * Callback used by ExpiryManager to physically remove the key. + * Unlike remove(), this doesn't need to call clearExpiry() because + * it was triggered BY the expiry manager. + */ private void removeKey(String key) { map.remove(key); } - /** - * Get the number of keys in the database. - */ public int size() { return map.size(); } - /** - * Gracefully shutdown the database and its expiry manager. - */ public void shutdown() { expiryManager.shutdown(); } @@ -249,36 +260,42 @@ public void shutdown() { // ==================== Atomic Operations ==================== /** - * Compute and install a new value for a key using the provided remapping function, preserving TTL for unexpired entries. - * - * The remapping function is invoked with the current value for the key, or `null` if the key is absent or expired. - * If the function returns `null`, the key is removed. If it returns a non-null value, that value is stored; - * an existing unexpired entry's expiry is preserved, otherwise the new entry has no expiry. - * The operation is performed atomically. - * - * @param key the key to compute - * @param remappingFunction function that receives the current `RedisValue` (or `null`) and returns the new `RedisValue`, or `null` to remove the key + * The "Holy Grail" of Thread-Safe Read-Modify-Write. + *

+ * Why use this? + * If you want to append a string, you cannot do: + * 

+     * val = get(k);
+     * put(k, val + "new");
+     *
+ * Between the get and put, another thread might have changed the value. + *

+ * How it works: + * ConcurrentHashMap locks the specific key bucket. It feeds the current value + * to your function, and atomically updates the map with your result. + * No other thread can touch this key until the function finishes. */ public void compute(String key, java.util.function.Function remappingFunction) { map.compute(key, (k, existingEntry) -> { - // Check if entry exists and is not expired + // 1. Validate existing data (Lazy Expiry check inside the lock) boolean validEntry = existingEntry != null && !isExpired(existingEntry); RedisValue currentValue = validEntry ? existingEntry.value() : null; - // Apply remapping function + // 2. Run the user's logic RedisValue newValue = remappingFunction.apply(currentValue); - // If function returns null, remove the key + // 3. Handle deletion if (newValue == null) { - return null; + return null; // Signals map.compute to remove the key } - // Micro-optimization: avoid new ValueEntry if value hasn't changed + // Optimization: If nothing changed, don't create new objects if (newValue == currentValue && validEntry) { return existingEntry; } - // Preserve expiry for existing non-expired entries, otherwise no expiry + // 4. Preserve TTL + // If the key existed, keep its old expiry. If it's new, it has no expiry. long expiry = validEntry ? existingEntry.expiryMillis() : Long.MAX_VALUE; return new ValueEntry(newValue, expiry); diff --git a/src/main/java/com/redis/storage/RedisValue.java b/src/main/java/com/redis/storage/RedisValue.java index 8400050..f409127 100644 --- a/src/main/java/com/redis/storage/RedisValue.java +++ b/src/main/java/com/redis/storage/RedisValue.java @@ -11,87 +11,76 @@ import java.util.concurrent.ConcurrentSkipListMap; /** - * Wrapper class for Redis values supporting multiple data types. - * Using Java 25 sealed interface and records for high scalability and type safety. + * The unified container for all Redis data types. + *

+ * Architecture: Sealed Interface + Records + * This design uses Java's modern type system to strictly define what a "Value" can be. + * By sealing the interface, we guarantee that the compiler knows exactly which 6 types exist, + * enabling exhaustive pattern matching and preventing invalid data types from entering the system. + *

+ * Thread Safety: + * While the 'Record' wrapper is immutable (you can't swap the underlying List reference), + * the collections inside (ArrayList, HashMap) are wrapped in thread-safe implementations + * (ConcurrentHashMap, SynchronizedList) to support concurrent access. */ -public sealed interface RedisValue permits - RedisValue.StringValue, - RedisValue.ListValue, - RedisValue.SetValue, - RedisValue.HashValue, - RedisValue.SortedSetValue, - RedisValue.StreamValue { +public sealed interface RedisValue permits RedisValue.StringValue, RedisValue.ListValue, RedisValue.SetValue, RedisValue.HashValue, RedisValue.SortedSetValue, RedisValue.StreamValue { /** - * Redis data types. + * Enumeration of supported Redis data types. + * Used for the "TYPE" command and error reporting. */ enum Type { - STRING, - LIST, - SET, - HASH, - SORTED_SET, - STREAM + STRING, LIST, SET, HASH, SORTED_SET, STREAM } /** - * Get the type of this value. + * @return The runtime type of the stored value. */ Type getType(); /** - * Get the raw data object. + * @return The raw underlying Java object (e.g., String, List, Map). + * Useful for generic serialization or debugging. */ Object getData(); // ==================== Factory Methods ==================== + // These static factories provide a clean API to create values without exposing + // the specific Record constructors directly. - /** - * Create a STRING value. - */ static RedisValue string(String value) { return new StringValue(value); } - /** - * Create a LIST value. - */ static RedisValue list(List value) { return new ListValue(value); } - /** - * Create a SET value. - */ static RedisValue set(Set value) { return new SetValue(value); } - /** - * Create a HASH value. - */ static RedisValue hash(Map value) { return new HashValue(value); } - /** - * Create a SORTED_SET value. - */ static RedisValue sortedSet(Map value) { return new SortedSetValue(value); } - /** - * Create a STREAM value. - */ static RedisValue stream(Map> value) { return new StreamValue(value); } // ==================== Type-Safe Accessors ==================== + // These methods implement the strict "WRONGTYPE" checking required by the Redis Protocol. + // If a user tries to run a List command on a String value, we must throw an exception. /** - * Get value as String. Throws if type mismatch. + * Extracts the String value. + * Uses Java Pattern Matching to simultaneously check type and cast. + * + * @throws IllegalStateException if the value is not a String. */ default String asString() { if (this instanceof StringValue(String value)) { @@ -101,7 +90,7 @@ default String asString() { } /** - * Get value as List. Throws if type mismatch. + * Extracts the List value. */ default List asList() { if (this instanceof ListValue(List list)) { @@ -111,7 +100,7 @@ default List asList() { } /** - * Get value as Set. Throws if type mismatch. + * Extracts the Set value. */ default Set asSet() { if (this instanceof SetValue(Set set)) { @@ -121,7 +110,7 @@ default Set asSet() { } /** - * Get value as Hash (Map). Throws if type mismatch. + * Extracts the Hash (Map) value. */ default Map asHash() { if (this instanceof HashValue(Map hash)) { @@ -131,7 +120,7 @@ default Map asHash() { } /** - * Get value as Sorted Set (Map). Throws if type mismatch. + * Extracts the Sorted Set (Map: Member -> Score). */ default Map asSortedSet() { if (this instanceof SortedSetValue(Map sortedSet)) { @@ -141,7 +130,7 @@ default Map asSortedSet() { } /** - * Get value as Stream (Map). Throws if type mismatch. + * Extracts the Stream (Map: StreamId -> Entry). */ default Map> asStream() { if (this instanceof StreamValue(Map> stream)) { @@ -151,66 +140,173 @@ default Map> asStream() { } /** - * Check if this value is of the specified type. + * Utility to check type equality. */ default boolean isType(Type expectedType) { return getType() == expectedType; } // ==================== Implementation Records ==================== + // Records serve as immutable containers for the mutable data structures. + // The constructor logic ensures that whatever list/map is passed in gets + // converted to the correct Thread-Safe implementation. + /** + * Storage for Redis Strings. + * Note: Redis Strings are binary safe, but here we use Java String (UTF-16) for simplicity. + */ record StringValue(String value) implements RedisValue { - @Override public Type getType() { return Type.STRING; } - @Override public Object getData() { return value; } - @Override public String toString() { return "RedisValue{type=STRING, data=" + value + "}"; } + @Override + public Type getType() { + return Type.STRING; + } + + @Override + public Object getData() { + return value; + } + + @Override + public String toString() { + return "RedisValue{type=STRING, data=" + value + "}"; + } } + /** + * Storage for Redis Lists (Linked Lists). + * Uses synchronizedList to ensure thread safety for simple operations. + */ record ListValue(List list) implements RedisValue { public ListValue { + // Defensiveness: Copy the input to a new ArrayList to detach from original source, + // then wrap in synchronizedList for thread safety. list = Collections.synchronizedList(new ArrayList<>(list)); } - @Override public Type getType() { return Type.LIST; } - @Override public Object getData() { return list; } - @Override public String toString() { return "RedisValue{type=LIST, data=" + list + "}"; } + + @Override + public Type getType() { + return Type.LIST; + } + + @Override + public Object getData() { + return list; + } + + @Override + public String toString() { + return "RedisValue{type=LIST, data=" + list + "}"; + } } + /** + * Storage for Redis Sets (Unordered, Unique). + * Uses ConcurrentHashMap.newKeySet() which is a thread-safe Set backed by a ConcurrentHashMap. + */ record SetValue(Set set) implements RedisValue { public SetValue { + // Defensiveness: Create a fresh thread-safe Set and populate it. Set newSet = ConcurrentHashMap.newKeySet(); newSet.addAll(set); set = newSet; } - @Override public Type getType() { return Type.SET; } - @Override public Object getData() { return set; } - @Override public String toString() { return "RedisValue{type=SET, data=" + set + "}"; } + + @Override + public Type getType() { + return Type.SET; + } + + @Override + public Object getData() { + return set; + } + + @Override + public String toString() { + return "RedisValue{type=SET, data=" + set + "}"; + } } + /** + * Storage for Redis Hashes (Field-Value pairs). + * Uses ConcurrentHashMap for high-concurrency read/write access. + */ record HashValue(Map hash) implements RedisValue { public HashValue { hash = new ConcurrentHashMap<>(hash); } - @Override public Type getType() { return Type.HASH; } - @Override public Object getData() { return hash; } - @Override public String toString() { return "RedisValue{type=HASH, data=" + hash + "}"; } + + @Override + public Type getType() { + return Type.HASH; + } + + @Override + public Object getData() { + return hash; + } + + @Override + public String toString() { + return "RedisValue{type=HASH, data=" + hash + "}"; + } } + /** + * Storage for Redis Sorted Sets (ZSET). + * Mapped as Member -> Score. + * Note: Real Redis ZSETs use a dual structure (SkipList + HashMap). + * This simple Map implementation is O(N) for range queries but O(1) for lookups. + */ record SortedSetValue(Map sortedSet) implements RedisValue { public SortedSetValue { sortedSet = new ConcurrentHashMap<>(sortedSet); } - @Override public Type getType() { return Type.SORTED_SET; } - @Override public Object getData() { return sortedSet; } - @Override public String toString() { return "RedisValue{type=SORTED_SET, data=" + sortedSet + "}"; } + + @Override + public Type getType() { + return Type.SORTED_SET; + } + + @Override + public Object getData() { + return sortedSet; + } + + @Override + public String toString() { + return "RedisValue{type=SORTED_SET, data=" + sortedSet + "}"; + } } + /** + * Storage for Redis Streams. + * Uses ConcurrentSkipListMap because Streams require: + * 1. Ordering (by StreamId) + * 2. Thread Safety + * SkipListMap provides O(log n) access and keeps keys sorted naturally. + */ record StreamValue(Map> stream) implements RedisValue { public StreamValue { + // Ensure we use the sorted, thread-safe map implementation if (!(stream instanceof ConcurrentSkipListMap)) { stream = new ConcurrentSkipListMap<>(stream); } } - @Override public Type getType() { return Type.STREAM; } - @Override public Object getData() { return stream; } - @Override public String toString() { return "RedisValue{type=STREAM, data=" + stream + "}"; } + + @Override + public Type getType() { + return Type.STREAM; + } + + @Override + public Object getData() { + return stream; + } + + @Override + public String toString() { + return "RedisValue{type=STREAM, data=" + stream + "}"; + } } -} +}; \ No newline at end of file diff --git a/src/main/java/com/redis/util/ExpiryTask.java b/src/main/java/com/redis/util/ExpiryTask.java index 7ad1071..2c80580 100644 --- a/src/main/java/com/redis/util/ExpiryTask.java +++ b/src/main/java/com/redis/util/ExpiryTask.java @@ -7,22 +7,7 @@ * A Delayed task representing a key expiration scheduled for a specific time. * Used by ExpiryManager to efficiently schedule and execute expiry cleanups. */ -public class ExpiryTask implements Delayed { - private final String key; - private final long expiryTimeMillis; - - public ExpiryTask(String key, long expiryTimeMillis) { - this.key = key; - this.expiryTimeMillis = expiryTimeMillis; - } - - public String getKey() { - return key; - } - - public long getExpiryTimeMillis() { - return expiryTimeMillis; - } +public record ExpiryTask(String key, long expiryTimeMillis) implements Delayed { @Override public long getDelay(TimeUnit unit) { @@ -39,10 +24,6 @@ public int compareTo(Delayed other) { @Override public String toString() { - return "ExpiryTask{" + - "key='" + key + '\'' + - ", expiryTimeMillis=" + expiryTimeMillis + - ", delayMs=" + getDelay(TimeUnit.MILLISECONDS) + - '}'; + return "ExpiryTask{" + "key='" + key + '\'' + ", expiryTimeMillis=" + expiryTimeMillis + ", delayMs=" + getDelay(TimeUnit.MILLISECONDS) + '}'; } } diff --git a/src/main/java/com/redis/util/StreamId.java b/src/main/java/com/redis/util/StreamId.java index 81bac06..8af10eb 100644 --- a/src/main/java/com/redis/util/StreamId.java +++ b/src/main/java/com/redis/util/StreamId.java @@ -3,21 +3,50 @@ import java.util.Objects; /** - * Represents a Redis Stream ID (milliseconds-sequence). + * Value Object representing a Redis Stream ID. + *

+ * Format: {@code -} + *

+ * Role in Architecture: + * This class serves as the Key in the {@code ConcurrentSkipListMap} backing the Stream data structure. + * It implements {@code Comparable} to ensure that the Map keeps messages strictly ordered by time. + *

+ * Why Record? + * As a Map Key, this object MUST be immutable. Records provide immutability, + * equals(), and hashCode() out of the box, preventing subtle hashing bugs. */ public record StreamId(long time, long sequence) implements Comparable { + // Sentinel values useful for Range Queries (XRANGE) public static final StreamId MIN = new StreamId(0, 0); public static final StreamId MAX = new StreamId(Long.MAX_VALUE, Long.MAX_VALUE); + /** + * Parses a string ID into a StreamId object. + *

+ * Supported Formats: + *

    + *
  • {@code "123-456"} -> Time: 123, Seq: 456
    • + *
  • {@code "123"} -> Time: 123, Seq: 0 (Implicit sequence 0)
    • + *
+ * + * @param id The string representation of the ID. + * @return The parsed immutable StreamId. + * @throws IllegalArgumentException if format is invalid. + */ public static StreamId parse(String id) { if (id == null || id.isEmpty()) { throw new IllegalArgumentException("Empty stream ID"); } + + // Edge case: "0" usually implies 0-0 if (id.equals("0")) { return MIN; } + String[] parts = id.split("-"); + + // Case 1: "123" (Time only provided) if (parts.length == 1) { try { return new StreamId(Long.parseLong(parts[0]), 0); @@ -25,9 +54,12 @@ public static StreamId parse(String id) { throw new IllegalArgumentException("Stream ID must be a number or