port modification made on 4.0 to master

lib/Alchemy/Phrasea/TaskManager/Job/WebhookJob.php

@@ -12,11 +12,21 @@
 namespace Alchemy\Phrasea\TaskManager\Job;
 
 use Alchemy\Phrasea\Core\Version;
+use Alchemy\Phrasea\Model\Entities\WebhookEvent;
+use Alchemy\Phrasea\Model\Entities\WebhookEventDelivery;
+use Alchemy\Phrasea\Model\Entities\ApiApplication;
 use Alchemy\Phrasea\TaskManager\Editor\DefaultEditor;
 use Alchemy\Phrasea\Webhook\EventProcessorFactory;
 use Guzzle\Http\Client as GuzzleClient;
-use Psr\Log\LoggerInterface;
+use Guzzle\Batch\BatchBuilder;
+use Guzzle\Http\Message\Request;
 use Silex\Application;
+use Guzzle\Common\Event;
+use Guzzle\Plugin\Backoff\BackoffPlugin;
+use Guzzle\Plugin\Backoff\TruncatedBackoffStrategy;
+use Guzzle\Plugin\Backoff\CallbackBackoffStrategy;
+use Guzzle\Plugin\Backoff\CurlBackoffStrategy;
+use Psr\Log\LoggerInterface;
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 use Symfony\Component\Translation\TranslatorInterface;
 
@@ -24,12 +34,15 @@ class WebhookJob extends AbstractJob
 {
     private $httpClient;
 
+    private $firstRun = true;
+
     public function __construct(
         TranslatorInterface $translator,
         EventDispatcherInterface $dispatcher = null,
         LoggerInterface $logger = null,
         GuzzleClient $httpClient = null
-    ) {
+    )
+    {
         parent::__construct($translator, $dispatcher, $logger);
 
         $this->httpClient = $httpClient ?: new GuzzleClient();
@@ -76,6 +89,57 @@ class WebhookJob extends AbstractJob
     {
         $app = $data->getApplication();
         $thirdPartyApplications = $app['repo.api-applications']->findWithDefinedWebhookCallback();
+        $that = $this;
+
+        if ($this->firstRun) {
+            $this->httpClient->getEventDispatcher()->addListener('request.error', function (Event $event) {
+                // override guzzle default behavior of throwing exceptions
+                // when 4xx & 5xx responses are encountered
+                $event->stopPropagation();
+            }, -254);
+
+            // Set callback which logs success or failure
+            $subscriber = new CallbackBackoffStrategy(function ($retries, Request $request, $response, $e) use ($app, $that) {
+                $retry = true;
+                if ($response && (null !== $deliverId = parse_url($request->getUrl(), PHP_URL_FRAGMENT))) {
+                    $delivery = $app['repo.webhook-delivery']->find($deliverId);
+
+                    $logContext = [ 'host' => $request->getHost() ];
+
+                    if ($response->isSuccessful()) {
+                        $app['manipulator.webhook-delivery']->deliverySuccess($delivery);
+
+                        $logType = 'info';
+                        $logEntry = sprintf('Deliver success event "%d:%s" for app "%s"',
+                            $delivery->getWebhookEvent()->getId(), $delivery->getWebhookEvent()->getName(),
+                            $delivery->getThirdPartyApplication()->getName()
+                        );
+
+                        $retry = false;
+                    } else {
+                        $app['manipulator.webhook-delivery']->deliveryFailure($delivery);
+
+                        $logType = 'error';
+                        $logEntry = sprintf('Deliver failure event "%d:%s" for app "%s"',
+                            $delivery->getWebhookEvent()->getId(), $delivery->getWebhookEvent()->getName(),
+                            $delivery->getThirdPartyApplication()->getName()
+                        );
+                    }
+
+                    $that->log($logType, $logEntry, $logContext);
+
+                    return $retry;
+                }
+            }, true, new CurlBackoffStrategy());
+
+            // set max retries
+            $subscriber = new TruncatedBackoffStrategy(WebhookEventDelivery::MAX_DELIVERY_TRIES, $subscriber);
+            $subscriber = new BackoffPlugin($subscriber);
+
+            $this->httpClient->addSubscriber($subscriber);
+
+            $this->firstRun = false;
+        }
 
         /** @var EventProcessorFactory $eventFactory */
         $eventFactory = $app['webhook.processor_factory'];
@@ -91,4 +155,41 @@ class WebhookJob extends AbstractJob
             $this->deliverEvent($eventFactory, $app, $thirdPartyApplications, $event);
         }
     }
-}
+
+    private function deliverEvent(EventProcessorFactory $eventFactory, Application $app, array $thirdPartyApplications, WebhookEvent $event)
+    {
+        if (count($thirdPartyApplications) === 0) {
+            $this->log('info', sprintf('No applications defined to listen for webhook events'));
+
+            return;
+        }
+
+        // format event data
+        $eventProcessor = $eventFactory->get($event);
+        $data = $eventProcessor->process($event);
+
+        // batch requests
+        $batch = BatchBuilder::factory()
+            ->transferRequests(10)
+            ->build();
+
+        foreach ($thirdPartyApplications as $thirdPartyApplication) {
+            $delivery = $app['manipulator.webhook-delivery']->create($thirdPartyApplication, $event);
+
+            // append delivery id as url anchor
+            $uniqueUrl = $this->getUrl($thirdPartyApplication, $delivery);
+
+            // create http request with data as request body
+            $batch->add($this->httpClient->createRequest('POST', $uniqueUrl, [
+                'Content-Type' => 'application/vnd.phraseanet.event+json'
+            ], json_encode($data)));
+        }
+
+        $batch->flush();
+    }
+
+    private function getUrl(ApiApplication $application, WebhookEventDelivery $delivery)
+    {
+        return sprintf('%s#%s', $application->getWebhookUrl(), $delivery->getId());
+    }
+}