socket = @stream_socket_client("tcp://{$host}:{$port}", $errno, $errstr, $timeout); if (!$this->socket) { // Log error silently or throw depending on preference throw new Exception("Redis connection failed: $errstr"); } } public function setnxEx($key, $value, $seconds) { // RESP Command: SET key value NX EX seconds return $this->execute('SET', [$key, $value, 'NX', 'EX', $seconds]); } /** * Push payload to a list (Queue) */ public function lpush($queue, $payload) { return $this->execute('LPUSH', [$queue, $payload]); } /** * Blocking Pop (Waits for data) * Used by the worker */ public function blpop($queue, $timeout = 0) { return $this->execute('BLPOP', [$queue, $timeout]); } public function lpop($queue) { return $this->execute('LPOP', [$queue]); } /** * Execute Raw RESP Command */ private function execute($command, array $args) { // Construct RESP Protocol // Format: *\r\n$\r\n\r\n... $cmd = '*' . (count($args) + 1) . "\r\n"; $cmd .= '$' . strlen($command) . "\r\n" . $command . "\r\n"; foreach ($args as $arg) { $cmd .= '$' . strlen($arg) . "\r\n" . $arg . "\r\n"; } // Write to socket fwrite($this->socket, $cmd); // Read Response (Simple implementation for success/fail) // For LPUSH, Redis returns an integer (list length). // For BLPOP, it returns an array. $line = fgets($this->socket); if(!$line) return null; $type = $line[0]; $val = substr($line, 1); // Handle Errors if ($type === '-') { throw new Exception("Redis Error: " . trim($val)); } // Handle Integers (LPUSH returns this) if ($type === ':') { return (int) $val; } // Handle Bulk Strings (BLPOP returns arrays of bulk strings) // Note: For a robust worker, we need a recursive parser, but // for BLPOP specifically, we can cheat slightly or write a parser. // Let's assume BLPOP for now returns the raw response we parse in the worker // or we implement a simple bulk reader here. return $this->parseResponse($type, $val); } private function parseResponse($type, $val) { if ($type === '$') { $len = (int) $val; // RESP null value (Queue is empty) if ($len === -1) { return null; } // Read the data ($len bytes) $buffer = ''; $bytesRead = 0; while ($bytesRead < $len) { $chunk = fread($this->socket, $len - $bytesRead); if ($chunk === false || $chunk === '') { throw new Exception("Redis Connection Lost mid-read"); } $buffer .= $chunk; $bytesRead += strlen($chunk); } // Consume the trailing \r\n that follows the data fgets($this->socket); return $buffer; } if ($type === '*') { // Array (BLPOP returns this) $count = (int) $val; $data = []; for ($i = 0; $i < $count; $i++) { // 1. Read the length of the next string $l = fgets($this->socket); $len = (int) substr($l, 1); // 2. ROBUST READ: Loop until we have all $len bytes $buffer = ''; $bytesRead = 0; while ($bytesRead < $len) { $chunk = fread($this->socket, $len - $bytesRead); if ($chunk === false || $chunk === '') { throw new Exception("Redis Connection Lost mid-read"); } $buffer .= $chunk; $bytesRead += strlen($chunk); } $data[] = $buffer; // 3. Consume the trailing \r\n fgets($this->socket); } return $data; } return $val; } }