fix bugs for publish_sync

mqtt/Exception/FwriteError.php

@@ -0,0 +1,52 @@
+<?php
+
+/**
+ * MQTT Client
+ *
+ * An open source MQTT client library in PHP
+ *
+ * This content is released under the MIT License (MIT)
+ *
+ * Copyright (c) 2013 - 2016, sskaje (https://sskaje.me/)
+ *
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ *
+ *
+ * @package    sskaje/mqtt
+ * @author     sskaje (https://sskaje.me/)
+ * @copyright  Copyright (c) 2013 - 2016, sskaje (https://sskaje.me/)
+ * @license    http://opensource.org/licenses/MIT MIT License
+ * @link       https://sskaje.me/mqtt/
+ */
+
+
+namespace sskaje\mqtt\Exception;
+use sskaje\mqtt\Exception;
+
+/**
+ * Exception: Connect failed with return value
+ *
+ *
+ *
+ */
+class FwriteError extends Exception {}
+
+
+# EOF

mqtt/MQTT.php

@@ -555,7 +555,14 @@ protected function do_publish($topic, $message, $qos=0, $retain=0, & $msgid=0, $
 
         $publishobj->setMsgID($msgid);
 
-        $publish_bytes_written = $this->message_write($publishobj);
+//        $publish_bytes_written = $this->message_write($publishobj);
+
+        $publish_bytes_written = $this->message_write($publishobj, $msg_len);
+        if( $publish_bytes_written != $msg_len )
+        {
+            $this->reconnect();
+        }
+
         Debug::Log(Debug::DEBUG, 'do_publish(): bytes written=' . $publish_bytes_written);
 
         if ($qos == 1) {
@@ -758,9 +765,20 @@ protected function do_unsubscribe()
      * @return bool
      * @throws Exception
      */
-    public function handle_message()
+    public function handle_message( $time_out = false )
     {
-        $selected = $this->socket->select($this->keepalive / 2);
+        if( empty($time_out) )
+        {
+            $selected = $this->socket->select($this->keepalive / 2);
+        }
+        else
+        {
+            $seconds = intval($time_out);
+            $useconds = ($time_out - $seconds) * pow( 10, 6 );
+
+            $selected = $this->socket->select($time_out, $useconds);
+        }
+
 
         if ($selected === false) {
             # Error
@@ -1137,6 +1155,53 @@ public function loop()
         return true;
     }
 
+    /**
+     * 外部驱动的loop
+     * Main Loop
+     *
+     * @return bool
+     * @throws \Exception
+     */
+    public function loop_driven_by_outside($time_out = 1)
+    {
+        Debug::Log(Debug::DEBUG, 'loop()');
+
+        # check if any commands awaits or topics to subscribe
+        if (!$this->cmdstore->countWaits() && empty($this->topics) && empty($this->topics_to_subscribe) && empty($this->subscribe_awaits)) {
+            Debug::Log(Debug::INFO, "loop(): No tasks, leaving...");
+            return false;
+        }
+
+        # Subscribe topics
+        if (!empty($this->topics_to_subscribe)) {
+            list($last_subscribe_msgid, $last_subscribe_topics) = $this->do_subscribe();
+            $this->subscribe_awaits[$last_subscribe_msgid] = $last_subscribe_topics;
+        }
+        # Unsubscribe topics
+        if (!empty($this->topics_to_unsubscribe)) {
+            list($last_unsubscribe_msgid, $last_unsubscribe_topics) = $this->do_unsubscribe();
+            $this->unsubscribe_awaits[$last_unsubscribe_msgid] = $last_unsubscribe_topics;
+        }
+
+        try {
+            # It is the responsibility of the Client to ensure that the interval between Control Packets
+            # being sent does not exceed the Keep Alive value. In the absence of sending any other Control
+            # Packets, the Client MUST send a PINGREQ Packet [MQTT-3.1.2-23].
+            $this->keepalive();
+
+            $this->handle_message($time_out);
+
+        } catch (Exception\NetworkError $e) {
+            Debug::Log(Debug::INFO, 'loop(): Connection lost.');
+            $this->reconnect();
+            $this->subscribe($this->topics);
+        } catch (\Exception $e) {
+            throw $e;
+        }
+
+        return true;
+    }
+
     protected $last_ping_time = 0;
 
     /**

mqtt/SocketClient.php

@@ -38,6 +38,7 @@
 
 
 namespace sskaje\mqtt;
+use sskaje\mqtt\Exception\FwriteError;
 
 /**
  * Socket Client
@@ -163,7 +164,15 @@ public function write($packet, $packet_size)
     {
         if (!$this->socket || !is_resource($this->socket)) return false;
         Debug::Log(Debug::DEBUG, "socket_write(length={$packet_size})", $packet);
-        return fwrite($this->socket, $packet, $packet_size);
+        $write_len = fwrite($this->socket, $packet, $packet_size);
+
+        $err_arr = error_get_last();
+        if(empty($err_arr))
+        {
+            return $write_len;
+        }
+
+        throw new FwriteError(json_encode($err_arr));
     }
 
     /**
@@ -220,12 +229,12 @@ public function eof()
      * @param int $timeout
      * @return int
      */
-    public function select($timeout)
+    public function select($timeout, $tv_usec = 0)
     {
         $read = array($this->socket);
         $write = $except = NULL;
 
-        return stream_select($read, $write, $except, $timeout);
+        return stream_select($read, $write, $except, $timeout, $tv_usec);
     }
 }