#28112

pending completion

Build # #28112

Build Type

push

github

Committed by

web-flow

Commit Message

<a href="https://github.com/api-platform/core/commit/<a class=hub.com/api-platform/core/commit/9edf76c34991e1464c20eaccdfa5414eaf3ea662">9edf76c34<a href="https://github.com/api-platform/core/commit/9edf76c34991e1464c20eaccdfa5414eaf3ea662">">Merge </a><a class="double-link" href="https://github.com/api-platform/core/commit/<a class="double-link" href="https://github.com/api-platform/core/commit/be15c6a66686379118a7556df92ba088fd8c377b">be15c6a66</a>">be15c6a66</a><a href="https://github.com/api-platform/core/commit/9edf76c34991e1464c20eaccdfa5414eaf3ea662"> into 8cc452c1f">8cc452c1f</a>

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

13178 of 16537 relevant lines covered (79.69%)

14.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.4

/src/Doctrine/EventListener/PublishMercureUpdatesListener.php

<?php

/*
 * This file is part of the API Platform project.
 *
 * (c) Kévin Dunglas <dunglas@gmail.com>
 *
 * 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\Doctrine\EventListener;

use ApiPlatform\Api\IriConverterInterface;
use ApiPlatform\Core\Api\ResourceClassResolverInterface;
use ApiPlatform\Core\Api\UrlGeneratorInterface;
use ApiPlatform\Exception\InvalidArgumentException;
use ApiPlatform\Exception\OperationNotFoundException;
use ApiPlatform\Exception\RuntimeException;
use ApiPlatform\GraphQl\Subscription\MercureSubscriptionIriGeneratorInterface as GraphQlMercureSubscriptionIriGeneratorInterface;
use ApiPlatform\GraphQl\Subscription\SubscriptionManagerInterface as GraphQlSubscriptionManagerInterface;
use ApiPlatform\Metadata\Resource\Factory\ResourceMetadataCollectionFactoryInterface;
use ApiPlatform\Symfony\Messenger\DispatchTrait;
use ApiPlatform\Util\ResourceClassInfoTrait;
use Doctrine\Common\EventArgs;
use Doctrine\ODM\MongoDB\Event\OnFlushEventArgs as MongoDbOdmOnFlushEventArgs;
use Doctrine\ORM\Event\OnFlushEventArgs as OrmOnFlushEventArgs;
use Symfony\Component\ExpressionLanguage\ExpressionFunction;
use Symfony\Component\ExpressionLanguage\ExpressionLanguage;
use Symfony\Component\HttpFoundation\JsonResponse;
use Symfony\Component\Mercure\HubRegistry;
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 <dunglas@gmail.com>
 *
 * @experimental
 */
final class PublishMercureUpdatesListener
{
    use DispatchTrait;
    use ResourceClassInfoTrait;
    private const ALLOWED_KEYS = [
        'topics' => true,
        'data' => true,
        'private' => true,
        'id' => true,
        'type' => true,
        'retry' => true,
        'normalization_context' => true,
        'hub' => true,
        'enable_async_update' => true,
    ];

    private $iriConverter;
    private $serializer;
    private $hubRegistry;
    private $expressionLanguage;
    private $createdObjects;
    private $updatedObjects;
    private $deletedObjects;
    private $formats;
    private $graphQlSubscriptionManager;
    private $graphQlMercureSubscriptionIriGenerator;

    /**
     * @param array<string, string[]|string> $formats
     * @param HubRegistry|callable           $hubRegistry
     */
    public function __construct(ResourceClassResolverInterface $resourceClassResolver, IriConverterInterface $iriConverter, ResourceMetadataCollectionFactoryInterface $resourceMetadataFactory, SerializerInterface $serializer, array $formats, MessageBusInterface $messageBus = null, $hubRegistry = null, ?GraphQlSubscriptionManagerInterface $graphQlSubscriptionManager = null, ?GraphQlMercureSubscriptionIriGeneratorInterface $graphQlMercureSubscriptionIriGenerator = null, ExpressionLanguage $expressionLanguage = null)
    {
        if (null === $messageBus && null === $hubRegistry) {
            throw new InvalidArgumentException('A message bus or a hub registry must be provided.');
        }

        $this->resourceClassResolver = $resourceClassResolver;
        $this->iriConverter = $iriConverter;

        $this->resourceMetadataFactory = $resourceMetadataFactory;
        $this->serializer = $serializer;
        $this->formats = $formats;
        $this->messageBus = $messageBus;
        $this->hubRegistry = $hubRegistry;
        $this->expressionLanguage = $expressionLanguage ?? (class_exists(ExpressionLanguage::class) ? new ExpressionLanguage() : null);
        $this->graphQlSubscriptionManager = $graphQlSubscriptionManager;
        $this->graphQlMercureSubscriptionIriGenerator = $graphQlMercureSubscriptionIriGenerator;
        $this->reset();

        if ($this->expressionLanguage) {
            $rawurlencode = ExpressionFunction::fromPhp('rawurlencode', 'escape');
            $this->expressionLanguage->addFunction($rawurlencode);

            $this->expressionLanguage->addFunction(
                new ExpressionFunction('iri', static function (string $apiResource, int $referenceType = UrlGeneratorInterface::ABS_URL): string {
                    return sprintf('iri(%s, %d)', $apiResource, $referenceType);
                }, static function (array $arguments, $apiResource, int $referenceType = UrlGeneratorInterface::ABS_URL) use ($iriConverter): string {
                    return $iriConverter->getIriFromItem($apiResource, null, $referenceType);
                })
            );
        }
    }

    /**
     * Collects created, updated and deleted objects.
     */
    public function onFlush(EventArgs $eventArgs): void
    {
        if ($eventArgs instanceof OrmOnFlushEventArgs) {
            $uow = $eventArgs->getEntityManager()->getUnitOfWork();
        } elseif ($eventArgs instanceof MongoDbOdmOnFlushEventArgs) {
            $uow = $eventArgs->getDocumentManager()->getUnitOfWork();
        } else {
            return;
        }

        $methodName = $eventArgs instanceof OrmOnFlushEventArgs ? 'getScheduledEntityInsertions' : 'getScheduledDocumentInsertions';
        foreach ($uow->{$methodName}() as $object) {
            $this->storeObjectToPublish($object, 'createdObjects');
        }

        $methodName = $eventArgs instanceof OrmOnFlushEventArgs ? 'getScheduledEntityUpdates' : 'getScheduledDocumentUpdates';
        foreach ($uow->{$methodName}() as $object) {
            $this->storeObjectToPublish($object, 'updatedObjects');
        }

        $methodName = $eventArgs instanceof OrmOnFlushEventArgs ? 'getScheduledEntityDeletions' : 'getScheduledDocumentDeletions';
        foreach ($uow->{$methodName}() as $object) {
            $this->storeObjectToPublish($object, 'deletedObjects');
        }
    }

    /**
     * Publishes updates for changes collected on flush, and resets the store.
     */
    public function postFlush(): void
    {
        try {
            foreach ($this->createdObjects as $object) {
                $this->publishUpdate($object, $this->createdObjects[$object], 'create');
            }

            foreach ($this->updatedObjects as $object) {
                $this->publishUpdate($object, $this->updatedObjects[$object], 'update');
            }

            foreach ($this->deletedObjects as $object) {
                $this->publishUpdate($object, $this->deletedObjects[$object], 'delete');
            }
        } finally {
            $this->reset();
        }
    }

    private function reset(): void
    {
        $this->createdObjects = new \SplObjectStorage();
        $this->updatedObjects = new \SplObjectStorage();
        $this->deletedObjects = new \SplObjectStorage();
    }

    /**
     * @param object $object
     */
    private function storeObjectToPublish($object, string $property): void
    {
        if (null === $resourceClass = $this->getResourceClass($object)) {
            return;
        }

        try {
            $options = $this->resourceMetadataFactory->create($resourceClass)->getOperation()->getMercure() ?? false;
        } catch (OperationNotFoundException $e) {
            return;
        }

        if (\is_string($options)) {
            if (null === $this->expressionLanguage) {
                throw new RuntimeException('The Expression Language component is not installed. Try running "composer require symfony/expression-language".');
            }

            $options = $this->expressionLanguage->evaluate($options, ['object' => $object]);
        }

        if (false === $options) {
            return;
        }

        if (true === $options) {
            $options = [];
        }

        if (!\is_array($options)) {
            throw new InvalidArgumentException(sprintf('The value of the "mercure" attribute of the "%s" resource class must be a boolean, an array of options or an expression returning this array, "%s" given.', $resourceClass, \gettype($options)));
        }

        foreach ($options as $key => $value) {
            if (0 === $key) {
                if (method_exists(Update::class, 'isPrivate')) {
                    throw new \InvalidArgumentException('Targets do not exist anymore since Mercure 0.10. Mark the update as private instead or downgrade the Mercure Component to version 0.3');
                }

                @trigger_error('Targets do not exist anymore since Mercure 0.10. Mark the update as private instead.', \E_USER_DEPRECATED);
                break;
            }

            if (!isset(self::ALLOWED_KEYS[$key])) {
                throw new InvalidArgumentException(sprintf('The option "%s" set in the "mercure" attribute of the "%s" resource does not exist. Existing options: "%s"', $key, $resourceClass, implode('", "', self::ALLOWED_KEYS)));
            }

            if ('hub' === $key && !$this->hubRegistry instanceof HubRegistry) {
                throw new InvalidArgumentException(sprintf('The option "hub" of the "mercure" attribute cannot be set on the "%s" resource . Try running "composer require symfony/mercure:^0.5".', $resourceClass));
            }
        }

        $options['enable_async_update'] = $options['enable_async_update'] ?? true;

        if ($options['topics'] ?? false) {
            $topics = [];
            foreach ((array) $options['topics'] as $topic) {
                if (!\is_string($topic)) {
                    $topics[] = $topic;
                    continue;
                }

                if (0 !== strpos($topic, '@=')) {
                    $topics[] = $topic;
                    continue;
                }

                if (null === $this->expressionLanguage) {
                    throw new \LogicException('The "@=" expression syntax cannot be used without the Expression Language component. Try running "composer require symfony/expression-language".');
                }

                $topics[] = $this->expressionLanguage->evaluate(substr($topic, 2), ['object' => $object]);
            }

            $options['topics'] = $topics;
        }

        if ('deletedObjects' === $property) {
            $this->deletedObjects[(object) [
                'id' => $this->iriConverter->getIriFromItem($object),
                'iri' => $this->iriConverter->getIriFromItem($object, null, UrlGeneratorInterface::ABS_URL),
            ]] = $options;

            return;
        }

        $this->{$property}[$object] = $options;
    }

    /**
     * @param object $object
     */
    private function publishUpdate($object, array $options, string $type): void
    {
        if ($object instanceof \stdClass) {
            // By convention, if the object 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 = $options['topics'] ?? $object->iri;
            /** @var string $data */
            $data = json_encode(['@id' => $object->id]);
        } else {
            $resourceClass = $this->getObjectClass($object);
            $context = array_replace([
             'resource_class' => $resourceClass,
            ], $options['normalization_context'] ?? $this->resourceMetadataFactory->create($resourceClass)->getOperation()->getNormalizationContext() ?? []);

            $iri = $options['topics'] ?? $this->iriConverter->getIriFromItem($object, null, UrlGeneratorInterface::ABS_URL);
            $data = $options['data'] ?? $this->serializer->serialize($object, key($this->formats), $context);
        }

        $updates = array_merge([$this->buildUpdate($iri, $data, $options)], $this->getGraphQlSubscriptionUpdates($object, $options, $type));

        foreach ($updates as $update) {
            if ($options['enable_async_update'] && $this->messageBus) {
                $this->dispatch($update);
                continue;
            }

            $this->hubRegistry instanceof HubRegistry ? $this->hubRegistry->getHub($options['hub'] ?? null)->publish($update) : ($this->hubRegistry)($update);
        }
    }

    /**
     * @param object $object
     *
     * @return Update[]
     */
    private function getGraphQlSubscriptionUpdates($object, array $options, string $type): array
    {
        if ('update' !== $type || !$this->graphQlSubscriptionManager || !$this->graphQlMercureSubscriptionIriGenerator) {
            return [];
        }

        $payloads = $this->graphQlSubscriptionManager->getPushPayloads($object);

        $updates = [];
        foreach ($payloads as [$subscriptionId, $data]) {
            $updates[] = $this->buildUpdate(
                $this->graphQlMercureSubscriptionIriGenerator->generateTopicIri($subscriptionId),
                (string) (new JsonResponse($data))->getContent(),
                $options
            );
        }

        return $updates;
    }

    /**
     * @param string|string[] $iri
     */
    private function buildUpdate($iri, string $data, array $options): Update
    {
        if (method_exists(Update::class, 'isPrivate')) {
            return new Update($iri, $data, $options['private'] ?? false, $options['id'] ?? null, $options['type'] ?? null, $options['retry'] ?? null);
        }

        // Mercure Component < 0.4.
        /* @phpstan-ignore-next-line */
        return new Update($iri, $data, $options);
    }
}

1	<?php
2
3	/*
4	* This file is part of the API Platform project.
5	*
6	* (c) Kévin Dunglas <dunglas@gmail.com>
7	*
8	* For the full copyright and license information, please view the LICENSE
9	* file that was distributed with this source code.
10	*/
11
12	declare(strict_types=1);
13
14	namespace ApiPlatform\Doctrine\EventListener;
15
16	use ApiPlatform\Api\IriConverterInterface;
17	use ApiPlatform\Core\Api\ResourceClassResolverInterface;
18	use ApiPlatform\Core\Api\UrlGeneratorInterface;
19	use ApiPlatform\Exception\InvalidArgumentException;
20	use ApiPlatform\Exception\OperationNotFoundException;
21	use ApiPlatform\Exception\RuntimeException;
22	use ApiPlatform\GraphQl\Subscription\MercureSubscriptionIriGeneratorInterface as GraphQlMercureSubscriptionIriGeneratorInterface;
23	use ApiPlatform\GraphQl\Subscription\SubscriptionManagerInterface as GraphQlSubscriptionManagerInterface;
24	use ApiPlatform\Metadata\Resource\Factory\ResourceMetadataCollectionFactoryInterface;
25	use ApiPlatform\Symfony\Messenger\DispatchTrait;
26	use ApiPlatform\Util\ResourceClassInfoTrait;
27	use Doctrine\Common\EventArgs;
28	use Doctrine\ODM\MongoDB\Event\OnFlushEventArgs as MongoDbOdmOnFlushEventArgs;
29	use Doctrine\ORM\Event\OnFlushEventArgs as OrmOnFlushEventArgs;
30	use Symfony\Component\ExpressionLanguage\ExpressionFunction;
31	use Symfony\Component\ExpressionLanguage\ExpressionLanguage;
32	use Symfony\Component\HttpFoundation\JsonResponse;
33	use Symfony\Component\Mercure\HubRegistry;
34	use Symfony\Component\Mercure\Update;
35	use Symfony\Component\Messenger\MessageBusInterface;
36	use Symfony\Component\Serializer\SerializerInterface;
37
38	/**
39	* Publishes resources updates to the Mercure hub.
40	*
41	* @author Kévin Dunglas <dunglas@gmail.com>
42	*
43	* @experimental
44	*/
45	final class PublishMercureUpdatesListener
46	{
47	use DispatchTrait;
48	use ResourceClassInfoTrait;
49	private const ALLOWED_KEYS = [
50	'topics' => true,
51	'data' => true,
52	'private' => true,
53	'id' => true,
54	'type' => true,
55	'retry' => true,
56	'normalization_context' => true,
57	'hub' => true,
58	'enable_async_update' => true,
59	];
60
61	private $iriConverter;
62	private $serializer;
63	private $hubRegistry;
64	private $expressionLanguage;
65	private $createdObjects;
66	private $updatedObjects;
67	private $deletedObjects;
68	private $formats;
69	private $graphQlSubscriptionManager;
70	private $graphQlMercureSubscriptionIriGenerator;
71
72	/**
73	* @param array<string, string[]\|string> $formats
74	* @param HubRegistry\|callable $hubRegistry
75	*/
76	public function __construct(ResourceClassResolverInterface $resourceClassResolver, IriConverterInterface $iriConverter, ResourceMetadataCollectionFactoryInterface $resourceMetadataFactory, SerializerInterface $serializer, array $formats, MessageBusInterface $messageBus = null, $hubRegistry = null, ?GraphQlSubscriptionManagerInterface $graphQlSubscriptionManager = null, ?GraphQlMercureSubscriptionIriGeneratorInterface $graphQlMercureSubscriptionIriGenerator = null, ExpressionLanguage $expressionLanguage = null)
77	{
78	if (null === $messageBus && null === $hubRegistry) {	11✔
79	throw new InvalidArgumentException('A message bus or a hub registry must be provided.');	×
80	}
81
82	$this->resourceClassResolver = $resourceClassResolver;	11✔
83	$this->iriConverter = $iriConverter;	11✔
84
85	$this->resourceMetadataFactory = $resourceMetadataFactory;	11✔
86	$this->serializer = $serializer;	11✔
87	$this->formats = $formats;	11✔
88	$this->messageBus = $messageBus;	11✔
89	$this->hubRegistry = $hubRegistry;	11✔
90	$this->expressionLanguage = $expressionLanguage ?? (class_exists(ExpressionLanguage::class) ? new ExpressionLanguage() : null);	11✔
91	$this->graphQlSubscriptionManager = $graphQlSubscriptionManager;	11✔
92	$this->graphQlMercureSubscriptionIriGenerator = $graphQlMercureSubscriptionIriGenerator;	11✔
93	$this->reset();	11✔
94
95	if ($this->expressionLanguage) {	11✔
96	$rawurlencode = ExpressionFunction::fromPhp('rawurlencode', 'escape');	11✔
97	$this->expressionLanguage->addFunction($rawurlencode);	11✔
98
99	$this->expressionLanguage->addFunction(	11✔
100	new ExpressionFunction('iri', static function (string $apiResource, int $referenceType = UrlGeneratorInterface::ABS_URL): string {	11✔
101	return sprintf('iri(%s, %d)', $apiResource, $referenceType);	×
102	}, static function (array $arguments, $apiResource, int $referenceType = UrlGeneratorInterface::ABS_URL) use ($iriConverter): string {	11✔
103	return $iriConverter->getIriFromItem($apiResource, null, $referenceType);	1✔
104	})	11✔
105	);
106	}
107	}	11✔
108
109	/**
110	* Collects created, updated and deleted objects.
111	*/
112	public function onFlush(EventArgs $eventArgs): void
113	{
114	if ($eventArgs instanceof OrmOnFlushEventArgs) {	11✔
115	$uow = $eventArgs->getEntityManager()->getUnitOfWork();	11✔
116	} elseif ($eventArgs instanceof MongoDbOdmOnFlushEventArgs) {	×
117	$uow = $eventArgs->getDocumentManager()->getUnitOfWork();	×
118	} else {
119	return;	×
120	}
121
122	$methodName = $eventArgs instanceof OrmOnFlushEventArgs ? 'getScheduledEntityInsertions' : 'getScheduledDocumentInsertions';	11✔
123	foreach ($uow->{$methodName}() as $object) {	11✔
124	$this->storeObjectToPublish($object, 'createdObjects');	10✔
125	}
126
127	$methodName = $eventArgs instanceof OrmOnFlushEventArgs ? 'getScheduledEntityUpdates' : 'getScheduledDocumentUpdates';	10✔
128	foreach ($uow->{$methodName}() as $object) {	10✔
129	$this->storeObjectToPublish($object, 'updatedObjects');	3✔
130	}
131
132	$methodName = $eventArgs instanceof OrmOnFlushEventArgs ? 'getScheduledEntityDeletions' : 'getScheduledDocumentDeletions';	10✔
133	foreach ($uow->{$methodName}() as $object) {	10✔
134	$this->storeObjectToPublish($object, 'deletedObjects');	2✔
135	}
136	}	10✔
137
138	/**
139	* Publishes updates for changes collected on flush, and resets the store.
140	*/
141	public function postFlush(): void
142	{
143	try {
144	foreach ($this->createdObjects as $object) {	10✔
145	$this->publishUpdate($object, $this->createdObjects[$object], 'create');	2✔
146	}
147
148	foreach ($this->updatedObjects as $object) {	10✔
149	$this->publishUpdate($object, $this->updatedObjects[$object], 'update');	3✔
150	}
151
152	foreach ($this->deletedObjects as $object) {	10✔
153	$this->publishUpdate($object, $this->deletedObjects[$object], 'delete');	2✔
154	}
155	} finally {	10✔
156	$this->reset();	10✔
157	}
158	}	10✔
159
160	private function reset(): void
161	{
162	$this->createdObjects = new \SplObjectStorage();	11✔
163	$this->updatedObjects = new \SplObjectStorage();	11✔
164	$this->deletedObjects = new \SplObjectStorage();	11✔
165	}	11✔
166
167	/**
168	* @param object $object
169	*/
170	private function storeObjectToPublish($object, string $property): void
171	{
172	if (null === $resourceClass = $this->getResourceClass($object)) {	11✔
173	return;	2✔
174	}
175
176	try {
177	$options = $this->resourceMetadataFactory->create($resourceClass)->getOperation()->getMercure() ?? false;	11✔
178	} catch (OperationNotFoundException $e) {	×
179	return;	×
180	}
181
182	if (\is_string($options)) {	11✔
183	if (null === $this->expressionLanguage) {	×
184	throw new RuntimeException('The Expression Language component is not installed. Try running "composer require symfony/expression-language".');	×
185	}
186
187	$options = $this->expressionLanguage->evaluate($options, ['object' => $object]);	×
188	}
189
190	if (false === $options) {	11✔
191	return;	9✔
192	}
193
194	if (true === $options) {	4✔
195	$options = [];	1✔
196	}
197
198	if (!\is_array($options)) {	4✔
199	throw new InvalidArgumentException(sprintf('The value of the "mercure" attribute of the "%s" resource class must be a boolean, an array of options or an expression returning this array, "%s" given.', $resourceClass, \gettype($options)));	1✔
200	}
201
202	foreach ($options as $key => $value) {	3✔
203	if (0 === $key) {	3✔
204	if (method_exists(Update::class, 'isPrivate')) {	×
205	throw new \InvalidArgumentException('Targets do not exist anymore since Mercure 0.10. Mark the update as private instead or downgrade the Mercure Component to version 0.3');	×
206	}
207
208	@trigger_error('Targets do not exist anymore since Mercure 0.10. Mark the update as private instead.', \E_USER_DEPRECATED);	×
209	break;	×
210	}
211
212	if (!isset(self::ALLOWED_KEYS[$key])) {	3✔
213	throw new InvalidArgumentException(sprintf('The option "%s" set in the "mercure" attribute of the "%s" resource does not exist. Existing options: "%s"', $key, $resourceClass, implode('", "', self::ALLOWED_KEYS)));	×
214	}
215
216	if ('hub' === $key && !$this->hubRegistry instanceof HubRegistry) {	3✔
217	throw new InvalidArgumentException(sprintf('The option "hub" of the "mercure" attribute cannot be set on the "%s" resource . Try running "composer require symfony/mercure:^0.5".', $resourceClass));	×
218	}
219	}
220
221	$options['enable_async_update'] = $options['enable_async_update'] ?? true;	3✔
222
223	if ($options['topics'] ?? false) {	3✔
224	$topics = [];	2✔
225	foreach ((array) $options['topics'] as $topic) {	2✔
226	if (!\is_string($topic)) {	2✔
227	$topics[] = $topic;	×
228	continue;	×
229	}
230
231	if (0 !== strpos($topic, '@=')) {	2✔
232	$topics[] = $topic;	2✔
233	continue;	2✔
234	}
235
236	if (null === $this->expressionLanguage) {	1✔
237	throw new \LogicException('The "@=" expression syntax cannot be used without the Expression Language component. Try running "composer require symfony/expression-language".');	×
238	}
239
240	$topics[] = $this->expressionLanguage->evaluate(substr($topic, 2), ['object' => $object]);	1✔
241	}
242
243	$options['topics'] = $topics;	2✔
244	}
245
246	if ('deletedObjects' === $property) {	3✔
247	$this->deletedObjects[(object) [	2✔
248	'id' => $this->iriConverter->getIriFromItem($object),	2✔
249	'iri' => $this->iriConverter->getIriFromItem($object, null, UrlGeneratorInterface::ABS_URL),	2✔
250	]] = $options;
251
252	return;	2✔
253	}
254
255	$this->{$property}[$object] = $options;	3✔
256	}	3✔
257
258	/**
259	* @param object $object
260	*/
261	private function publishUpdate($object, array $options, string $type): void
262	{
263	if ($object instanceof \stdClass) {	3✔
264	// By convention, if the object has been deleted, we send only its IRI.
265	// This may change in the feature, because it's not JSON Merge Patch compliant,
266	// and I'm not a fond of this approach.
267	$iri = $options['topics'] ?? $object->iri;	2✔
268	/** @var string $data */
269	$data = json_encode(['@id' => $object->id]);	2✔
270	} else {
271	$resourceClass = $this->getObjectClass($object);	3✔
272	$context = array_replace([	3✔
273	'resource_class' => $resourceClass,	3✔
274	], $options['normalization_context'] ?? $this->resourceMetadataFactory->create($resourceClass)->getOperation()->getNormalizationContext() ?? []);	3✔
275
276	$iri = $options['topics'] ?? $this->iriConverter->getIriFromItem($object, null, UrlGeneratorInterface::ABS_URL);	3✔
277	$data = $options['data'] ?? $this->serializer->serialize($object, key($this->formats), $context);	3✔
278	}
279
280	$updates = array_merge([$this->buildUpdate($iri, $data, $options)], $this->getGraphQlSubscriptionUpdates($object, $options, $type));	3✔
281
282	foreach ($updates as $update) {	3✔
283	if ($options['enable_async_update'] && $this->messageBus) {	3✔
284	$this->dispatch($update);	×
285	continue;	×
286	}
287
288	$this->hubRegistry instanceof HubRegistry ? $this->hubRegistry->getHub($options['hub'] ?? null)->publish($update) : ($this->hubRegistry)($update);	3✔
289	}
290	}	3✔
291
292	/**
293	* @param object $object
294	*
295	* @return Update[]
296	*/
297	private function getGraphQlSubscriptionUpdates($object, array $options, string $type): array
298	{
299	if ('update' !== $type \|\| !$this->graphQlSubscriptionManager \|\| !$this->graphQlMercureSubscriptionIriGenerator) {	3✔
300	return [];	2✔
301	}
302
303	$payloads = $this->graphQlSubscriptionManager->getPushPayloads($object);	1✔
304
305	$updates = [];	1✔
306	foreach ($payloads as [$subscriptionId, $data]) {	1✔
307	$updates[] = $this->buildUpdate(	1✔
308	$this->graphQlMercureSubscriptionIriGenerator->generateTopicIri($subscriptionId),	1✔
309	(string) (new JsonResponse($data))->getContent(),	1✔
310	$options
311	);
312	}
313
314	return $updates;	1✔
315	}
316
317	/**
318	* @param string\|string[] $iri
319	*/
320	private function buildUpdate($iri, string $data, array $options): Update
321	{
322	if (method_exists(Update::class, 'isPrivate')) {	3✔
323	return new Update($iri, $data, $options['private'] ?? false, $options['id'] ?? null, $options['type'] ?? null, $options['retry'] ?? null);	3✔
324	}
325
326	// Mercure Component < 0.4.
327	/* @phpstan-ignore-next-line */
328	return new Update($iri, $data, $options);	×
329	}
330	}

api-platform / core / #28112

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous