php-amqplib · lukebakken · Feb 18, 2019 · Jan 21, 2019
diff --git a/PhpAmqpLib/Channel/AbstractChannel.php b/PhpAmqpLib/Channel/AbstractChannel.php
@@ -4,6 +4,7 @@
 use PhpAmqpLib\Connection\AbstractConnection;
 use PhpAmqpLib\Exception\AMQPChannelClosedException;
 use PhpAmqpLib\Exception\AMQPInvalidFrameException;
+use PhpAmqpLib\Exception\AMQPNoDataException;
 use PhpAmqpLib\Exception\AMQPNotImplementedException;
 use PhpAmqpLib\Exception\AMQPOutOfBoundsException;
 use PhpAmqpLib\Exception\AMQPOutOfRangeException;
@@ -213,7 +214,7 @@ public function dispatch($method_sig, $args, $amqpMessage)
     }
 
     /**
-     * @param int $timeout
+     * @param int|float|null $timeout
      * @return array|mixed
      */
     public function next_frame($timeout = 0)
@@ -318,7 +319,7 @@ protected function createMessage($propertyReader, $contentReader)
      *
      * @param array $allowed_methods
      * @param bool $non_blocking
-     * @param int $timeout
+     * @param int|float|null $timeout
      * @throws \PhpAmqpLib\Exception\AMQPOutOfBoundsException
      * @throws \PhpAmqpLib\Exception\AMQPRuntimeException
      * @throws \PhpAmqpLib\Exception\AMQPTimeoutException
@@ -334,9 +335,19 @@ public function wait($allowed_methods = null, $non_blocking = false, $timeout =
             return $this->dispatch_deferred_method($deferred['queued_method']);
         }
 
+        // timeouts must be deactivated for non-blocking actions
+        if (true === $non_blocking) {
+            $timeout = null;
+        }
+
         // No deferred methods?  wait for new ones
         while (true) {
-            list($frame_type, $payload) = $this->next_frame($timeout);
+            try {
+                list($frame_type, $payload) = $this->next_frame($timeout);
+            } catch (AMQPNoDataException $e) {
+                // no data ready for non-blocking actions - stop and exit
+                break;
+            }
 
             $this->validate_method_frame($frame_type);
             $this->validate_frame_payload($payload);

diff --git a/PhpAmqpLib/Connection/AbstractConnection.php b/PhpAmqpLib/Connection/AbstractConnection.php
@@ -1,11 +1,12 @@
 <?php
 namespace PhpAmqpLib\Connection;
 
-use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Channel\AbstractChannel;
+use PhpAmqpLib\Channel\AMQPChannel;
 use PhpAmqpLib\Exception\AMQPConnectionClosedException;
 use PhpAmqpLib\Exception\AMQPHeartbeatMissedException;
 use PhpAmqpLib\Exception\AMQPInvalidFrameException;
+use PhpAmqpLib\Exception\AMQPNoDataException;
 use PhpAmqpLib\Exception\AMQPProtocolConnectionException;
 use PhpAmqpLib\Exception\AMQPRuntimeException;
 use PhpAmqpLib\Exception\AMQPTimeoutException;
@@ -260,6 +261,8 @@ protected function connect()
             // Something went wrong, set the connection status
             $this->setIsConnected(false);
             $this->closeChannels();
+            $this->close_input();
+            $this->close_socket();
             throw $e; // Rethrow exception
         }
     }
@@ -330,7 +333,7 @@ public function set_close_on_destruct($close = true)
 
     protected function close_input()
     {
-        $this->debug->debug_msg('closing input');
+        $this->debug && $this->debug->debug_msg('closing input');
 
         if (!is_null($this->input)) {
             $this->input->close();
@@ -340,7 +343,7 @@ protected function close_input()
 
     protected function close_socket()
     {
-        $this->debug->debug_msg('closing socket');
+        $this->debug && $this->debug->debug_msg('closing socket');
 
         if (!is_null($this->getIO())) {
             $this->getIO()->close();
@@ -516,7 +519,7 @@ protected function prepare_channel_method_frame($channel, $method_sig, $args = '
     /**
      * Waits for a frame from the server
      *
-     * @param int $timeout
+     * @param int|float|null $timeout
      * @return array
      * @throws \Exception
      * @throws \PhpAmqpLib\Exception\AMQPTimeoutException
@@ -556,6 +559,9 @@ protected function wait_frame($timeout = 0)
         } catch (AMQPTimeoutException $e) {
             $this->input->setTimeout($currentTimeout);
             throw $e;
+        } catch (AMQPNoDataException $e) {
+            $this->input->setTimeout($currentTimeout);
+            throw $e;
         }
 
         $this->input->setTimeout($currentTimeout);
@@ -574,7 +580,7 @@ protected function wait_frame($timeout = 0)
      * Waits for a frame from the server destined for a particular channel.
      *
      * @param string $channel_id
-     * @param int $timeout
+     * @param int|float|null $timeout
      * @return array
      */
     protected function wait_channel($channel_id, $timeout = 0)
@@ -585,8 +591,7 @@ protected function wait_channel($channel_id, $timeout = 0)
             $now = time();
             try {
                 list($frame_type, $frame_channel, $payload) = $this->wait_frame($_timeout);
-            }
-            catch ( AMQPTimeoutException $e ) {
+            } catch (AMQPTimeoutException $e) {
                 if ( $this->heartbeat && microtime(true) - ($this->heartbeat*2) > $this->last_frame ) {
                     $this->debug->debug_msg("missed server heartbeat (at threshold * 2)");
                     $this->setIsConnected(false);

diff --git a/PhpAmqpLib/Exception/AMQPNoDataException.php b/PhpAmqpLib/Exception/AMQPNoDataException.php
@@ -0,0 +1,10 @@
+<?php
+
+namespace PhpAmqpLib\Exception;
+
+/**
+ * Used mostly in non-blocking methods when no data is ready for processing.
+ */
+class AMQPNoDataException extends AMQPRuntimeException
+{
+}
diff --git a/PhpAmqpLib/Helper/MiscHelper.php b/PhpAmqpLib/Helper/MiscHelper.php
@@ -32,11 +32,11 @@ public static function saveBytes($bytes)
      * different arguments
      *
      * @param int|float $number
-     * @return array
+     * @return int[]
      */
     public static function splitSecondsMicroseconds($number)
     {
-        return array(floor($number), ($number - floor($number)) * 1000000);
+        return array((int)floor($number), (int)(fmod($number, 1) * 1000000));
     }
 
     /**

diff --git a/PhpAmqpLib/Wire/AMQPReader.php b/PhpAmqpLib/Wire/AMQPReader.php
@@ -3,6 +3,7 @@
 
 use PhpAmqpLib\Exception\AMQPDataReadException;
 use PhpAmqpLib\Exception\AMQPInvalidArgumentException;
+use PhpAmqpLib\Exception\AMQPNoDataException;
 use PhpAmqpLib\Exception\AMQPOutOfBoundsException;
 use PhpAmqpLib\Exception\AMQPTimeoutException;
 use PhpAmqpLib\Exception\AMQPIOWaitException;
@@ -41,7 +42,7 @@ class AMQPReader extends AbstractClient
     /** @var bool */
     protected $is64bits;
 
-    /** @var int */
+    /** @var int|float|null */
     protected $timeout;
 
     /** @var int */
@@ -111,28 +112,40 @@ public function read($n)
      *
      * AMQPTimeoutException can be raised if the timeout is set
      *
-     * @throws \PhpAmqpLib\Exception\AMQPIOWaitException
-     * @throws \PhpAmqpLib\Exception\AMQPTimeoutException
+     * @throws \PhpAmqpLib\Exception\AMQPIOWaitException on network errors
+     * @throws \PhpAmqpLib\Exception\AMQPTimeoutException when timeout is set and no data received
+     * @throws \PhpAmqpLib\Exception\AMQPNoDataException when no data is ready to read from IO
      */
     protected function wait()
     {
-        if ($this->getTimeout() == 0) {
-            return null;
+        $timeout = $this->getTimeout();
+        if (null === $timeout) {
+            // timeout=null just poll state and return instantly
+            $sec = 0;
+            $usec = 0;
+        } elseif ($timeout > 0) {
+            list($sec, $usec) = MiscHelper::splitSecondsMicroseconds($this->getTimeout());
+        } else {
+            // wait indefinitely for data if timeout=0
+            $sec = null;
+            $usec = 0;
         }
 
-        // wait ..
-        list($sec, $usec) = MiscHelper::splitSecondsMicroseconds($this->getTimeout());
         $result = $this->io->select($sec, $usec);
 
         if ($result === false) {
-            throw new AMQPIOWaitException('A network error occured while awaiting for incoming data');
+            throw new AMQPIOWaitException('A network error occurred while awaiting for incoming data');
         }
 
         if ($result === 0) {
-            throw new AMQPTimeoutException(sprintf(
-                'The connection timed out after %s sec while awaiting incoming data',
-                $this->getTimeout()
-            ));
+            if ($timeout > 0) {
+                throw new AMQPTimeoutException(sprintf(
+                    'The connection timed out after %s sec while awaiting incoming data',
+                    $timeout
+                ));
+            } else {
+                throw new AMQPNoDataException('No data is ready to read');
+            }
         }
     }
 
@@ -141,6 +154,7 @@ protected function wait()
      * @return string
      * @throws \RuntimeException
      * @throws \PhpAmqpLib\Exception\AMQPDataReadException
+     * @throws \PhpAmqpLib\Exception\AMQPNoDataException
      */
     protected function rawread($n)
     {
@@ -533,15 +547,15 @@ protected function tell()
     /**
      * Sets the timeout (second)
      *
-     * @param int $timeout
+     * @param int|float|null $timeout
      */
     public function setTimeout($timeout)
     {
         $this->timeout = $timeout;
     }
 
     /**
-     * @return int
+     * @return int|float
      */
     public function getTimeout()
     {

diff --git a/PhpAmqpLib/Wire/IO/SocketIO.php b/PhpAmqpLib/Wire/IO/SocketIO.php
@@ -3,6 +3,7 @@
 
 use PhpAmqpLib\Exception\AMQPHeartbeatMissedException;
 use PhpAmqpLib\Exception\AMQPIOException;
+use PhpAmqpLib\Exception\AMQPIOWaitException;
 use PhpAmqpLib\Exception\AMQPSocketException;
 use PhpAmqpLib\Helper\MiscHelper;
 use PhpAmqpLib\Wire\AMQPWriter;
@@ -36,6 +37,9 @@ class SocketIO extends AbstractIO
     /** @var bool */
     private $keepalive;
 
+    /** @var array */
+    private $last_error;
+
     /**
      * @param string $host
      * @param int $port
@@ -205,8 +209,17 @@ public function select($sec, $usec)
         $read = array($this->sock);
         $write = null;
         $except = null;
+        $result = false;
+
+        $this->set_error_handler();
+        try {
+            $result = socket_select($read, $write, $except, $sec, $usec);
+            $this->cleanup_error_handler();
+        } catch (\ErrorException $e) {
+            throw new AMQPIOWaitException($e->getMessage(), $e->getCode(), $e);
+        }
 
-        return socket_select($read, $write, $except, $sec, $usec);
+        return $result;
     }
 
     /**
@@ -258,4 +271,44 @@ protected function write_heartbeat()
         $pkt->write_octet(0xCE);
         $this->write($pkt->getvalue());
     }
+
+    public function error_handler($errno, $errstr, $errfile, $errline, $errcontext = null)
+    {
+        // socket_select warning that it has been interrupted by a signal - EINTR
+        if (false !== strrpos($errstr, socket_strerror(SOCKET_EINTR))) {
+            // it's allowed while processing signals
+            return;
+        }
+
+        $this->last_error = compact('errno', 'errstr', 'errfile', 'errline', 'errcontext');
+    }
+
+    /**
+     * Begin tracking errors and set the error handler
+     */
+    protected function set_error_handler()
+    {
+        $this->last_error = null;
+        socket_clear_error($this->sock);
+        set_error_handler(array($this, 'error_handler'));
+    }
+
+    /**
+     * throws an ErrorException if an error was handled
+     * @throws \ErrorException
+     */
+    protected function cleanup_error_handler()
+    {
+        restore_error_handler();
+
+        if ($this->last_error !== null) {
+            throw new \ErrorException(
+                $this->last_error['errstr'],
+                0,
+                $this->last_error['errno'],
+                $this->last_error['errfile'],
+                $this->last_error['errline']
+            );
+        }
+    }
 }
diff --git a/demo/amqp_consumer_non_blocking.php b/demo/amqp_consumer_non_blocking.php
@@ -80,12 +80,7 @@ function shutdown($channel, $connection)
 
 // Loop as long as the channel has callbacks registered
 while (count($channel->callbacks)) {
-    $read = array($connection->getSocket()); // add here other sockets that you need to attend
-    $write = null;
-    $except = null;
-    if (false === ($changeStreamsCount = stream_select($read, $write, $except, 60))) {
-        /* Error handling */
-    } elseif ($changeStreamsCount > 0 || $channel->hasPendingMethods()) {
-        $channel->wait();
-    }
+    $channel->wait(null, true);
+    // do something else
+    usleep(300000);
 }