Improved response handling module

src/danog/MadelineProto/API.php

@@ -21,6 +21,9 @@ class API
         set_error_handler(['\danog\MadelineProto\Exception', 'ExceptionErrorHandler']);
         $this->session = new MTProto($params);
         $future_salts = $this->get_future_salts(3);
+        var_dump($future_salts);
+        $future_salts = $this->get_future_salts(3);
+        var_dump($future_salts);
     }
 
     public function __destruct()

src/danog/MadelineProto/MTProtoTools/CallHandler.php

@@ -17,22 +17,26 @@ namespace danog\MadelineProto\MTProtoTools;
  */
 class CallHandler extends AuthKeyHandler
 {
-    public function wait_for_response($res_id)
+    public function wait_for_response($last_sent)
     {
         $response = null;
         $count = 0;
         while ($response == null && $count++ < $this->settings['max_tries']['response']) {
             $server_answer = $this->recv_message();
             $deserialized = $this->tl->deserialize(\danog\MadelineProto\Tools::fopen_and_write('php://memory', 'rw+b', $server_answer));
-            $tempres = $this->handle_message($deserialized);
+            end($this->incoming_messages);
+            $tempres = $this->handle_message($deserialized, $last_sent, key($this->incoming_messages));
             if ($tempres == null) {
-                if (isset($this->outgoing_messages[$res_id]['response'])) {
+                if (isset($this->outgoing_messages[$last_sent]['response']) && isset($this->incoming_messages[$this->outgoing_messages[$last_sent]['response']]['content'])) {
-                    $response = $this->outgoing_messages[$res_id]['response'];
+                    $response = $this->incoming_messages[$this->outgoing_messages[$last_sent]['response']]['content'];
                 }
             } else {
                 $response = $tempres;
             }
         }
+        if ($response == null) {
+            throw new Exception("Couldn't get response");
+        }
         switch ($response['_']) {
             case 'rpc_error':
                 throw new Exception('Got rpc error '.$response['error_code'].': '.$response['error_message']);
@@ -48,8 +52,7 @@ class CallHandler extends AuthKeyHandler
         foreach (range(1, $this->settings['max_tries']['query']) as $i) {
             try {
                 $int_message_id = $this->send_message($this->tl->serialize_method($method, $args), $this->tl->content_related($method));
-                $this->outgoing_messages[$int_message_id]['method'] = $method;
+                $this->outgoing_messages[$int_message_id]['content'] = ['method' => $method, 'args' => $args ];
-                $this->outgoing_messages[$int_message_id]['args'] = $args;
                 $server_answer = $this->wait_for_response($int_message_id);
             } catch (Exception $e) {
                 $this->log->log('An error occurred while calling method '.$method.': '.$e->getMessage().' in '.$e->getFile().':'.$e->getLine().'. Recreating connection and retrying to call method...');
@@ -72,8 +75,7 @@ class CallHandler extends AuthKeyHandler
             try {
                 $int_message_id = $this->send_message($this->tl->serialize_obj($object, $args), $this->tl->content_related($object));
 //                $server_answer = $this->recv_message();
-                $this->outgoing_messages[$int_message_id]['method'] = $object;
+                $this->outgoing_messages[$int_message_id]['content'] = ['object' => $object, 'args' => $args ];
-                $this->outgoing_messages[$int_message_id]['args'] = $args;
             } catch (Exception $e) {
                 $this->log->log('An error occurred while calling object '.$object.': '.$e->getMessage().' in '.$e->getFile().':'.$e->getLine().'. Recreating connection and retrying to call object...');
                 unset($this->sock);

src/danog/MadelineProto/MTProtoTools/CallHandler.php.save

@@ -0,0 +1,97 @@
+<?php
+/*
+Copyright 2016 Daniil Gentili
+(https://daniil.it)
+This file is part of MadelineProto.
+MadelineProto is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.
+MadelineProto is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+See the GNU Affero General Public License for more details.
+You should have received a copy of the GNU General Public License along with the MadelineProto.
+If not, see <http://www.gnu.org/licenses/>.
+*/
+
+namespace danog\MadelineProto\MTProtoTools;
+
+/**
+ * Manages method and object calls.
+ */
+class CallHandler extends AuthKeyHandler
+{
+    public function wait_for_response($last_sent)
+    {
+        $response = null;
+        $count = 0;
+        while ($response == null && $count++ < $this->settings['max_tries']['response']) {
+            $server_answer = $this->recv_message();
+            $deserialized = $this->tl->deserialize(\danog\MadelineProto\Tools::fopen_and_write('php://memory', 'rw+b', $server_answer));
+
+            $tempres = $this->handle_message($deserialized, $last_sent, );
+            if ($tempres == null) {
+                if (isset($this->outgoing_messages[$last_sent]['response'])) {
+                    $response = $this->outgoing_messages[$last_sent]['response'];
+                }
+            } else {
+                $response = $tempres;
+            }
+        }
+        if ($response == null) {
+            throw new Excpetion("Couldn't get response");
+        }
+        switch ($response['_']) {
+            case 'rpc_error':
+                throw new Exception('Got rpc error '.$response['error_code'].': '.$response['error_message']);
+                break;
+            default:
+                return $response;
+                break;
+        }
+    }
+
+    public function method_call($method, $args)
+    {
+        foreach (range(1, $this->settings['max_tries']['query']) as $i) {
+            try {
+                $int_message_id = $this->send_message($this->tl->serialize_method($method, $args), $this->tl->content_related($method));
+                $this->outgoing_messages[$int_message_id]['method'] = $method;
+                $this->outgoing_messages[$int_message_id]['args'] = $args;
+                $server_answer = $this->wait_for_response($int_message_id);
+            } catch (Exception $e) {
+                $this->log->log('An error occurred while calling method '.$method.': '.$e->getMessage().' in '.$e->getFile().':'.$e->getLine().'. Recreating connection and retrying to call method...');
+                unset($this->sock);
+                $this->sock = new \danog\MadelineProto\Connection($this->settings['connection']['ip_address'], $this->settings['connection']['port'], $this->settings['connection']['protocol']);
+                continue;
+            }
+            if ($server_answer == null) {
+                throw new Exception('An error occurred while calling method '.$method.'.');
+            }
+
+            return $server_answer;
+        }
+        throw new Exception('An error occurred while calling method '.$method.'.');
+    }
+
+    public function object_call($object, $args)
+    {
+        foreach (range(1, $this->settings['max_tries']['query']) as $i) {
+            try {
+                $int_message_id = $this->send_message($this->tl->serialize_obj($object, $args), $this->tl->content_related($object));
+//                $server_answer = $this->recv_message();
+                $this->outgoing_messages[$int_message_id]['method'] = $object;
+                $this->outgoing_messages[$int_message_id]['args'] = $args;
+            } catch (Exception $e) {
+                $this->log->log('An error occurred while calling object '.$object.': '.$e->getMessage().' in '.$e->getFile().':'.$e->getLine().'. Recreating connection and retrying to call object...');
+                unset($this->sock);
+                $this->sock = new \danog\MadelineProto\Connection($this->settings['connection']['ip_address'], $this->settings['connection']['port'], $this->settings['connection']['protocol']);
+                continue;
+            }
+
+            return;
+//            if ($server_answer == null) {
+//                throw new Exception('An error occurred while calling object '.$object.'.');
+//            }
+//            $deserialized = $this->tl->deserialize(\danog\MadelineProto\Tools::fopen_and_write('php://memory', 'rw+b', $server_answer));
+//            return $deserialized;
+        }
+        throw new Exception('An error occurred while calling object '.$object.'.');
+    }
+}

src/danog/MadelineProto/MTProtoTools/ResponseHandler.php

@@ -17,7 +17,7 @@ namespace danog\MadelineProto\MTProtoTools;
  */
 class ResponseHandler extends MsgIdHandler
 {
-    public function handle_message($response)
+    public function handle_message($response, $last_sent, $last_received)
     {
         switch ($response['_']) {
             case 'msgs_ack':
@@ -27,40 +27,39 @@ class ResponseHandler extends MsgIdHandler
                 break;
 
             case 'rpc_result':
-                end($this->incoming_messages);
+                $this->ack_incoming_message_id($last_received); // Acknowledge that I received the server's response
-                end($this->outgoing_messages);
-                $this->ack_incoming_message_id(key($this->incoming_messages)); // Acknowledge that I received the server's response
                 $this->ack_outgoing_message_id($response['req_msg_id']); // Acknowledge that the server received my request
-                if ($response['req_msg_id'] != key($this->outgoing_messages)) {
-                    throw new Exception('Message id mismatch; req_msg_id ('.$response['req_msg_id'].') != last sent msg id ('.key($this->outgoing_messages).').');
-                }
 
                 return $this->handle_response($response['result'], $response['req_msg_id']);
                 break;
 
             case 'future_salts':
-                end($this->outgoing_messages);
                 $this->ack_outgoing_message_id($response['req_msg_id']); // Acknowledge that the server received my request
-                if ($response['req_msg_id'] != key($this->outgoing_messages)) {
-                    throw new Exception('Message id mismatch; req_msg_id ('.$response['req_msg_id'].') != last sent msg id ('.key($this->outgoing_messages).').');
-                }
-                $this->log->log('Received future salts.');
 
                 return $this->handle_response($response, $response['req_msg_id']);
                 break;
 
+            case 'bad_msg_notification':
+            case 'bad_server_salt':
+                $this->ack_outgoing_message_id($response['bad_msg_id']); // Acknowledge that the server received my request
+
+                return $this->handle_response($response, $response['bad_msg_id']);
+                break;
+
             case 'pong':
-                foreach ($this->outgoing_messages as $omessage) {
+                foreach ($this->outgoing_messages as $msg_id => &$omessage) {
-                    if ($omessage['args']['ping_id'] == $response['ping_id']) {
+                    if (isset($omessage['content']['args']['ping_id']) && $omessage['content']['args']['ping_id'] == $response['ping_id']) {
-                        $this->outgoing_messages[$response['msg_id']]['response'] = $response;
+                        $omessage['response'] = $response['msg_id'];
+                        $this->incoming_messages[$response['msg_id']]['content'] = $response;
+                        $this->ack_outgoing_message_id($msg_id);
                     }
                 }
                 break;
             case 'new_session_created':
-                end($this->incoming_messages);
+                $this->ack_incoming_message_id($last_received); // Acknowledge that I received the server's response
-                $this->ack_incoming_message_id(key($this->incoming_messages)); // Acknowledge that I received the server's response
                 $this->log->log('new session created');
                 $this->log->log($response);
+                $this->settings['authorization']['temp_auth_key']['server_salt'] = $response['server_salt'];
                 break;
             case 'msg_container':
                 $responses = [];
@@ -68,43 +67,61 @@ class ResponseHandler extends MsgIdHandler
                 $this->log->log($response['messages']);
                 foreach ($response['messages'] as $message) {
                     $this->incoming_messages[$message['msg_id']] = ['seq_no' => $message['seqno']];
-                    $responses[] = $this->handle_message($message['body']);
+                    $responses[] = $this->handle_message($message['body'], $last_sent, $message['msg_id']);
                 }
-                foreach ($responses as $response) {
+                foreach ($responses as $key => $response) {
-                    if ($response !== null) {
+                    if ($response == null) {
-                        return $response;
+                        unset($responses[$key]);
                     }
                 }
+                switch (count($responses)) {
+                    case 0:
+                        return;
+
+                    case 1:
+                        return end($responses);
+                        break;
+                    default:
+                        $this->log->log("Received multiple responses, returning last one");
+                        $this->log->log($responses);
+                        return end($responses);
+                        break;
+                }
                 break;
             case 'msg_copy':
-                return $this->handle_response($response['orig_message']);
+                $this->ack_incoming_message_id($last_received); // Acknowledge that I received the server's response
-                break;
+                if (isset($this->incoming_messages[$response['orig_message']['msg_id']])) {
-            case 'gzip_packed':
+                    $this->ack_incoming_message_id($response['orig_message']['msg_id']); // Acknowledge that I received the server's response
-                return $this->handle_response(gzdecode($response));
+                } else {
+                    return $this->handle_response($response['orig_message']);
+                }
                 break;
             case 'http_wait':
                 $this->log->log('Received http wait.');
                 $this->log->log($response);
                 break;
             default:
-                end($this->incoming_messages);
+                $this->ack_incoming_message_id($last_received); // Acknowledge that I received the server's response
-                $this->ack_incoming_message_id(key($this->incoming_messages)); // Acknowledge that I received the server's response
                 return $this->handle_response($response);
                 break;
         }
     }
 
-    public function handle_response($response, $res_id = null)
+    public function handle_response($response, $res_id = null, $last_received = null)
     {
         if ($res_id == null) {
             return $response;
         }
         switch ($response['_']) {
+            case 'gzip_packed':
+                return $this->handle_response(gzdecode($response), $res_id, $last_received);
+                break;
             case 'rpc_answer_dropped_running':
             case 'rpc_answer_dropped':
                 $this->ack_outgoing_message_id($response['req_msg_id']); // Acknowledge that the server received the original query (the same one, the response to which we wish to forget)
             default:
-                $this->outgoing_messages[$res_id]['response'] = $response;
+                $this->outgoing_messages[$res_id]['response'] = $last_received;
+                $this->incoming_messages[$last_received]['content'] = $response;
                 break;
         }
     }