• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

api-platform / core / 14375359694

10 Apr 2025 08:01AM UTC coverage: 8.491% (+0.07%) from 8.425%
14375359694

push

github

web-flow
fix(hydra): use correctly enable_docs (#7062)

18 of 118 new or added lines in 5 files covered. (15.25%)

508 existing lines in 190 files now uncovered.

13401 of 157825 relevant lines covered (8.49%)

22.87 hits per line

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

74.19
/src/Symfony/Doctrine/EventListener/PublishMercureUpdatesListener.php
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\Symfony\Doctrine\EventListener;
15

16
use ApiPlatform\Doctrine\Common\Messenger\DispatchTrait;
17
use ApiPlatform\GraphQl\Subscription\MercureSubscriptionIriGeneratorInterface as GraphQlMercureSubscriptionIriGeneratorInterface;
18
use ApiPlatform\GraphQl\Subscription\SubscriptionManagerInterface as GraphQlSubscriptionManagerInterface;
19
use ApiPlatform\Metadata\Exception\InvalidArgumentException;
20
use ApiPlatform\Metadata\Exception\OperationNotFoundException;
21
use ApiPlatform\Metadata\Exception\RuntimeException;
22
use ApiPlatform\Metadata\HttpOperation;
23
use ApiPlatform\Metadata\IriConverterInterface;
24
use ApiPlatform\Metadata\Operation;
25
use ApiPlatform\Metadata\Resource\Factory\ResourceMetadataCollectionFactoryInterface;
26
use ApiPlatform\Metadata\ResourceClassResolverInterface;
27
use ApiPlatform\Metadata\UrlGeneratorInterface;
28
use ApiPlatform\Metadata\Util\ResourceClassInfoTrait;
29
use Doctrine\Common\EventArgs;
30
use Doctrine\ODM\MongoDB\Event\OnFlushEventArgs as MongoDbOdmOnFlushEventArgs;
31
use Doctrine\ORM\Event\OnFlushEventArgs as OrmOnFlushEventArgs;
32
use Symfony\Component\ExpressionLanguage\ExpressionFunction;
33
use Symfony\Component\ExpressionLanguage\ExpressionLanguage;
34
use Symfony\Component\HttpFoundation\JsonResponse;
35
use Symfony\Component\Mercure\HubRegistry;
36
use Symfony\Component\Mercure\Update;
37
use Symfony\Component\Messenger\MessageBusInterface;
38
use Symfony\Component\Serializer\SerializerInterface;
39

40
/**
41
 * Publishes resources updates to the Mercure hub.
42
 *
43
 * @author Kévin Dunglas <dunglas@gmail.com>
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
    private readonly ?ExpressionLanguage $expressionLanguage;
61
    private \SplObjectStorage $createdObjects;
62
    private \SplObjectStorage $updatedObjects;
63
    private \SplObjectStorage $deletedObjects;
64

65
    /**
66
     * @param array<string, string[]|string> $formats
67
     */
68
    public function __construct(ResourceClassResolverInterface $resourceClassResolver, private readonly IriConverterInterface $iriConverter, ResourceMetadataCollectionFactoryInterface $resourceMetadataFactory, private readonly SerializerInterface $serializer, private readonly array $formats, ?MessageBusInterface $messageBus = null, private readonly ?HubRegistry $hubRegistry = null, private readonly ?GraphQlSubscriptionManagerInterface $graphQlSubscriptionManager = null, private readonly ?GraphQlMercureSubscriptionIriGeneratorInterface $graphQlMercureSubscriptionIriGenerator = null, ?ExpressionLanguage $expressionLanguage = null, private bool $includeType = false)
69
    {
70
        if (null === $messageBus && null === $hubRegistry) {
1,013✔
71
            throw new InvalidArgumentException('A message bus or a hub registry must be provided.');
×
72
        }
73

74
        $this->resourceClassResolver = $resourceClassResolver;
1,013✔
75

76
        $this->resourceMetadataFactory = $resourceMetadataFactory;
1,013✔
77
        $this->messageBus = $messageBus;
1,013✔
78
        $this->expressionLanguage = $expressionLanguage ?? (class_exists(ExpressionLanguage::class) ? new ExpressionLanguage() : null);
1,013✔
79
        $this->reset();
1,013✔
80

81
        if ($this->expressionLanguage) {
1,013✔
82
            $rawurlencode = ExpressionFunction::fromPhp('rawurlencode', 'escape');
1,013✔
83
            $this->expressionLanguage->addFunction($rawurlencode);
1,013✔
84

85
            $this->expressionLanguage->addFunction(
1,013✔
86
                new ExpressionFunction('get_operation', static fn (string $apiResource, string $name): string => \sprintf('getOperation(%s, %s)', $apiResource, $name), static fn (array $arguments, $apiResource, string $name): Operation => $resourceMetadataFactory->create($resourceClassResolver->getResourceClass($apiResource))->getOperation($name))
1,013✔
87
            );
1,013✔
88
            $this->expressionLanguage->addFunction(
1,013✔
89
                new ExpressionFunction('iri', static fn (string $apiResource, int $referenceType = UrlGeneratorInterface::ABS_URL, ?string $operation = null): string => \sprintf('iri(%s, %d, %s)', $apiResource, $referenceType, $operation), static fn (array $arguments, $apiResource, int $referenceType = UrlGeneratorInterface::ABS_URL, $operation = null): string => $iriConverter->getIriFromResource($apiResource, $referenceType, $operation))
1,013✔
90
            );
1,013✔
91
        }
92

93
        if (false === $this->includeType) {
1,013✔
94
            trigger_deprecation('api-platform/core', '3.1', 'Having mercure.include_type (always include @type in Mercure updates, even delete ones) set to false in the configuration is deprecated. It will be true by default in API Platform 4.0.');
×
95
        }
96
    }
97

98
    /**
99
     * Collects created, updated and deleted objects.
100
     */
101
    public function onFlush(EventArgs $eventArgs): void
102
    {
103
        if ($eventArgs instanceof OrmOnFlushEventArgs) {
1,013✔
104
            // @phpstan-ignore-next-line
105
            $uow = method_exists($eventArgs, 'getObjectManager') ? $eventArgs->getObjectManager()->getUnitOfWork() : $eventArgs->getEntityManager()->getUnitOfWork();
658✔
UNCOV
106
        } elseif ($eventArgs instanceof MongoDbOdmOnFlushEventArgs) {
355✔
UNCOV
107
            $uow = $eventArgs->getDocumentManager()->getUnitOfWork();
355✔
108
        } else {
109
            return;
×
110
        }
111

112
        $methodName = $eventArgs instanceof OrmOnFlushEventArgs ? 'getScheduledEntityInsertions' : 'getScheduledDocumentInsertions';
1,013✔
113
        foreach ($uow->{$methodName}() as $object) {
1,013✔
114
            $this->storeObjectToPublish($object, 'createdObjects');
922✔
115
        }
116

117
        $methodName = $eventArgs instanceof OrmOnFlushEventArgs ? 'getScheduledEntityUpdates' : 'getScheduledDocumentUpdates';
1,013✔
118
        foreach ($uow->{$methodName}() as $object) {
1,013✔
UNCOV
119
            $this->storeObjectToPublish($object, 'updatedObjects');
94✔
120
        }
121

122
        $methodName = $eventArgs instanceof OrmOnFlushEventArgs ? 'getScheduledEntityDeletions' : 'getScheduledDocumentDeletions';
1,013✔
123
        foreach ($uow->{$methodName}() as $object) {
1,013✔
UNCOV
124
            $this->storeObjectToPublish($object, 'deletedObjects');
20✔
125
        }
126
    }
127

128
    /**
129
     * Publishes updates for changes collected on flush, and resets the store.
130
     */
131
    public function postFlush(): void
132
    {
133
        try {
134
            $creatingObjects = clone $this->createdObjects;
1,013✔
135
            foreach ($creatingObjects as $object) {
1,013✔
UNCOV
136
                if ($this->createdObjects->contains($object)) {
8✔
UNCOV
137
                    $this->createdObjects->detach($object);
8✔
138
                }
UNCOV
139
                $this->publishUpdate($object, $creatingObjects[$object], 'create');
8✔
140
            }
141

142
            $updatingObjects = clone $this->updatedObjects;
1,013✔
143
            foreach ($updatingObjects as $object) {
1,013✔
UNCOV
144
                if ($this->updatedObjects->contains($object)) {
2✔
UNCOV
145
                    $this->updatedObjects->detach($object);
2✔
146
                }
UNCOV
147
                $this->publishUpdate($object, $updatingObjects[$object], 'update');
2✔
148
            }
149

150
            $deletingObjects = clone $this->deletedObjects;
1,013✔
151
            foreach ($deletingObjects as $object) {
1,013✔
152
                $options = $this->deletedObjects[$object];
×
153
                if ($this->deletedObjects->contains($object)) {
×
154
                    $this->deletedObjects->detach($object);
×
155
                }
156
                $this->publishUpdate($object, $deletingObjects[$object], 'delete');
×
157
            }
158
        } finally {
159
            $this->reset();
1,013✔
160
        }
161
    }
162

163
    private function reset(): void
164
    {
165
        $this->createdObjects = new \SplObjectStorage();
1,013✔
166
        $this->updatedObjects = new \SplObjectStorage();
1,013✔
167
        $this->deletedObjects = new \SplObjectStorage();
1,013✔
168
    }
169

170
    private function storeObjectToPublish(object $object, string $property): void
171
    {
172
        if (null === $resourceClass = $this->getResourceClass($object)) {
996✔
173
            return;
88✔
174
        }
175

176
        $operation = $this->resourceMetadataFactory->create($resourceClass)->getOperation();
981✔
177
        try {
178
            $options = $operation->getMercure() ?? false;
981✔
179
        } catch (OperationNotFoundException) {
×
180
            return;
×
181
        }
182

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

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

191
        if (false === $options) {
981✔
192
            return;
975✔
193
        }
194

UNCOV
195
        if (true === $options) {
10✔
UNCOV
196
            $options = [];
6✔
197
        }
198

UNCOV
199
        if (!\is_array($options)) {
10✔
200
            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)));
×
201
        }
202

UNCOV
203
        foreach ($options as $key => $value) {
10✔
UNCOV
204
            if (!isset(self::ALLOWED_KEYS[$key])) {
4✔
205
                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('", "', array_keys(self::ALLOWED_KEYS))));
×
206
            }
207
        }
208

UNCOV
209
        $options['enable_async_update'] ??= true;
10✔
210

UNCOV
211
        if ('deletedObjects' === $property) {
10✔
212
            $types = $operation instanceof HttpOperation ? $operation->getTypes() : null;
×
213
            if (null === $types) {
×
214
                $types = [$operation->getShortName()];
×
215
            }
216

217
            // We need to evaluate it here, because in publishUpdate() the resource would be already deleted
218
            $this->evaluateTopics($options, $object);
×
219

220
            $this->deletedObjects[(object) [
×
221
                'id' => $this->iriConverter->getIriFromResource($object),
×
222
                'iri' => $this->iriConverter->getIriFromResource($object, UrlGeneratorInterface::ABS_URL),
×
223
                'type' => 1 === \count($types) ? $types[0] : $types,
×
224
            ]] = $options;
×
225

226
            return;
×
227
        }
228

UNCOV
229
        $this->{$property}[$object] = $options;
10✔
230
    }
231

232
    private function publishUpdate(object $object, array $options, string $type): void
233
    {
UNCOV
234
        if ($object instanceof \stdClass) {
10✔
235
            // By convention, if the object has been deleted, we send only its IRI and its type.
236
            // This may change in the feature, because it's not JSON Merge Patch compliant,
237
            // and I'm not a fond of this approach.
238
            $iri = $options['topics'] ?? $object->iri;
×
239
            /** @var string $data */
240
            $data = json_encode(['@id' => $object->id] + ($this->includeType ? ['@type' => $object->type] : []), \JSON_THROW_ON_ERROR);
×
241
        } else {
UNCOV
242
            $resourceClass = $this->getObjectClass($object);
10✔
UNCOV
243
            $context = $options['normalization_context'] ?? $this->resourceMetadataFactory->create($resourceClass)->getOperation()->getNormalizationContext() ?? [];
10✔
244

245
            // We need to evaluate it here, because in storeObjectToPublish() the resource would not have been persisted yet
UNCOV
246
            $this->evaluateTopics($options, $object);
10✔
247

UNCOV
248
            $iri = $options['topics'] ?? $this->iriConverter->getIriFromResource($object, UrlGeneratorInterface::ABS_URL);
10✔
UNCOV
249
            $data = $options['data'] ?? $this->serializer->serialize($object, key($this->formats), $context);
10✔
250
        }
251

UNCOV
252
        $updates = array_merge([$this->buildUpdate($iri, $data, $options)], $this->getGraphQlSubscriptionUpdates($object, $options, $type));
10✔
UNCOV
253
        foreach ($updates as $update) {
10✔
UNCOV
254
            if ($options['enable_async_update'] && $this->messageBus) {
10✔
UNCOV
255
                $this->dispatch($update);
10✔
UNCOV
256
                continue;
10✔
257
            }
258

259
            $this->hubRegistry->getHub($options['hub'] ?? null)->publish($update);
×
260
        }
261
    }
262

263
    private function evaluateTopics(array &$options, object $object): void
264
    {
UNCOV
265
        if (!($options['topics'] ?? false)) {
10✔
UNCOV
266
            return;
6✔
267
        }
268

UNCOV
269
        $topics = [];
4✔
UNCOV
270
        foreach ((array) $options['topics'] as $topic) {
4✔
UNCOV
271
            if (!\is_string($topic)) {
4✔
272
                $topics[] = $topic;
×
273
                continue;
×
274
            }
275

UNCOV
276
            if (!str_starts_with($topic, '@=')) {
4✔
277
                $topics[] = $topic;
×
278
                continue;
×
279
            }
280

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

UNCOV
285
            $topics[] = $this->expressionLanguage->evaluate(substr($topic, 2), ['object' => $object]);
4✔
286
        }
287

UNCOV
288
        $options['topics'] = $topics;
4✔
289
    }
290

291
    /**
292
     * @return Update[]
293
     */
294
    private function getGraphQlSubscriptionUpdates(object $object, array $options, string $type): array
295
    {
UNCOV
296
        if ('update' !== $type || !$this->graphQlSubscriptionManager || !$this->graphQlMercureSubscriptionIriGenerator) {
10✔
UNCOV
297
            return [];
8✔
298
        }
299

UNCOV
300
        $payloads = $this->graphQlSubscriptionManager->getPushPayloads($object);
2✔
301

UNCOV
302
        $updates = [];
2✔
UNCOV
303
        foreach ($payloads as [$subscriptionId, $data]) {
2✔
UNCOV
304
            $updates[] = $this->buildUpdate(
2✔
UNCOV
305
                $this->graphQlMercureSubscriptionIriGenerator->generateTopicIri($subscriptionId),
2✔
UNCOV
306
                (string) (new JsonResponse($data))->getContent(),
2✔
UNCOV
307
                $options
2✔
UNCOV
308
            );
2✔
309
        }
310

UNCOV
311
        return $updates;
2✔
312
    }
313

314
    /**
315
     * @param string|string[] $iri
316
     */
317
    private function buildUpdate(string|array $iri, string $data, array $options): Update
318
    {
UNCOV
319
        return new Update($iri, $data, $options['private'] ?? false, $options['id'] ?? null, $options['type'] ?? null, $options['retry'] ?? null);
10✔
320
    }
321
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc