diff --git a/lib/Alchemy/Phrasea/TaskManager/Job/WebhookJob.php b/lib/Alchemy/Phrasea/TaskManager/Job/WebhookJob.php index dbe4d7bd41..fb69af186b 100644 --- a/lib/Alchemy/Phrasea/TaskManager/Job/WebhookJob.php +++ b/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()); + } +} \ No newline at end of file