[Http] Adding an AsyncPlugin

src/Guzzle/Http/Curl/CurlHandle.php

@@ -220,7 +220,10 @@ class CurlHandle
             $request->setHeader($key, $value);
         }
 
-        return new static($handle, $curlOptions);
+        $handle = new static($handle, $curlOptions);
+        $mediator->setCurlHandle($handle);
+
+        return $handle;
     }
 
     /**

src/Guzzle/Http/Curl/CurlMulti.php

@@ -526,9 +526,14 @@ class CurlMulti extends AbstractHasDispatcher implements CurlMultiInterface
                 'request'   => $this,
                 'exception' => $e
             ));
+
             // Allow things to ignore the error if possible
-            if ($request->getState() != RequestInterface::STATE_TRANSFER) {
+            $state = $request->getState();
+            if ($state != RequestInterface::STATE_TRANSFER) {
                 $this->remove($request);
+            }
+            // The error was not handled, so fail
+            if ($state == RequestInterface::STATE_ERROR) {
                 throw $e;
             }
         } else {
@@ -579,8 +584,8 @@ class CurlMulti extends AbstractHasDispatcher implements CurlMultiInterface
         $e = new CurlException(sprintf('[curl] %s: %s [url] %s [info] %s [debug] %s',
             $handle->getErrorNo(), $handle->getError(), $handle->getUrl(),
             var_export($handle->getInfo(), true), $handle->getStderr()));
-
+        $e->setCurlHandle($handle)
-        $e->setRequest($request)
+          ->setRequest($request)
           ->setError($handle->getError(), $handle->getErrorNo());
 
         return $e;

src/Guzzle/Http/Curl/RequestMediator.php

@@ -19,6 +19,11 @@ class RequestMediator
      */
     protected $emitIo;
 
+    /**
+     * @var CurlHandle
+     */
+    protected $curlHandle;
+
     /**
      * @param RequestInterface $request Request to mediate
      */
@@ -28,6 +33,20 @@ class RequestMediator
         $this->emitIo = $request->getParams()->get('curl.emit_io');
     }
 
+    /**
+     * Set the associated CurlHandle object
+     *
+     * @param CurlHandle $handle Curl handle
+     *
+     * @return RequestMediator
+     */
+    public function setCurlHandle(CurlHandle $handle)
+    {
+        $this->curlHandle = $handle;
+
+        return $this;
+    }
+
     /**
      * Receive a response header from curl
      *
@@ -53,6 +72,7 @@ class RequestMediator
     {
         $this->request->dispatch('curl.callback.progress', array(
             'request'       => $this->request,
+            'handle'        => $this->curlHandle,
             'download_size' => $downloadSize,
             'downloaded'    => $downloaded,
             'upload_size'   => $uploadSize,

src/Guzzle/Http/Exception/CurlException.php

@@ -2,6 +2,8 @@
 
 namespace Guzzle\Http\Exception;
 
+use Guzzle\Http\Curl\CurlHandle;
+
 /**
  * cURL request exception
  */
@@ -9,12 +11,15 @@ class CurlException extends BadResponseException
 {
     private $curlError;
     private $curlErrorNo;
+    private $handle;
 
     /**
      * Set the cURL error message
      *
      * @param string $error  Curl error
      * @param int    $number Curl error number
+     *
+     * @return self
      */
     public function setError($error, $number)
     {
@@ -24,6 +29,30 @@ class CurlException extends BadResponseException
         return $this;
     }
 
+    /**
+     * Set the associated curl handle
+     *
+     * @param CurlHandle $handle Curl handle
+     *
+     * @return self
+     */
+    public function setCurlHandle(CurlHandle $handle)
+    {
+        $this->handle = $handle;
+
+        return $this;
+    }
+
+    /**
+     * Get the associated cURL handle
+     *
+     * @return CurlHandle|null
+     */
+    public function getCurlHandle()
+    {
+        return $this->handle;
+    }
+
     /**
      * Get the associated cURL error message
      *

src/Guzzle/Http/Plugin/AsyncPlugin.php

@@ -0,0 +1,92 @@
+<?php
+
+namespace Guzzle\Http\Plugin;
+
+use Guzzle\Common\Event;
+use Guzzle\Http\Message\Response;
+use Guzzle\Http\Exception\CurlException;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+
+/**
+ * Sends requests but does not wait for the response
+ */
+class AsyncPlugin implements EventSubscriberInterface
+{
+    /**
+     * {@inheritdoc}
+     */
+    public static function getSubscribedEvents()
+    {
+        return array(
+            'request.before_send'    => 'onBeforeSend',
+            'request.exception'      => 'onRequestTimeout',
+            'request.sent'           => 'onRequestSent',
+            'curl.callback.progress' => 'onCurlProgess'
+        );
+    }
+
+    /**
+     * Event emitted before a request is sent.  Ensure that progress callback
+     * are emitted from the curl handle's request mediator.
+     *
+     * @param Event $event
+     */
+    public function onBeforeSend(Event $event)
+    {
+        // Ensure that progress callbacks are dispatched
+        $event['request']->getCurlOptions()->set('progress', true);
+    }
+
+    /**
+     * Event emitted when a curl progress function is called.  When the amount
+     * of data uploaded == the amount of data to upload OR any bytes have been
+     * downloaded, then time the request out after 1ms because we're done with
+     * transmitting the request, and tell curl not download a body.
+     *
+     * @param Event $event
+     */
+    public function onCurlProgess(Event $event)
+    {
+        if (!$event['handle']) {
+            return;
+        }
+
+        if ($event['downloaded'] || ($event['uploaded'] || $event['upload_size'] === $event['uploaded'])) {
+            $event['handle']->getOptions()
+                ->set(CURLOPT_TIMEOUT_MS, 1)
+                ->set(CURLOPT_NOBODY, true);
+            // Timeout after 1ms
+            curl_setopt($event['handle']->getHandle(), CURLOPT_TIMEOUT_MS, 1);
+            // Even if the response is quick, tell curl not to download the body
+            curl_setopt($event['handle']->getHandle(), CURLOPT_NOBODY, true);
+        }
+    }
+
+    /**
+     * Event emitted when a curl exception occurs. Ignore the exception and 
+     * set a mock response.
+     *
+     * @param Event $event
+     */
+    public function onRequestTimeout(Event $event)
+    {
+        if ($event['exception'] instanceof CurlException) {
+            $event['request']->setResponse(new Response(200, array(
+                'X-Guzzle-Async' => 'Did not wait for the response'
+            )));
+        }
+    }
+
+    /**
+     * Event emitted when a request completes because it took less than 1ms.
+     * Add an X-Guzzle-Async header to notify the caller that there is no 
+     * body in the message.
+     *
+     * @param Event $event
+     */
+    public function onRequestSent(Event $event)
+    {
+        // Let the caller know this was meant to be async
+        $event['request']->getResponse()->setHeader('X-Guzzle-Async', 'Did not wait for the response');
+    }
+}

tests/Guzzle/Tests/Http/Curl/CurlMultiTest.php

@@ -485,7 +485,7 @@ class CurlMultiTest extends \Guzzle\Tests\GuzzleTestCase
     /**
      * @covers Guzzle\Http\Curl\CurlMulti::send
      */
-    public function testDoesNotThrowExceptionsWhenRequestsRecover()
+    public function testDoesNotThrowExceptionsWhenRequestsRecoverWithRetry()
     {
         $this->getServer()->flush();
         $client = new Client($this->getServer()->getUrl());
@@ -500,6 +500,29 @@ class CurlMultiTest extends \Guzzle\Tests\GuzzleTestCase
         $this->assertEquals(0, count($this->getServer()->getReceivedRequests(false)));
     }
 
+    /**
+     * @covers Guzzle\Http\Curl\CurlMulti::send
+     */
+    public function testDoesNotThrowExceptionsWhenRequestsRecoverWithSuccess()
+    {
+        // Attempt a port that 99.9% is not listening
+        $client = new Client('http://localhost:123');
+        $request = $client->get();
+        // Ensure it times out quickly if needed
+        $request->getCurlOptions()->set(CURLOPT_TIMEOUT_MS, 1)->set(CURLOPT_CONNECTTIMEOUT_MS, 1);
+
+        $request->getEventDispatcher()->addListener('request.exception', function(Event $event) use (&$count) {
+            $event['request']->setResponse(new Response(200));
+        });
+
+        $multi = new CurlMulti();
+        $multi->add($request);
+        $multi->send();
+
+        // Ensure that the exception was caught, and the response was set manually
+        $this->assertEquals(200, $request->getResponse()->getStatusCode());
+    }
+
     /**
      * @covers Guzzle\Http\Curl\CurlMulti::reset
      */

tests/Guzzle/Tests/Http/Exception/CurlExceptionTest.php

@@ -3,6 +3,7 @@
 namespace Guzzle\Tests\Http\Exception;
 
 use Guzzle\Http\Exception\CurlException;
+use Guzzle\Http\Curl\CurlHandle;
 
 /**
  * @covers Guzzle\Http\Exception\CurlException
@@ -17,5 +18,10 @@ class CurlExceptionTest extends \Guzzle\Tests\GuzzleTestCase
         $this->assertSame($e, $e->setError('test', 12));
         $this->assertEquals('test', $e->getError());
         $this->assertEquals(12, $e->getErrorNo());
+
+        $handle = new CurlHandle(curl_init(), array());
+        $e->setCurlHandle($handle);
+        $this->assertSame($handle, $e->getCurlHandle());
+        $handle->close();
     }
 }

tests/Guzzle/Tests/Http/Plugin/AsyncPluginTest.php

@@ -0,0 +1,105 @@
+<?php
+
+namespace Guzzle\Tests\Http\Plugin;
+
+use Guzzle\Http\Plugin\AsyncPlugin;
+use Guzzle\Http\Message\RequestInterface;
+use Guzzle\Http\Message\RequestFactory;
+use Guzzle\Http\Plugin\OauthPlugin;
+use Guzzle\Http\Curl\CurlHandle;
+use Guzzle\Http\Exception\CurlException;
+use Guzzle\Common\Event;
+use Guzzle\Http\Client;
+
+class AsyncPluginTest extends \Guzzle\Tests\GuzzleTestCase
+{
+    /**
+     * @covers Guzzle\Http\Plugin\AsyncPlugin::getSubscribedEvents
+     */
+    public function testSubscribesToEvents()
+    {
+        $events = AsyncPlugin::getSubscribedEvents();
+        $this->assertArrayHasKey('request.before_send', $events);
+        $this->assertArrayHasKey('request.exception', $events);
+        $this->assertArrayHasKey('curl.callback.progress', $events);
+    }
+
+    /**
+     * @covers Guzzle\Http\Plugin\AsyncPlugin::onBeforeSend
+     */
+    public function testEnablesProgressCallbacks()
+    {
+        $p = new AsyncPlugin();
+        $request = RequestFactory::getInstance()->create('PUT', 'http://www.example.com');
+        $event = new Event(array(
+            'request' => $request
+        ));
+        $p->onBeforeSend($event);
+        $this->assertEquals(true, $request->getCurlOptions()->get('progress'));
+    }
+
+    /**
+     * @covers Guzzle\Http\Plugin\AsyncPlugin::onCurlProgess
+     */
+    public function testAddsTimesOutAfterSending()
+    {
+        $p = new AsyncPlugin();
+        $request = RequestFactory::getInstance()->create('PUT', 'http://www.example.com');
+        $handle = CurlHandle::factory($request);
+        $event = new Event(array(
+            'request'     => $request,
+            'handle'      => $handle,
+            'uploaded'    => 10,
+            'upload_size' => 10,
+            'downloaded'  => 0
+        ));
+        $p->onCurlProgess($event);
+        $this->assertEquals(1, $handle->getOptions()->get(CURLOPT_TIMEOUT_MS));
+        $this->assertEquals(true, $handle->getOptions()->get(CURLOPT_NOBODY));
+    }
+
+    /**
+     * @covers Guzzle\Http\Plugin\AsyncPlugin::onCurlProgess
+     */
+    public function testEnsuresRequestIsSet()
+    {
+        $p = new AsyncPlugin();
+        $event = new Event(array(
+            'uploaded'    => 10,
+            'upload_size' => 10,
+            'downloaded'  => 0
+        ));
+        $p->onCurlProgess($event);
+    }
+
+    /**
+     * @covers Guzzle\Http\Plugin\AsyncPlugin::onRequestTimeout
+     */
+    public function testMasksCurlExceptions()
+    {
+        $p = new AsyncPlugin();
+        $request = RequestFactory::getInstance()->create('PUT', 'http://www.example.com');
+        $e = new CurlException('Error');
+        $event = new Event(array(
+            'request'   => $request,
+            'exception' => $e
+        ));
+        $p->onRequestTimeout($event);
+        $this->assertEquals(RequestInterface::STATE_COMPLETE, $request->getState());
+        $this->assertEquals(200, $request->getResponse()->getStatusCode());
+        $this->assertTrue($request->getResponse()->hasHeader('X-Guzzle-Async'));
+    }
+
+    public function testEnsuresIntegration()
+    {
+        $this->getServer()->enqueue("HTTP/1.1 204 FOO\r\nContent-Length: 4\r\n\r\ntest");
+        $client = new Client($this->getServer()->getUrl());
+        $request = $client->post('/', null, array(
+            'foo' => 'bar'
+        ));
+        $request->getEventDispatcher()->addSubscriber(new AsyncPlugin());
+        $request->send();
+        $this->assertEquals('', $request->getResponse()->getBody(true));
+        $this->assertTrue($request->getResponse()->hasHeader('X-Guzzle-Async'));
+    }
+}