Merge branch 'release/0.1.0'
This commit is contained in:
commit
c981d571a5
22 changed files with 260 additions and 197 deletions
8
.woodpecker/code-style.yml
Normal file
8
.woodpecker/code-style.yml
Normal file
|
@ -0,0 +1,8 @@
|
||||||
|
pipeline:
|
||||||
|
code-style:
|
||||||
|
image: composer
|
||||||
|
commands:
|
||||||
|
- composer global config repositories.repo-name vcs https://git.digital-competence.de/Packages/php-codesniffer
|
||||||
|
- composer global config --no-plugins allow-plugins.dealerdirect/phpcodesniffer-composer-installer true
|
||||||
|
- composer global require digicomp/php-codesniffer:@dev
|
||||||
|
- composer global exec -- phpcs --runtime-set ignore_warnings_on_exit 1 --standard=DigiComp Classes/ Tests/
|
27
.woodpecker/functional-tests.yml
Normal file
27
.woodpecker/functional-tests.yml
Normal file
|
@ -0,0 +1,27 @@
|
||||||
|
workspace:
|
||||||
|
base: /woodpecker
|
||||||
|
path: package
|
||||||
|
|
||||||
|
matrix:
|
||||||
|
include:
|
||||||
|
- FLOW_VERSION: 8.2
|
||||||
|
PHP_VERSION: 8.1
|
||||||
|
|
||||||
|
pipeline:
|
||||||
|
functional-tests:
|
||||||
|
image: "thecodingmachine/php:${PHP_VERSION}-v4-cli"
|
||||||
|
environment:
|
||||||
|
# Enable the PDO_SQLITE extension
|
||||||
|
- "PHP_EXTENSION_PDO_SQLITE=1"
|
||||||
|
- "FLOW_VERSION=${FLOW_VERSION}"
|
||||||
|
- "NEOS_BUILD_DIR=/woodpecker/Build-${FLOW_VERSION}"
|
||||||
|
commands:
|
||||||
|
- "sudo mkdir $NEOS_BUILD_DIR"
|
||||||
|
- "sudo chown -R docker:docker $NEOS_BUILD_DIR"
|
||||||
|
- "cd $NEOS_BUILD_DIR"
|
||||||
|
- "composer create-project --no-install neos/flow-base-distribution:^$FLOW_VERSION ."
|
||||||
|
- "composer config repositories.repo-name path /woodpecker/package"
|
||||||
|
- "composer config --no-plugins allow-plugins.neos/composer-plugin true"
|
||||||
|
- "composer remove --dev --no-update neos/behat || composer remove --no-update neos/behat"
|
||||||
|
- "composer require digicomp/flow-symfony-bridge-messenger:@dev"
|
||||||
|
- "bin/phpunit --configuration Build/BuildEssentials/PhpUnit/FunctionalTests.xml Packages/Application/DigiComp.FlowSymfonyBridge.Messenger/Tests/Functional"
|
|
@ -14,16 +14,10 @@ class FailedCommandController extends CommandController
|
||||||
{
|
{
|
||||||
use RunSymfonyCommandTrait;
|
use RunSymfonyCommandTrait;
|
||||||
|
|
||||||
/**
|
#[Flow\Inject(name: 'DigiComp.FlowSymfonyBridge.Messenger:ReceiversContainer')]
|
||||||
* @Flow\Inject(name="DigiComp.FlowSymfonyBridge.Messenger:ReceiversContainer")
|
protected ContainerInterface $receiverContainer;
|
||||||
* @var ContainerInterface
|
|
||||||
*/
|
|
||||||
protected $receiverContainer;
|
|
||||||
|
|
||||||
/**
|
#[Flow\InjectConfiguration]
|
||||||
* @Flow\InjectConfiguration
|
|
||||||
* @var array
|
|
||||||
*/
|
|
||||||
protected array $configuration;
|
protected array $configuration;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -39,7 +33,7 @@ class FailedCommandController extends CommandController
|
||||||
*
|
*
|
||||||
* Optional arguments are -q (quiet) -v[v[v]] (verbosity) and --force (do not ask)
|
* Optional arguments are -q (quiet) -v[v[v]] (verbosity) and --force (do not ask)
|
||||||
*/
|
*/
|
||||||
public function showCommand()
|
public function showCommand(): void
|
||||||
{
|
{
|
||||||
$command = new FailedMessagesShowCommand(
|
$command = new FailedMessagesShowCommand(
|
||||||
$this->configuration['failureTransport'],
|
$this->configuration['failureTransport'],
|
||||||
|
@ -59,7 +53,7 @@ class FailedCommandController extends CommandController
|
||||||
*
|
*
|
||||||
* Optional arguments are -q (quiet) -v[v[v]] (verbosity) and --force (do not ask)
|
* Optional arguments are -q (quiet) -v[v[v]] (verbosity) and --force (do not ask)
|
||||||
*/
|
*/
|
||||||
public function removeCommand()
|
public function removeCommand(): void
|
||||||
{
|
{
|
||||||
$command = new FailedMessagesRemoveCommand(
|
$command = new FailedMessagesRemoveCommand(
|
||||||
$this->configuration['failureTransport'],
|
$this->configuration['failureTransport'],
|
||||||
|
@ -87,7 +81,7 @@ class FailedCommandController extends CommandController
|
||||||
*
|
*
|
||||||
* @noinspection PhpParamsInspection
|
* @noinspection PhpParamsInspection
|
||||||
*/
|
*/
|
||||||
public function retryCommand()
|
public function retryCommand(): void
|
||||||
{
|
{
|
||||||
$command = new FailedMessagesRetryCommand(
|
$command = new FailedMessagesRetryCommand(
|
||||||
$this->configuration['failureTransport'],
|
$this->configuration['failureTransport'],
|
||||||
|
|
|
@ -17,47 +17,29 @@ class MessengerCommandController extends CommandController
|
||||||
{
|
{
|
||||||
use RunSymfonyCommandTrait;
|
use RunSymfonyCommandTrait;
|
||||||
|
|
||||||
/**
|
#[Flow\Inject(name: 'DigiComp.FlowSymfonyBridge.Messenger:RoutableMessageBus')]
|
||||||
* @Flow\Inject(name="DigiComp.FlowSymfonyBridge.Messenger:RoutableMessageBus")
|
protected RoutableMessageBus $routableBus;
|
||||||
* @var RoutableMessageBus
|
|
||||||
*/
|
|
||||||
protected $routableBus;
|
|
||||||
|
|
||||||
/**
|
#[Flow\Inject(name: 'DigiComp.FlowSymfonyBridge.Messenger:ReceiversContainer')]
|
||||||
* @Flow\Inject(name="DigiComp.FlowSymfonyBridge.Messenger:ReceiversContainer")
|
protected ContainerInterface $receiverContainer;
|
||||||
* @var ContainerInterface
|
|
||||||
*/
|
|
||||||
protected $receiverContainer;
|
|
||||||
|
|
||||||
/**
|
#[Flow\Inject(name: 'DigiComp.FlowSymfonyBridge.Messenger:EventDispatcher')]
|
||||||
* @Flow\Inject(name="DigiComp.FlowSymfonyBridge.Messenger:EventDispatcher")
|
protected EventDispatcherInterface $eventDispatcher;
|
||||||
* @var EventDispatcherInterface
|
|
||||||
*/
|
|
||||||
protected $eventDispatcher;
|
|
||||||
|
|
||||||
/**
|
#[Flow\Inject]
|
||||||
* @Flow\Inject(lazy=false)
|
|
||||||
* @var LoggerInterface
|
|
||||||
*/
|
|
||||||
protected LoggerInterface $logger;
|
protected LoggerInterface $logger;
|
||||||
|
|
||||||
/**
|
#[Flow\InjectConfiguration]
|
||||||
* @Flow\InjectConfiguration
|
|
||||||
* @var array
|
|
||||||
*/
|
|
||||||
protected array $configuration;
|
protected array $configuration;
|
||||||
|
|
||||||
/**
|
#[Flow\Inject(name: 'DigiComp.FlowSymfonyBridge.Messenger:RestartSignalCachePool')]
|
||||||
* @Flow\Inject(name="DigiComp.FlowSymfonyBridge.Messenger:RestartSignalCachePool")
|
protected CacheItemPoolInterface $restartSignalCachePool;
|
||||||
* @var CacheItemPoolInterface
|
|
||||||
*/
|
|
||||||
protected $restartSignalCachePool;
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Consumes messages and dispatches them to the message bus
|
* Consumes messages and dispatches them to the message bus
|
||||||
*
|
*
|
||||||
* To receive from multiple transports, pass each name:
|
* To receive from multiple transports, pass each name:
|
||||||
* <info>worker:consume receiver1 receiver2</info>
|
* <info>messenger:consume receiver1 receiver2</info>
|
||||||
*
|
*
|
||||||
* Options are:
|
* Options are:
|
||||||
* --limit limits the number of messages received
|
* --limit limits the number of messages received
|
||||||
|
@ -71,7 +53,7 @@ class MessengerCommandController extends CommandController
|
||||||
*
|
*
|
||||||
* Optional arguments are -q (quiet) and -v[v[v]] (verbosity)
|
* Optional arguments are -q (quiet) and -v[v[v]] (verbosity)
|
||||||
*/
|
*/
|
||||||
public function consumeCommand()
|
public function consumeCommand(): void
|
||||||
{
|
{
|
||||||
if ($this->receiverContainer instanceof DependencyProxy) {
|
if ($this->receiverContainer instanceof DependencyProxy) {
|
||||||
$this->receiverContainer->_activateDependency();
|
$this->receiverContainer->_activateDependency();
|
||||||
|
@ -92,7 +74,7 @@ class MessengerCommandController extends CommandController
|
||||||
/**
|
/**
|
||||||
* List all available receivers
|
* List all available receivers
|
||||||
*/
|
*/
|
||||||
public function listReceiversCommand()
|
public function listReceiversCommand(): void
|
||||||
{
|
{
|
||||||
foreach (\array_keys($this->configuration['transports']) as $transportName) {
|
foreach (\array_keys($this->configuration['transports']) as $transportName) {
|
||||||
$this->outputLine('- ' . $transportName);
|
$this->outputLine('- ' . $transportName);
|
||||||
|
@ -106,7 +88,7 @@ class MessengerCommandController extends CommandController
|
||||||
* and then exit. Worker commands are *not* automatically restarted: that
|
* and then exit. Worker commands are *not* automatically restarted: that
|
||||||
* should be handled by a process control system.
|
* should be handled by a process control system.
|
||||||
*/
|
*/
|
||||||
public function stopWorkersCommand()
|
public function stopWorkersCommand(): void
|
||||||
{
|
{
|
||||||
$cacheItem = $this->restartSignalCachePool->getItem(
|
$cacheItem = $this->restartSignalCachePool->getItem(
|
||||||
StopWorkerOnRestartSignalListener::RESTART_REQUESTED_TIMESTAMP_KEY
|
StopWorkerOnRestartSignalListener::RESTART_REQUESTED_TIMESTAMP_KEY
|
||||||
|
|
|
@ -10,7 +10,7 @@ use Symfony\Component\Console\Output\OutputInterface;
|
||||||
|
|
||||||
trait RunSymfonyCommandTrait
|
trait RunSymfonyCommandTrait
|
||||||
{
|
{
|
||||||
protected function run(Command $command)
|
protected function run(Command $command): void
|
||||||
{
|
{
|
||||||
$definition = $command->getDefinition();
|
$definition = $command->getDefinition();
|
||||||
$definition->setArguments(\array_merge(
|
$definition->setArguments(\array_merge(
|
||||||
|
@ -29,7 +29,7 @@ trait RunSymfonyCommandTrait
|
||||||
$command->run($input, $this->output->getOutput());
|
$command->run($input, $this->output->getOutput());
|
||||||
}
|
}
|
||||||
|
|
||||||
protected function configureIO($input, $output)
|
protected function configureIO($input, $output): void
|
||||||
{
|
{
|
||||||
switch ($shellVerbosity = (int)\getenv('SHELL_VERBOSITY')) {
|
switch ($shellVerbosity = (int)\getenv('SHELL_VERBOSITY')) {
|
||||||
case -1:
|
case -1:
|
||||||
|
|
|
@ -11,19 +11,13 @@ use Symfony\Component\EventDispatcher\EventSubscriberInterface;
|
||||||
// TODO: Maybe an own package for EntityManager bridge?
|
// TODO: Maybe an own package for EntityManager bridge?
|
||||||
class EventDispatcherFactory
|
class EventDispatcherFactory
|
||||||
{
|
{
|
||||||
/**
|
#[Flow\Inject(lazy: false)]
|
||||||
* @Flow\Inject
|
protected ObjectManagerInterface $objectManager;
|
||||||
* @var ObjectManagerInterface
|
|
||||||
*/
|
|
||||||
protected $objectManager;
|
|
||||||
|
|
||||||
/**
|
#[Flow\InjectConfiguration]
|
||||||
* @Flow\InjectConfiguration
|
protected array $configuration;
|
||||||
* @var array
|
|
||||||
*/
|
|
||||||
protected $configuration;
|
|
||||||
|
|
||||||
public function create()
|
public function create(): EventDispatcher
|
||||||
{
|
{
|
||||||
$eventDispatcher = new EventDispatcher();
|
$eventDispatcher = new EventDispatcher();
|
||||||
|
|
||||||
|
@ -36,7 +30,7 @@ class EventDispatcherFactory
|
||||||
return $eventDispatcher;
|
return $eventDispatcher;
|
||||||
}
|
}
|
||||||
|
|
||||||
private function addLazySubscribers(EventDispatcherInterface $eventDispatcher, $subscriberId)
|
private function addLazySubscribers(EventDispatcherInterface $eventDispatcher, $subscriberId): void
|
||||||
{
|
{
|
||||||
$subscriberClass = $this->objectManager->getClassNameByObjectName($subscriberId);
|
$subscriberClass = $this->objectManager->getClassNameByObjectName($subscriberId);
|
||||||
if (! \is_a($subscriberClass, EventSubscriberInterface::class, true)) {
|
if (! \is_a($subscriberClass, EventSubscriberInterface::class, true)) {
|
||||||
|
|
|
@ -11,27 +11,19 @@ use Symfony\Component\Messenger\Event\WorkerStartedEvent;
|
||||||
|
|
||||||
// This is a 1 to one copy of the original event listener, with a modified RESTART_REQUESTED_TIMESTAMP_KEY to match
|
// This is a 1 to one copy of the original event listener, with a modified RESTART_REQUESTED_TIMESTAMP_KEY to match
|
||||||
// the restriction of the cache ids in flow.
|
// the restriction of the cache ids in flow.
|
||||||
// Also the DI is simplified
|
// Additionally, the DI is simplified
|
||||||
|
|
||||||
/**
|
#[Flow\Scope('singleton')]
|
||||||
* @Flow\Scope("singleton")
|
|
||||||
*/
|
|
||||||
class StopWorkerOnRestartSignalListener implements EventSubscriberInterface
|
class StopWorkerOnRestartSignalListener implements EventSubscriberInterface
|
||||||
{
|
{
|
||||||
public const RESTART_REQUESTED_TIMESTAMP_KEY = 'workers_restart_requested_timestamp';
|
public const RESTART_REQUESTED_TIMESTAMP_KEY = 'workers_restart_requested_timestamp';
|
||||||
|
|
||||||
/**
|
#[Flow\Inject(name: 'DigiComp.FlowSymfonyBridge.Messenger:RestartSignalCachePool', lazy: false)]
|
||||||
* @Flow\Inject(name="DigiComp.FlowSymfonyBridge.Messenger:RestartSignalCachePool")
|
protected CacheItemPoolInterface $cachePool;
|
||||||
* @var CacheItemPoolInterface
|
|
||||||
*/
|
|
||||||
protected $cachePool;
|
|
||||||
|
|
||||||
/**
|
#[Flow\Inject(lazy: false)]
|
||||||
* @Flow\Inject
|
protected LoggerInterface $logger;
|
||||||
* @var LoggerInterface
|
private float $workerStartedAt;
|
||||||
*/
|
|
||||||
protected $logger;
|
|
||||||
private $workerStartedAt;
|
|
||||||
|
|
||||||
public function onWorkerStarted(): void
|
public function onWorkerStarted(): void
|
||||||
{
|
{
|
||||||
|
@ -42,13 +34,11 @@ class StopWorkerOnRestartSignalListener implements EventSubscriberInterface
|
||||||
{
|
{
|
||||||
if ($this->shouldRestart()) {
|
if ($this->shouldRestart()) {
|
||||||
$event->getWorker()->stop();
|
$event->getWorker()->stop();
|
||||||
if (null !== $this->logger) {
|
$this->logger->info('Worker stopped because a restart was requested.');
|
||||||
$this->logger->info('Worker stopped because a restart was requested.');
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
public static function getSubscribedEvents()
|
public static function getSubscribedEvents(): array
|
||||||
{
|
{
|
||||||
return [
|
return [
|
||||||
WorkerStartedEvent::class => 'onWorkerStarted',
|
WorkerStartedEvent::class => 'onWorkerStarted',
|
||||||
|
|
|
@ -5,31 +5,23 @@ namespace DigiComp\FlowSymfonyBridge\Messenger;
|
||||||
use Neos\Flow\Annotations as Flow;
|
use Neos\Flow\Annotations as Flow;
|
||||||
use Neos\Flow\ObjectManagement\ObjectManagerInterface;
|
use Neos\Flow\ObjectManagement\ObjectManagerInterface;
|
||||||
use Neos\Flow\Reflection\ReflectionService;
|
use Neos\Flow\Reflection\ReflectionService;
|
||||||
|
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
|
||||||
use Symfony\Component\Messenger\Handler\HandlerDescriptor;
|
use Symfony\Component\Messenger\Handler\HandlerDescriptor;
|
||||||
use Symfony\Component\Messenger\Handler\HandlersLocator;
|
use Symfony\Component\Messenger\Handler\HandlersLocator;
|
||||||
use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;
|
use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;
|
||||||
|
|
||||||
class HandlersLocatorFactory
|
class HandlersLocatorFactory
|
||||||
{
|
{
|
||||||
/**
|
#[Flow\InjectConfiguration]
|
||||||
* @Flow\InjectConfiguration
|
protected array $configuration;
|
||||||
* @var array
|
|
||||||
*/
|
|
||||||
protected $configuration;
|
|
||||||
|
|
||||||
/**
|
#[Flow\Inject(lazy: false)]
|
||||||
* @Flow\Inject
|
protected ObjectManagerInterface $objectManager;
|
||||||
* @var ObjectManagerInterface
|
|
||||||
*/
|
|
||||||
protected $objectManager;
|
|
||||||
|
|
||||||
/**
|
#[Flow\Inject(lazy: false)]
|
||||||
* @Flow\Inject
|
protected ReflectionService $reflectionService;
|
||||||
* @var ReflectionService
|
|
||||||
*/
|
|
||||||
protected $reflectionService;
|
|
||||||
|
|
||||||
public function create($busName = 'default')
|
public function create($busName = 'default'): HandlersLocator
|
||||||
{
|
{
|
||||||
$messageHandlerClasses = $this->reflectionService
|
$messageHandlerClasses = $this->reflectionService
|
||||||
->getAllImplementationClassNamesForInterface(MessageSubscriberInterface::class);
|
->getAllImplementationClassNamesForInterface(MessageSubscriberInterface::class);
|
||||||
|
@ -50,6 +42,30 @@ class HandlersLocatorFactory
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
$asHandlerClasses = $this->reflectionService
|
||||||
|
->getClassNamesByAnnotation(AsMessageHandler::class);
|
||||||
|
foreach ($asHandlerClasses as $asHandlerClass) {
|
||||||
|
/** @var AsMessageHandler[] $annotations */
|
||||||
|
$annotations = $this->reflectionService->getClassAnnotations($asHandlerClass, AsMessageHandler::class);
|
||||||
|
foreach ($annotations as $annotation) {
|
||||||
|
$config['from_transport'] = $annotation->fromTransport;
|
||||||
|
$config['priority'] = $annotation->priority;
|
||||||
|
$method = $annotation->method ?? '__invoke';
|
||||||
|
$messageName = $annotation->handles;
|
||||||
|
if ($messageName === null) {
|
||||||
|
$arguments = $this->reflectionService->getMethodParameters($asHandlerClass, $method);
|
||||||
|
$messageName = $arguments[\array_key_first($arguments)]['class'];
|
||||||
|
}
|
||||||
|
if ($annotation->bus !== null && $annotation->bus !== $busName) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
$handler = $this->objectManager->get($asHandlerClass);
|
||||||
|
$handlerDescriptors[$messageName][] = new HandlerDescriptor(
|
||||||
|
$this->objectManager->get($asHandlerClass),
|
||||||
|
$config
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
// TODO: Maybe we can allow handlers to be added to bus or globally by configuration?
|
// TODO: Maybe we can allow handlers to be added to bus or globally by configuration?
|
||||||
|
|
||||||
return new HandlersLocator($handlerDescriptors);
|
return new HandlersLocator($handlerDescriptors);
|
||||||
|
|
|
@ -7,25 +7,17 @@ use Neos\Flow\Annotations as Flow;
|
||||||
use Psr\Container\ContainerInterface;
|
use Psr\Container\ContainerInterface;
|
||||||
use Symfony\Component\Messenger\MessageBus;
|
use Symfony\Component\Messenger\MessageBus;
|
||||||
|
|
||||||
/**
|
#[Flow\Scope('singleton')]
|
||||||
* @Flow\Scope("singleton")
|
|
||||||
*/
|
|
||||||
class MessageBusContainer implements ContainerInterface
|
class MessageBusContainer implements ContainerInterface
|
||||||
{
|
{
|
||||||
/**
|
#[Flow\InjectConfiguration(path: 'buses')]
|
||||||
* @Flow\InjectConfiguration(path="buses")
|
protected array $configuration;
|
||||||
* @var array
|
|
||||||
*/
|
|
||||||
protected $configuration;
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @var MessageBus[]
|
* @var MessageBus[]
|
||||||
*/
|
*/
|
||||||
protected array $buses = [];
|
protected array $buses = [];
|
||||||
|
|
||||||
/**
|
|
||||||
* @inheritDoc
|
|
||||||
*/
|
|
||||||
public function get(string $id)
|
public function get(string $id)
|
||||||
{
|
{
|
||||||
if (! isset($this->buses[$id])) {
|
if (! isset($this->buses[$id])) {
|
||||||
|
@ -35,10 +27,7 @@ class MessageBusContainer implements ContainerInterface
|
||||||
return $this->buses[$id];
|
return $this->buses[$id];
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
public function has(string $id): bool
|
||||||
* @inheritDoc
|
|
||||||
*/
|
|
||||||
public function has(string $id)
|
|
||||||
{
|
{
|
||||||
return isset($this->configuration[$id]);
|
return isset($this->configuration[$id]);
|
||||||
}
|
}
|
||||||
|
|
|
@ -23,7 +23,7 @@ class ChainedContainer implements ContainerInterface
|
||||||
throw new \InvalidArgumentException('Service id is unknown: ' . $id);
|
throw new \InvalidArgumentException('Service id is unknown: ' . $id);
|
||||||
}
|
}
|
||||||
|
|
||||||
public function has(string $id)
|
public function has(string $id): bool
|
||||||
{
|
{
|
||||||
foreach ($this->childContainers as $childContainer) {
|
foreach ($this->childContainers as $childContainer) {
|
||||||
if ($childContainer->has($id)) {
|
if ($childContainer->has($id)) {
|
||||||
|
|
|
@ -15,11 +15,8 @@ use Psr\Log\LoggerInterface;
|
||||||
*/
|
*/
|
||||||
class RewindableGenerator implements \IteratorAggregate, \Countable
|
class RewindableGenerator implements \IteratorAggregate, \Countable
|
||||||
{
|
{
|
||||||
/**
|
#[Flow\Inject(lazy: false)]
|
||||||
* @Flow\Inject
|
protected ObjectManagerInterface $objectManager;
|
||||||
* @var ObjectManagerInterface
|
|
||||||
*/
|
|
||||||
protected $objectManager;
|
|
||||||
|
|
||||||
private array $serviceIds;
|
private array $serviceIds;
|
||||||
|
|
||||||
|
@ -46,14 +43,14 @@ class RewindableGenerator implements \IteratorAggregate, \Countable
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
public function getIterator()
|
public function getIterator(): \Traversable
|
||||||
{
|
{
|
||||||
$g = $this->generator;
|
$g = $this->generator;
|
||||||
|
|
||||||
return $g();
|
return $g();
|
||||||
}
|
}
|
||||||
|
|
||||||
public function count()
|
public function count(): int
|
||||||
{
|
{
|
||||||
return \count($this->serviceIds);
|
return \count($this->serviceIds);
|
||||||
}
|
}
|
||||||
|
|
|
@ -8,22 +8,14 @@ use Psr\Container\ContainerInterface;
|
||||||
use Symfony\Component\Messenger\Retry\MultiplierRetryStrategy;
|
use Symfony\Component\Messenger\Retry\MultiplierRetryStrategy;
|
||||||
use Symfony\Component\Messenger\Retry\RetryStrategyInterface;
|
use Symfony\Component\Messenger\Retry\RetryStrategyInterface;
|
||||||
|
|
||||||
/**
|
#[Flow\Scope('singleton')]
|
||||||
* @Flow\Scope("singleton")
|
|
||||||
*/
|
|
||||||
class RetryStrategiesContainer implements ContainerInterface
|
class RetryStrategiesContainer implements ContainerInterface
|
||||||
{
|
{
|
||||||
/**
|
#[Flow\InjectConfiguration]
|
||||||
* @Flow\InjectConfiguration
|
|
||||||
* @var array
|
|
||||||
*/
|
|
||||||
protected array $configuration;
|
protected array $configuration;
|
||||||
|
|
||||||
/**
|
#[Flow\Inject]
|
||||||
* @Flow\Inject
|
protected ObjectManagerInterface $objectManager;
|
||||||
* @var ObjectManagerInterface
|
|
||||||
*/
|
|
||||||
protected $objectManager;
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @var RetryStrategyInterface[]
|
* @var RetryStrategyInterface[]
|
||||||
|
@ -54,7 +46,7 @@ class RetryStrategiesContainer implements ContainerInterface
|
||||||
return $this->retryStrategies[$id];
|
return $this->retryStrategies[$id];
|
||||||
}
|
}
|
||||||
|
|
||||||
public function has(string $id)
|
public function has(string $id): bool
|
||||||
{
|
{
|
||||||
return isset($this->configuration['transports'][$id]);
|
return isset($this->configuration['transports'][$id]);
|
||||||
}
|
}
|
||||||
|
|
31
Classes/Transport/FailureTransportContainer.php
Normal file
31
Classes/Transport/FailureTransportContainer.php
Normal file
|
@ -0,0 +1,31 @@
|
||||||
|
<?php
|
||||||
|
|
||||||
|
namespace DigiComp\FlowSymfonyBridge\Messenger\Transport;
|
||||||
|
|
||||||
|
use Neos\Flow\Annotations as Flow;
|
||||||
|
use Psr\Container\ContainerInterface;
|
||||||
|
use Symfony\Component\Messenger\Transport\TransportInterface;
|
||||||
|
|
||||||
|
#[Flow\Scope('singleton')]
|
||||||
|
class FailureTransportContainer implements ContainerInterface
|
||||||
|
{
|
||||||
|
/**
|
||||||
|
* @var TransportInterface[]
|
||||||
|
*/
|
||||||
|
protected array $transports;
|
||||||
|
|
||||||
|
public function get(string $id)
|
||||||
|
{
|
||||||
|
return $this->transports[$id];
|
||||||
|
}
|
||||||
|
|
||||||
|
public function has(string $id): bool
|
||||||
|
{
|
||||||
|
return isset($this->transports[$id]);
|
||||||
|
}
|
||||||
|
|
||||||
|
public function set(string $id, TransportInterface $transport): void
|
||||||
|
{
|
||||||
|
$this->transports[$id] = $transport;
|
||||||
|
}
|
||||||
|
}
|
|
@ -13,9 +13,7 @@ use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
|
||||||
use Symfony\Component\Messenger\Transport\TransportFactoryInterface;
|
use Symfony\Component\Messenger\Transport\TransportFactoryInterface;
|
||||||
use Symfony\Component\Messenger\Transport\TransportInterface;
|
use Symfony\Component\Messenger\Transport\TransportInterface;
|
||||||
|
|
||||||
/**
|
#[Flow\Scope('singleton')]
|
||||||
* @Flow\Scope("singleton")
|
|
||||||
*/
|
|
||||||
class FlowDoctrineTransportFactory implements TransportFactoryInterface
|
class FlowDoctrineTransportFactory implements TransportFactoryInterface
|
||||||
{
|
{
|
||||||
private EntityManagerInterface $entityManager;
|
private EntityManagerInterface $entityManager;
|
||||||
|
@ -51,11 +49,8 @@ class FlowDoctrineTransportFactory implements TransportFactoryInterface
|
||||||
return new DoctrineTransport($connection, $serializer);
|
return new DoctrineTransport($connection, $serializer);
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* @inheritDoc
|
|
||||||
*/
|
|
||||||
public function supports(string $dsn, array $options): bool
|
public function supports(string $dsn, array $options): bool
|
||||||
{
|
{
|
||||||
return 0 === \strpos($dsn, 'flow-doctrine://');
|
return \str_starts_with($dsn, 'flow-doctrine://');
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -7,33 +7,21 @@ use Symfony\Component\Messenger\Transport\TransportInterface;
|
||||||
|
|
||||||
class NullTransport implements TransportInterface
|
class NullTransport implements TransportInterface
|
||||||
{
|
{
|
||||||
/**
|
|
||||||
* @inheritDoc
|
|
||||||
*/
|
|
||||||
public function get(): iterable
|
public function get(): iterable
|
||||||
{
|
{
|
||||||
return new \EmptyIterator();
|
return new \EmptyIterator();
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* @inheritDoc
|
|
||||||
*/
|
|
||||||
public function ack(Envelope $envelope): void
|
public function ack(Envelope $envelope): void
|
||||||
{
|
{
|
||||||
// do nothing
|
// do nothing
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* @inheritDoc
|
|
||||||
*/
|
|
||||||
public function reject(Envelope $envelope): void
|
public function reject(Envelope $envelope): void
|
||||||
{
|
{
|
||||||
// do nothing
|
// do nothing
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* @inheritDoc
|
|
||||||
*/
|
|
||||||
public function send(Envelope $envelope): Envelope
|
public function send(Envelope $envelope): Envelope
|
||||||
{
|
{
|
||||||
return $envelope;
|
return $envelope;
|
||||||
|
|
|
@ -8,19 +8,13 @@ use Symfony\Component\Messenger\Transport\TransportInterface;
|
||||||
|
|
||||||
class NullTransportFactory implements TransportFactoryInterface
|
class NullTransportFactory implements TransportFactoryInterface
|
||||||
{
|
{
|
||||||
/**
|
|
||||||
* @inheritDoc
|
|
||||||
*/
|
|
||||||
public function createTransport(string $dsn, array $options, SerializerInterface $serializer): TransportInterface
|
public function createTransport(string $dsn, array $options, SerializerInterface $serializer): TransportInterface
|
||||||
{
|
{
|
||||||
return new NullTransport();
|
return new NullTransport();
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* @inheritDoc
|
|
||||||
*/
|
|
||||||
public function supports(string $dsn, array $options): bool
|
public function supports(string $dsn, array $options): bool
|
||||||
{
|
{
|
||||||
return 0 === \strpos($dsn, 'null://');
|
return \str_starts_with($dsn, 'null://');
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -8,28 +8,20 @@ use Psr\Container\ContainerInterface;
|
||||||
use Symfony\Component\Messenger\Transport\TransportFactoryInterface;
|
use Symfony\Component\Messenger\Transport\TransportFactoryInterface;
|
||||||
use Symfony\Component\Messenger\Transport\TransportInterface;
|
use Symfony\Component\Messenger\Transport\TransportInterface;
|
||||||
|
|
||||||
/**
|
#[Flow\Scope('singleton')]
|
||||||
* @Flow\Scope("singleton")
|
|
||||||
*/
|
|
||||||
class TransportsContainer implements ContainerInterface
|
class TransportsContainer implements ContainerInterface
|
||||||
{
|
{
|
||||||
/**
|
#[Flow\InjectConfiguration]
|
||||||
* @Flow\InjectConfiguration
|
|
||||||
* @var array
|
|
||||||
*/
|
|
||||||
protected array $configuration;
|
protected array $configuration;
|
||||||
|
|
||||||
/**
|
#[Flow\Inject(lazy: false)]
|
||||||
* @Flow\Inject
|
protected ObjectManagerInterface $objectManager;
|
||||||
* @var ObjectManagerInterface
|
|
||||||
*/
|
|
||||||
protected $objectManager;
|
|
||||||
|
|
||||||
/**
|
#[Flow\Inject(name: 'DigiComp.FlowSymfonyBridge.Messenger:TransportFactory', lazy: false)]
|
||||||
* @Flow\Inject(name="DigiComp.FlowSymfonyBridge.Messenger:TransportFactory")
|
protected TransportFactoryInterface $transportFactory;
|
||||||
* @var TransportFactoryInterface
|
|
||||||
*/
|
#[Flow\Inject(lazy: false)]
|
||||||
protected $transportFactory;
|
protected FailureTransportContainer $failureTransports;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @var TransportInterface[]
|
* @var TransportInterface[]
|
||||||
|
@ -65,11 +57,16 @@ class TransportsContainer implements ContainerInterface
|
||||||
$transportDefinition['options'],
|
$transportDefinition['options'],
|
||||||
$this->objectManager->get($transportDefinition['serializer'])
|
$this->objectManager->get($transportDefinition['serializer'])
|
||||||
);
|
);
|
||||||
|
if (isset($transportDefinition['failureTransport'])) {
|
||||||
|
$this->failureTransports->set($id, $this->get($transportDefinition['failureTransport']));
|
||||||
|
} elseif (isset($this->configuration['failureTransport'])) {
|
||||||
|
$this->failureTransports->set($id, $this->get($this->configuration['failureTransport']));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
return $this->transports[$id];
|
return $this->transports[$id];
|
||||||
}
|
}
|
||||||
|
|
||||||
public function has(string $id)
|
public function has(string $id): bool
|
||||||
{
|
{
|
||||||
return isset($this->configuration['transports'][$id]);
|
return isset($this->configuration['transports'][$id]);
|
||||||
}
|
}
|
||||||
|
|
|
@ -89,6 +89,13 @@ DigiComp.FlowSymfonyBridge.Messenger:ReceiversContainer:
|
||||||
object: 'DigiComp\FlowSymfonyBridge\Messenger\Transport\TransportsContainer'
|
object: 'DigiComp\FlowSymfonyBridge\Messenger\Transport\TransportsContainer'
|
||||||
# TODO: add own receivers here, which are no transports
|
# TODO: add own receivers here, which are no transports
|
||||||
|
|
||||||
|
DigiComp.FlowSymfonyBridge.Messenger:FailureSenderContainer:
|
||||||
|
className: 'DigiComp\FlowSymfonyBridge\Messenger\ObjectManagement\ChainedContainer'
|
||||||
|
scope: 'singleton'
|
||||||
|
arguments:
|
||||||
|
1:
|
||||||
|
object: 'DigiComp\FlowSymfonyBridge\Messenger\Transport\FailureTransportContainer'
|
||||||
|
|
||||||
DigiComp.FlowSymfonyBridge.Messenger:EventDispatcher:
|
DigiComp.FlowSymfonyBridge.Messenger:EventDispatcher:
|
||||||
className: 'Symfony\Component\EventDispatcher\EventDispatcher'
|
className: 'Symfony\Component\EventDispatcher\EventDispatcher'
|
||||||
scope: 'singleton'
|
scope: 'singleton'
|
||||||
|
@ -133,10 +140,6 @@ Symfony\Component\Messenger\EventListener\SendFailedMessageToFailureTransportLis
|
||||||
arguments:
|
arguments:
|
||||||
1:
|
1:
|
||||||
object:
|
object:
|
||||||
factoryObjectName: 'DigiComp.FlowSymfonyBridge.Messenger:SendersContainer'
|
name: 'DigiComp.FlowSymfonyBridge.Messenger:FailureSenderContainer'
|
||||||
factoryMethodName: 'get'
|
|
||||||
arguments:
|
|
||||||
1:
|
|
||||||
setting: 'DigiComp.FlowSymfonyBridge.Messenger.failureTransport'
|
|
||||||
2:
|
2:
|
||||||
object: 'Psr\Log\LoggerInterface'
|
object: 'Psr\Log\LoggerInterface'
|
||||||
|
|
|
@ -10,7 +10,6 @@ DigiComp:
|
||||||
Messenger:
|
Messenger:
|
||||||
defaultBusName: "default"
|
defaultBusName: "default"
|
||||||
defaultSerializerName: "DigiComp.FlowSymfonyBridge.Messenger:DefaultSerializer"
|
defaultSerializerName: "DigiComp.FlowSymfonyBridge.Messenger:DefaultSerializer"
|
||||||
# TODO: use this
|
|
||||||
defaultRetryStrategyOptions:
|
defaultRetryStrategyOptions:
|
||||||
maxRetries: 3
|
maxRetries: 3
|
||||||
# milliseconds delay
|
# milliseconds delay
|
||||||
|
@ -19,8 +18,8 @@ DigiComp:
|
||||||
# e.g. 1 second delay, 2 seconds, 4 seconds
|
# e.g. 1 second delay, 2 seconds, 4 seconds
|
||||||
multiplier: 2
|
multiplier: 2
|
||||||
maxDelay: 0
|
maxDelay: 0
|
||||||
# override all of this with a service that
|
# override all of this with a service that implements
|
||||||
# implements Symfony\Component\Messenger\Retry\RetryStrategyInterface
|
# Symfony\Component\Messenger\Retry\RetryStrategyInterface
|
||||||
service: null
|
service: null
|
||||||
|
|
||||||
eventDispatcher:
|
eventDispatcher:
|
||||||
|
@ -55,7 +54,4 @@ DigiComp:
|
||||||
dsn: "null://"
|
dsn: "null://"
|
||||||
|
|
||||||
failureTransport: "discard"
|
failureTransport: "discard"
|
||||||
# TODO: Receivers and Senders? (As far as I can see not possible in Symfony)
|
|
||||||
# receivers:[]
|
|
||||||
# senders: []
|
|
||||||
routing: []
|
routing: []
|
||||||
|
|
58
README.md
Normal file
58
README.md
Normal file
|
@ -0,0 +1,58 @@
|
||||||
|
# DigiComp.FlowSymfonyBridge.Messenger
|
||||||
|
|
||||||
|

|
||||||
|
|
||||||
|
This packages brings a DI configuration for the `symfony/messenger` component, so it can be used easily in `neos/flow` projects.
|
||||||
|
|
||||||
|
To see how to use it, you probably want to have a look at the [documentation](https://symfony.com/doc/current/messenger.html) of `symfony/messenger`.
|
||||||
|
|
||||||
|
|
||||||
|
## Getting started
|
||||||
|
|
||||||
|
To get it integrated, you all need to do is to get message bus injected:
|
||||||
|
|
||||||
|
```php
|
||||||
|
#[Flow\Inject]
|
||||||
|
protected MessageBusInterface $messageBus;
|
||||||
|
```
|
||||||
|
|
||||||
|
And later in your method:
|
||||||
|
|
||||||
|
```php
|
||||||
|
$this->messageBus->dispatch(new CustomMessage())
|
||||||
|
```
|
||||||
|
|
||||||
|
You should configure a routing, to let the messenger know, over which transport your message should be handled:
|
||||||
|
|
||||||
|
```yaml
|
||||||
|
DigiComp:
|
||||||
|
FlowSymfonyBridge:
|
||||||
|
Messenger:
|
||||||
|
transports:
|
||||||
|
"custom-messages":
|
||||||
|
dsn: "flow-doctrine://default?table_name=test_messenger_messages"
|
||||||
|
routing:
|
||||||
|
Acme\Vendor\Messenger\CustomMessage:
|
||||||
|
- "custom-messages"
|
||||||
|
```
|
||||||
|
|
||||||
|
In this example we are using a doctrine transport (the speciality "flow-transport" is a transport which uses the already existing connection to doctrine instead of creating a new one - for the rest of the DSN-Format have a look in the documentation of `symfony/messenger`)
|
||||||
|
|
||||||
|
A handler for your CustomMessage could look like this:
|
||||||
|
|
||||||
|
```php
|
||||||
|
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
|
||||||
|
|
||||||
|
#[AsMessageHandler]
|
||||||
|
class CustomMessageHandler
|
||||||
|
{
|
||||||
|
public function __invoke(CustomMessage $message)
|
||||||
|
{
|
||||||
|
//your code here
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
It will be automatically found by Flow // the messenger and messages arriving at the bus will be handled by your handler.
|
||||||
|
|
||||||
|
Probably you'll want to consume the messengers with long living processes or as a cronjob. The Flow command for that task is `messenger:consume` (more help available)
|
|
@ -2,15 +2,11 @@
|
||||||
|
|
||||||
namespace DigiComp\FlowSymfonyBridge\Messenger\Tests\Functional\Fixtures\Message;
|
namespace DigiComp\FlowSymfonyBridge\Messenger\Tests\Functional\Fixtures\Message;
|
||||||
|
|
||||||
use Symfony\Component\Messenger\Handler\MessageSubscriberInterface;
|
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
|
||||||
|
|
||||||
class TestMessageHandler implements MessageSubscriberInterface
|
#[AsMessageHandler]
|
||||||
|
class TestMessageHandler
|
||||||
{
|
{
|
||||||
public static function getHandledMessages(): iterable
|
|
||||||
{
|
|
||||||
yield TestMessage::class => [];
|
|
||||||
}
|
|
||||||
|
|
||||||
public function __invoke(TestMessage $message)
|
public function __invoke(TestMessage $message)
|
||||||
{
|
{
|
||||||
//do nothing for now
|
//do nothing for now
|
||||||
|
|
|
@ -1,12 +1,12 @@
|
||||||
{
|
{
|
||||||
"name": "digicomp/flow-symfony-bridge-messenger",
|
"name": "digicomp/flow-symfony-bridge-messenger",
|
||||||
"type": "neos-package",
|
"type": "neos-package",
|
||||||
"license": "MIT",
|
|
||||||
"description": "Flow dependency injection bridge to symfony/messenger",
|
"description": "Flow dependency injection bridge to symfony/messenger",
|
||||||
"require": {
|
"require": {
|
||||||
"neos/flow": "^6.3",
|
"php": "^8.1",
|
||||||
"symfony/doctrine-messenger": "^5.2.5",
|
"neos/flow": "^8.0",
|
||||||
"symfony/event-dispatcher": "^4.2 | ^5.2"
|
"symfony/doctrine-messenger": "^6.2",
|
||||||
|
"symfony/event-dispatcher": "^4.2 | ^5.2 | ^6.2"
|
||||||
},
|
},
|
||||||
"autoload": {
|
"autoload": {
|
||||||
"psr-4": {
|
"psr-4": {
|
||||||
|
@ -20,10 +20,26 @@
|
||||||
},
|
},
|
||||||
"extra": {
|
"extra": {
|
||||||
"branch-alias": {
|
"branch-alias": {
|
||||||
"dev-master": "0.0.x-dev"
|
"dev-master": "0.1.x-dev"
|
||||||
},
|
},
|
||||||
"neos": {
|
"neos": {
|
||||||
"package-key": "DigiComp.FlowSymfonyBridge.Messenger"
|
"package-key": "DigiComp.FlowSymfonyBridge.Messenger"
|
||||||
}
|
}
|
||||||
}
|
},
|
||||||
|
"authors": [
|
||||||
|
{
|
||||||
|
"name": "Ferdinand Kuhl",
|
||||||
|
"email": "f.kuhl@digital-competence.de",
|
||||||
|
"homepage": "https://www.digital-competence.de",
|
||||||
|
"role": "Developer"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"license": "MIT",
|
||||||
|
"homepage": "https://github.com/digital-competence/DigiComp.FlowSymfonyBridge.Messenger",
|
||||||
|
"keywords": [
|
||||||
|
"Neos",
|
||||||
|
"Flow",
|
||||||
|
"symfony",
|
||||||
|
"messenger"
|
||||||
|
]
|
||||||
}
|
}
|
||||||
|
|
Loading…
Add table
Reference in a new issue