Automatically dispatch updates to clients using the Mercure protocol

composer.json

@@ -23,6 +23,7 @@
         "symfony/property-access": "^2.7 || ^3.0 || ^4.0",
         "symfony/property-info": "^3.3.11 || ^4.0",
         "symfony/serializer": "^4.1",
+        "symfony/web-link": "^4.1",
         "willdurand/negotiation": "^2.0.3"
     },
     "require-dev": {
@@ -58,6 +59,9 @@
         "symfony/finder": "^3.3 || ^4.0",
         "symfony/form": "^3.3 || ^4.0",
         "symfony/framework-bundle": "^3.3 || ^4.0",
+        "symfony/mercure": "*",
+        "symfony/mercure-bundle": "*",
+        "symfony/messenger": "^4.1",
         "symfony/phpunit-bridge": "^3.3 || ^4.0",
         "symfony/routing": "^3.3 || ^4.0",
         "symfony/security": "^3.0 || ^4.0",

features/mercure/discover.feature

@@ -0,0 +1,13 @@
+Feature: Mercure discovery support
+  In order to let the client discovering the Mercure hub
+  As a client software developer
+  I need to retrieve the hub URL through a Link HTTP header
+
+  @createSchema
+  Scenario: Checks that the Mercure Link is added
+    Given I send a "GET" request to "/dummy_mercures"
+    Then the header "Link" should be equal to '<http://example.com/docs.jsonld>; rel="http://www.w3.org/ns/hydra/core#apiDocumentation",<https://demo.mercure.rocks/hub>; rel="mercure"'
+
+  Scenario: Checks that the Mercure Link is not added on endpoints where updates are not dispatched
+    Given I send a "GET" request to "/"
+    Then the header "Link" should be equal to '<http://example.com/docs.jsonld>; rel="http://www.w3.org/ns/hydra/core#apiDocumentation"'

src/Annotation/ApiResource.php

@@ -41,6 +41,7 @@
  *     @Attribute("iri", type="string"),
  *     @Attribute("itemOperations", type="array"),
  *     @Attribute("maximumItemsPerPage", type="int"),
+ *     @Attribute("mercure", type="mixed"),
  *     @Attribute("normalizationContext", type="array"),
  *     @Attribute("order", type="array"),
  *     @Attribute("outputClass", type="string"),
@@ -175,6 +176,13 @@ final class ApiResource
      */
     private $maximumItemsPerPage;
 
+    /**
+     * @see https://github.com/Haehnchen/idea-php-annotation-plugin/issues/112
+     *
+     * @var mixed
+     */
+    private $mercure;
+
     /**
      * @see https://github.com/Haehnchen/idea-php-annotation-plugin/issues/112
      *

src/Bridge/Doctrine/EventListener/PublishMercureUpdatesListener.php

@@ -0,0 +1,178 @@
+<?php
+
+/*
+ * This file is part of the API Platform project.
+ *
+ * (c) Kévin Dunglas <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+declare(strict_types=1);
+
+namespace ApiPlatform\Core\Bridge\Doctrine\EventListener;
+
+use ApiPlatform\Core\Api\IriConverterInterface;
+use ApiPlatform\Core\Api\ResourceClassResolverInterface;
+use ApiPlatform\Core\Api\UrlGeneratorInterface;
+use ApiPlatform\Core\Exception\InvalidArgumentException;
+use ApiPlatform\Core\Exception\RuntimeException;
+use ApiPlatform\Core\Metadata\Resource\Factory\ResourceMetadataFactoryInterface;
+use ApiPlatform\Core\Util\ClassInfoTrait;
+use Doctrine\ORM\Event\OnFlushEventArgs;
+use Symfony\Component\ExpressionLanguage\ExpressionLanguage;
+use Symfony\Component\Mercure\Update;
+use Symfony\Component\Messenger\MessageBusInterface;
+use Symfony\Component\Serializer\SerializerInterface;
+
+/**
+ * Publishes resources updates to the Mercure hub.
+ *
+ * @author Kévin Dunglas <[email protected]>
+ *
+ * @experimental
+ */
+final class PublishMercureUpdatesListener
+{
+    use ClassInfoTrait;
+
+    private $resourceClassResolver;
+    private $iriConverter;
+    private $resourceMetadataFactory;
+    private $serializer;
+    private $messageBus;
+    private $publisher;
+    private $expressionLanguage;
+    private $createdEntities;
+    private $updatedEntities;
+    private $deletedEntities;
+
+    public function __construct(ResourceClassResolverInterface $resourceClassResolver, IriConverterInterface $iriConverter, ResourceMetadataFactoryInterface $resourceMetadataFactory, SerializerInterface $serializer, MessageBusInterface $messageBus = null, callable $publisher = null, ExpressionLanguage $expressionLanguage = null)
+    {
+        if (null === $messageBus && null === $publisher) {
+            throw new InvalidArgumentException('A message bus or a publisher must be provided.');
+        }
+
+        $this->resourceClassResolver = $resourceClassResolver;
+        $this->iriConverter = $iriConverter;
+        $this->resourceMetadataFactory = $resourceMetadataFactory;
+        $this->serializer = $serializer;
+        $this->messageBus = $messageBus;
+        $this->publisher = $publisher;
+        $this->expressionLanguage = $expressionLanguage ?? class_exists(ExpressionLanguage::class) ? new ExpressionLanguage() : null;
+        $this->reset();
+    }
+
+    /**
+     * Collects created, updated and deleted entities.
+     */
+    public function onFlush(OnFlushEventArgs $eventArgs)
+    {
+        $uow = $eventArgs->getEntityManager()->getUnitOfWork();
+
+        foreach ($uow->getScheduledEntityInsertions() as $entity) {
+            $this->storeEntityToPublish($entity, 'createdEntities');
+        }
+
+        foreach ($uow->getScheduledEntityUpdates() as $entity) {
+            $this->storeEntityToPublish($entity, 'updatedEntities');
+        }
+
+        foreach ($uow->getScheduledEntityDeletions() as $entity) {
+            $this->storeEntityToPublish($entity, 'deletedEntities');
+        }
+    }
+
+    /**
+     * Publishes updates for changes collected on flush, and resets the store.
+     */
+    public function postFlush()
+    {
+        try {
+            foreach ($this->createdEntities as $entity) {
+                $this->publishUpdate($entity, $this->createdEntities[$entity]);
+            }
+
+            foreach ($this->updatedEntities as $entity) {
+                $this->publishUpdate($entity, $this->updatedEntities[$entity]);
+            }
+
+            foreach ($this->deletedEntities as $entity) {
+                $this->publishUpdate($entity, $this->deletedEntities[$entity]);
+            }
+        } finally {
+            $this->reset();
+        }
+    }
+
+    private function reset(): void
+    {
+        $this->createdEntities = new \SplObjectStorage();
+        $this->updatedEntities = new \SplObjectStorage();
+        $this->deletedEntities = new \SplObjectStorage();
+    }
+
+    /**
+     * @param object $entity
+     */
+    private function storeEntityToPublish($entity, string $property): void
+    {
+        $resourceClass = $this->getObjectClass($entity);
+        if (!$this->resourceClassResolver->isResourceClass($resourceClass)) {
+            return;
+        }
+
+        $value = $this->resourceMetadataFactory->create($resourceClass)->getAttribute('mercure', false);
+        if (false === $value) {
+            return;
+        }
+
+        if (\is_string($value)) {
+            if (null === $this->expressionLanguage) {
+                throw new RuntimeException('The Expression Language component is not installed. Try running "composer require symfony/expression-language".');
+            }
+
+            $value = $this->expressionLanguage->evaluate($value, ['object' => $entity]);
+        }
+
+        if (true === $value) {
+            $value = [];
+        }
+
+        if (!\is_array($value)) {
+            throw new InvalidArgumentException(sprintf('The value of the "mercure" attribute of the "%s" resource class must be a boolean, an array of targets or a valid expression, "%s" given.', $resourceClass, \gettype($value)));
+        }
+
+        if ('deletedEntities' === $property) {
+            $this->deletedEntities[(object) [
+                'id' => $this->iriConverter->getIriFromItem($entity),
+                'iri' => $this->iriConverter->getIriFromItem($entity, UrlGeneratorInterface::ABS_URL),
+            ]] = $value;
+
+            return;
+        }
+
+        $this->$property[$entity] = $value;
+    }
+
+    /**
+     * @param object|string $entity
+     */
+    private function publishUpdate($entity, array $targets): void
+    {
+        if ($entity instanceof \stdClass) {
+            // By convention, if the entity has been deleted, we send only its IRI
+            // This may change in the feature, because it's not JSON Merge Patch compliant,
+            // and I'm not a fond of this approach
+            $iri = $entity->iri;
+            $data = json_encode(['@id' => $entity->id]);
+        } else {
+            $iri = $this->iriConverter->getIriFromItem($entity, UrlGeneratorInterface::ABS_URL);
+            $data = $this->serializer->serialize($entity, 'jsonld');
+        }
+
+        $update = new Update($iri, $data, $targets);
+        $this->messageBus ? $this->messageBus->dispatch($update) : ($this->publisher)($update);
+    }
+}

src/Bridge/Symfony/Bundle/DependencyInjection/ApiPlatformExtension.php

@@ -144,6 +144,7 @@ public function load(array $configs, ContainerBuilder $container)
         $this->registerHttpCacheConfiguration($container, $config, $loader, $useDoctrine);
         $this->registerValidatorConfiguration($container, $config);
         $this->registerDataCollectorConfiguration($container, $config, $loader);
+        $this->registerMercureConfiguration($container, $config, $loader, $useDoctrine);
     }
 
     /**
@@ -525,4 +526,18 @@ private function registerDataCollectorConfiguration(ContainerBuilder $container,
             $loader->load('debug.xml');
         }
     }
+
+    private function registerMercureConfiguration(ContainerBuilder $container, array $config, XmlFileLoader $loader, bool $useDoctrine)
+    {
+        if (!$config['mercure']['enabled'] || !$container->hasParameter('mercure.default_hub')) {
+            return;
+        }
+
+        $loader->load('mercure.xml');
+        $container->getDefinition('api_platform.mercure.listener.response.add_link_header')->addArgument($config['mercure']['hub_url'] ?? '%mercure.default_hub%');
+
+        if ($useDoctrine) {
+            $loader->load('doctrine_orm_mercure_publisher.xml');
+        }
+    }
 }

src/Bridge/Symfony/Bundle/DependencyInjection/Configuration.php

@@ -23,6 +23,7 @@
 use Symfony\Component\Config\Definition\ConfigurationInterface;
 use Symfony\Component\Config\Definition\Exception\InvalidConfigurationException;
 use Symfony\Component\HttpFoundation\Response;
+use Symfony\Component\Mercure\Update;
 use Symfony\Component\Serializer\Exception\ExceptionInterface;
 
 /**
@@ -218,6 +219,13 @@ public function getConfigTreeBuilder()
                     ->end()
                 ->end()
 
+                ->arrayNode('mercure')
+                    ->{class_exists(Update::class) ? 'canBeDisabled' : 'canBeEnabled'}()
+                    ->children()
+                        ->scalarNode('hub_url')->defaultNull()->info('The URL send in the Link HTTP header. If not set, will default to the URL for the Symfony\'s bundle default hub.')
+                    ->end()
+                ->end()
+
             ->end();
 
         $this->addExceptionToStatusSection($rootNode);

src/Bridge/Symfony/Bundle/Resources/config/doctrine_orm_mercure_publisher.xml

@@ -0,0 +1,25 @@
+<?xml version="1.0" ?>
+
+<container xmlns="http://symfony.com/schema/dic/services"
+           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+           xsi:schemaLocation="http://symfony.com/schema/dic/services http://symfony.com/schema/dic/services/services-1.0.xsd">
+
+    <services>
+
+        <!-- Event listener -->
+
+        <service id="api_platform.doctrine.listener.mercure.publish" class="ApiPlatform\Core\Bridge\Doctrine\EventListener\PublishMercureUpdatesListener">
+            <argument type="service" id="api_platform.resource_class_resolver" />
+            <argument type="service" id="api_platform.iri_converter" />
+            <argument type="service" id="api_platform.metadata.resource.metadata_factory" />
+            <argument type="service" id="api_platform.serializer" />
+            <argument type="service" id="message_bus" on-invalid="ignore" />
+            <argument type="service" id="mercure.hub.default.publisher" />
+
+            <tag name="doctrine.event_listener" event="onFlush" />
+            <tag name="kernel.event_listener" event="kernel.terminate" method="postFlush" />
+        </service>
+
+    </services>
+
+</container>

src/Bridge/Symfony/Bundle/Resources/config/mercure.xml

@@ -0,0 +1,16 @@
+<?xml version="1.0" ?>
+
+<container xmlns="http://symfony.com/schema/dic/services"
+           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+           xsi:schemaLocation="http://symfony.com/schema/dic/services http://symfony.com/schema/dic/services/services-1.0.xsd">
+
+    <services>
+        <!-- Event listener -->
+
+        <service id="api_platform.mercure.listener.response.add_link_header" class="ApiPlatform\Core\Mercure\EventListener\AddLinkHeaderListener">
+            <argument type="service" id="api_platform.metadata.resource.metadata_factory" />
+
+            <tag name="kernel.event_listener" event="kernel.response" method="onKernelResponse" />
+        </service>
+    </services>
+</container>

src/Hydra/EventListener/AddLinkHeaderListener.php

@@ -15,6 +15,8 @@
 
 use ApiPlatform\Core\Api\UrlGeneratorInterface;
 use ApiPlatform\Core\JsonLd\ContextBuilder;
+use Fig\Link\GenericLinkProvider;
+use Fig\Link\Link;
 use Symfony\Component\HttpKernel\Event\FilterResponseEvent;
 
 /**
@@ -36,9 +38,15 @@ public function __construct(UrlGeneratorInterface $urlGenerator)
      */
     public function onKernelResponse(FilterResponseEvent $event)
     {
-        $event->getResponse()->headers->set('Link', sprintf(
-            '<%s>; rel="%sapiDocumentation"',
-            $this->urlGenerator->generate('api_doc', ['_format' => 'jsonld'], UrlGeneratorInterface::ABS_URL), ContextBuilder::HYDRA_NS)
-        );
+        $apiDocUrl = $this->urlGenerator->generate('api_doc', ['_format' => 'jsonld'], UrlGeneratorInterface::ABS_URL);
+        $link = new Link(ContextBuilder::HYDRA_NS.'apiDocumentation', $apiDocUrl);
+
+        $attributes = $event->getRequest()->attributes;
+        if (null === $linkProvider = $attributes->get('_links')) {
+            $attributes->set('_links', new GenericLinkProvider([$link]));
+
+            return;
+        }
+        $attributes->set('_links', $linkProvider->withLink($link));
     }
 }