support new protocol

1 files changed, 188 insertions, 29 deletions

diff --git a/src/Client.php b/src/Client.php
index 5d0048b..2d4b6e1 100644
--- a/src/Client.php
+++ b/src/Client.php
@@ -9,11 +9,14 @@ class Client {
     const MASTER_PORT = 7081;
 
     const EOT = "\4";
+    const REQUEST_NO_LIMIT = 999999;
 
     protected $host;
     protected $port;
     protected $password;
     protected $sock;
+    protected $passwordSent = false;
+    protected $lastOutgoingRequestNo = null;
 
     /**
      * JobdClient constructor.
@@ -30,6 +33,9 @@ class Client {
         $this->sock = fsockopen($this->host, $this->port);
         if (!$this->sock)
             throw new \Exception("Failed to connect to {$this->host}:{$this->port}");
+
+        // 0 is reserved
+        $this->lastOutgoingRequestNo = mt_rand(1, self::REQUEST_NO_LIMIT);
     }
 
     /**
@@ -37,7 +43,7 @@ class Client {
      * @throws \Exception
      */
     public function ping() {
-        $this->send(new RequestMessage('ping'));
+        $this->send(new PingMessage());
         return $this->recv();
     }
 
@@ -47,8 +53,9 @@ class Client {
      * @throws \Exception
      */
     public function poke(array $targets) {
-        $this->send(new RequestMessage('poke', ['targets' => $targets]));
-        return $this->recv();
+        return $this->recv(
+            $this->sendRequest(new RequestMessage('poke', ['targets' => $targets]))
+        );
     }
 
     /**
@@ -56,8 +63,9 @@ class Client {
      * @throws \Exception
      */
     public function status() {
-        $this->send(new RequestMessage('status'));
-        return $this->recv();
+        return $this->recv(
+            $this->sendRequest(new RequestMessage('status'))
+        );
     }
 
     /**
@@ -66,8 +74,9 @@ class Client {
      * @throws \Exception
      */
     public function poll(array $targets) {
-        $this->send(new RequestMessage('poll', ['targets' => $targets]));
-        return $this->recv();
+        return $this->recv(
+            $this->sendRequest(new RequestMessage('poll', ['targets' => $targets]))
+        );
     }
 
     /**
@@ -75,27 +84,43 @@ class Client {
      * @return ResponseMessage
      */
     public function runManual(int $id) {
-        $this->send(new RequestMessage('run-manual', ['id' => $id]));
-        return $this->recv();
+        return $this->recv(
+            $this->sendRequest(new RequestMessage('run-manual', ['id' => $id]))
+        );
     }
 
     /**
      * @param RequestMessage $request
+     * @return int
      */
-    public function send(RequestMessage $request) {
-        if ($this->password)
+    public function sendRequest(RequestMessage $request) {
+        if ($this->password && !$this->passwordSent) {
             $request->setPassword($this->password);
+            $this->passwordSent = true;
+        }
+
+        $no = $this->getNextOutgoingRequestNo();
+        $request->setRequestNo($no);
 
-        $serialized = $request->serialize();
+        $this->send($request);
 
+        return $no;
+    }
+
+    /**
+     * @param Message $message
+     */
+    protected function send(Message $message) {
+        $serialized = $message->serialize();
         fwrite($this->sock, $serialized . self::EOT);
     }
 
     /**
+     * @param int $request_no
      * @return ResponseMessage
      * @throws \Exception
      */
-    public function recv() {
+    public function recv(int $request_no = -1) {
         $messages = [];
         $buf = '';
         while (!feof($this->sock)) {
@@ -117,51 +142,128 @@ class Client {
             }
         } while ($eot_pos !== false && $offset < $buflen-1);
 
-        if (empty($message))
+        if (empty($messages))
             throw new \Exception("Malformed response: no messages found. Response: {$buf}");
 
         if (count($messages) > 1)
             trigger_error(__METHOD__.": received more than one message");
 
-        $response = self::parseMessage($messages[0]);
-        if (!($response instanceof ResponseMessage))
-            throw new \Exception('Unexpected message type');
+        $response = null;
+        $messages = array_map('self::parseMessage', $messages);
+        if ($request_no != -1) {
+            /**
+             * @var ResponseMessage[] $messages
+             */
+            $messages = array_filter(
+                $messages,
+
+                /**
+                 * @param ResponseMessage|RequestMessage $message
+                 */
+                function(Message $message) use ($request_no) {
+                    return $message instanceof ResponseMessage
+                        && ($message->getRequestNo() === $request_no || $message->getRequestNo() === 0);
+                }
+            );
+
+            if (empty($messages))
+                throw new \Exception("Malformed response: response for {$request_no} not found.");
+
+
+            if (count($messages) == 2) {
+                // weird, we caught response for our $request_no AND a message with reserved zero no
+                // but anyway
+
+                for ($i = 0; $i < count($messages); $i++) {
+                    $message = $messages[$i];
+
+                    if ($message->getRequestNo() == $request_no)
+                        $response = $message;
+
+                    else if ($message->getRequestNo() == 0)
+                        trigger_error(__METHOD__.': received an error with reqno=0: '.($message->getError() ?? null));
+                }
+            }
+        }
 
-        if ($error = $response->getError())
-            throw new \Exception('jobd error: '.$response->getError());
+        if (is_null($response))
+            $response = $messages[0];
+
+        if ($response instanceof ResponseMessage) {
+            if ($error = $response->getError())
+                throw new \Exception($response->getError());
+        }
 
         return $response;
     }
 
     /**
+     * @return int
+     */
+    protected function getNextOutgoingRequestNo() {
+        $this->lastOutgoingRequestNo++;
+
+        if ($this->lastOutgoingRequestNo >= self::REQUEST_NO_LIMIT)
+            $this->lastOutgoingRequestNo = 1; // 0 is reserved
+
+        return $this->lastOutgoingRequestNo;
+    }
+
+    /**
      * @param string $raw_string
      * @return RequestMessage|ResponseMessage
      * @throws \Exception
      */
     protected static function parseMessage(string $raw_string) {
         $raw = json_decode($raw_string, true);
-        if (!is_array($raw) || count($raw) != 2)
+        if (!is_array($raw) || count($raw) < 1)
             throw new \Exception("Malformed message: {$raw_string}");
 
-        list($type, $data) = $raw;
+        list($type) = $raw;
 
         switch ($type) {
             case Message::REQUEST:
-                if (!$data || !is_array($data) || !isset($data['type']) || !is_string($data['type']))
-                    throw new \Exception('Malformed REQUEST message');
-
-                $message = new RequestMessage($data['type'], $data['data'] ?? null);
+                $data = $raw[1];
+                try {
+                    self::validateData($data, [
+                        // name      type     required
+                        ['type',     's',     true],
+                        ['no',       'i',     true],
+                        ['password', 's',     false],
+                        ['data',     'aifs',  false]
+                    ]);
+                } catch (\Exception $e) {
+                    throw new \Exception("Malformed REQUEST message: {$e->getMessage()}");
+                }
+
+                $message = new RequestMessage($data['no'], $data['type'], $data['data'] ?? null);
                 if (isset($data['password']))
                     $message->setPassword($data['password']);
 
                 return $message;
 
             case Message::RESPONSE:
-                if (!is_array($data) || count($data) < 2)
-                    throw new \Exception('Malformed RESPONSE message');
+                $data = $raw[1];
+                try {
+                    self::validateData($data, [
+                        // name   type     required
+                        ['no',    'i',     true],
+                        ['data',  'aifs',  false],
+                        ['error', 's',     false],
+                    ]);
+                } catch (\Exception $e) {
+                    throw new \Exception("Malformed RESPONSE message: {$e->getMessage()}");
+                }
+
+                return new ResponseMessage($data['no'], $data['error'] ?? null, $data['data'] ?? null);
+
+            case Message::PING:
+                return new PingMessage();
+                break;
 
-                $message = new ResponseMessage(...$data);
-                return $message;
+            case Message::PONG:
+                return new PongMessage();
+                break;
 
             default:
                 throw new \Exception("Malformed message: unexpected type {$type}");
@@ -169,6 +271,63 @@ class Client {
     }
 
     /**
+     * @param mixed $data
+     * @param array $schema
+     * @return bool
+     */
+    protected static function validateData($data, array $schema) {
+        if (!$data || !is_array($data))
+            throw new \Exception('data must be array');
+
+        foreach ($schema as $schema_item) {
+            list ($key_name, $key_types, $key_required) = $schema_item;
+            if (!isset($data[$key_name])) {
+                if ($key_required)
+                    throw new \Exception("'{$key_name}' is missing");
+
+                continue;
+            }
+
+            $passed = false;
+            for ($i = 0; $i < strlen($key_types); $i++) {
+                $type = $key_types[$i];
+
+                switch ($type) {
+                    case 'i':
+                        if (is_int($data[$key_name]))
+                            $passed = true;
+                        break;
+
+                    case 'f':
+                        if (is_float($data[$key_name]))
+                            $passed = true;
+                        break;
+
+                    case 's':
+                        if (is_string($data[$key_name]))
+                            $passed = true;
+                        break;
+
+                    case 'a':
+                        if (is_array($data[$key_name]))
+                            $passed = true;
+                        break;
+
+                    default:
+                        trigger_error(__METHOD__.': unexpected type '.$type);
+                        break;
+                }
+
+                if ($passed)
+                    break;
+            }
+
+            if (!$passed)
+                throw new \Exception("{$key_name}: required type is '{$key_types}'");
+        }
+    }
+
+    /**
      * @return bool
      */
     public function close() {