CommandDispatcher::dispatchCommand() - Code Metrics - Inspection of "added Scheduled commands" - xprt64/cqrs-es - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 4321c6...27997e )

by Constantin

created 2017-03-17 08:28 UTC

CommandDispatcher::dispatchCommand() B

↳ Parent: CommandDispatcher

Complexity

Conditions	6
Paths	9

Size

Total Lines	27
Code Lines	13

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	12
CRAP Score	6

Importance

Changes

Metric	Value
dl	0
loc	27
ccs	12
cts	12
cp	1
rs	8.439
c	0
b	0
f	0
cc	6
eloc	13
nc	9
nop	1
crap	6

<?php
/******************************************************************************
 * Copyright (c) 2016 Constantin Galbenu <[email protected]>             *
 ******************************************************************************/

namespace Gica\Cqrs\Command;


use Gica\Cqrs\Aggregate\AggregateRepository;
use Gica\Cqrs\Command;
use Gica\Cqrs\Command\CommandDispatcher\AuthenticatedIdentityReaderService;
use Gica\Cqrs\Command\CommandDispatcher\ConcurrentProofFunctionCaller;
use Gica\Cqrs\Command\Exception\CommandValidationFailed;
use Gica\Cqrs\Command\ValueObject\CommandHandlerAndAggregate;
use Gica\Cqrs\Event;
use Gica\Cqrs\Event\EventDispatcher;
use Gica\Cqrs\Event\EventsApplier\EventsApplierOnAggregate;
use Gica\Cqrs\Event\EventWithMetaData;
use Gica\Cqrs\Event\MetaData;
use Gica\Cqrs\Event\ScheduledEvent;
use Gica\Cqrs\FutureEventsStore;
use Gica\Cqrs\ScheduledCommandStore;
use Gica\Cqrs\Scheduling\ScheduledCommand;
use Gica\Cqrs\Scheduling\ScheduledMessage;

class CommandDispatcher
{
    const MAXIMUM_SAVE_RETRIES = 50;

    /**
     * @var CommandSubscriber
     */
    private $commandSubscriber;
    /**
     * @var EventDispatcher
     */
    private $eventDispatcher;
    /**
     * @var CommandApplier
     */
    private $commandApplier;
    /**
     * @var AggregateRepository
     */
    private $aggregateRepository;
    /**
     * @var ConcurrentProofFunctionCaller
     */
    private $concurrentProofFunctionCaller;
    /**
     * @var CommandValidator
     */
    private $commandValidator;
    /**
     * @var AuthenticatedIdentityReaderService
     */
    private $authenticatedIdentityServiceReader;
    /**
     * @var FutureEventsStore
     */
    private $futureEventsStore;
    /**
     * @var EventsApplierOnAggregate
     */
    private $eventsApplierOnAggregate;
    /**
     * @var ScheduledCommandStore|null
     */
    private $scheduledCommandStore;

    /**
     * @param CommandSubscriber $commandSubscriber
     * @param EventDispatcher $eventDispatcher
     * @param CommandApplier $commandApplier
     * @param AggregateRepository $aggregateRepository
     * @param ConcurrentProofFunctionCaller $functionCaller
     * @param CommandValidator $commandValidator
     * @param AuthenticatedIdentityReaderService $authService
     * @param FutureEventsStore|null $futureEventsStore
     * @param ScheduledCommandStore|null $commandStore
     * @param EventsApplierOnAggregate $eventsApplier
     */
    public function __construct(
        CommandSubscriber $commandSubscriber,
        EventDispatcher $eventDispatcher,
        CommandApplier $commandApplier,
        AggregateRepository $aggregateRepository,
        ConcurrentProofFunctionCaller $functionCaller,
        CommandValidator $commandValidator,
        AuthenticatedIdentityReaderService $authService,
        ?FutureEventsStore $futureEventsStore = null,
        EventsApplierOnAggregate $eventsApplier,
        ?ScheduledCommandStore $commandStore = null
    )
    {
        $this->commandSubscriber = $commandSubscriber;
        $this->eventDispatcher = $eventDispatcher;
        $this->commandApplier = $commandApplier;
        $this->aggregateRepository = $aggregateRepository;
        $this->concurrentProofFunctionCaller = $functionCaller;
        $this->commandValidator = $commandValidator;
        $this->authenticatedIdentityServiceReader = $authService;
        $this->futureEventsStore = $futureEventsStore;
        $this->eventsApplierOnAggregate = $eventsApplier;
        $this->scheduledCommandStore = $commandStore;
    }

    public function dispatchCommand(Command $command)
    {
        $errors = $this->commandValidator->validateCommand($command);

        if (!empty($errors)) {
            throw new CommandValidationFailed($errors);
        }

        /** @var EventWithMetaData[] $eventsWithMetaData */
        /** @var ScheduledCommand[] $scheduledCommands */

        list($eventsWithMetaData, $futureEventsWithMeta, $scheduledCommands) = $this->concurrentProofFunctionCaller->executeFunction(function () use ($command) {
            return $this->tryDispatchCommandAndSaveAggregate($command);
        }, self::MAXIMUM_SAVE_RETRIES);

        foreach ($eventsWithMetaData as $eventWithMetaData) {
            $this->eventDispatcher->dispatchEvent($eventWithMetaData);
        }

        if ($this->futureEventsStore) {
            $this->futureEventsStore->scheduleEvents($futureEventsWithMeta);
        }

        if ($this->scheduledCommandStore && $scheduledCommands) {

            $this->scheduledCommandStore->scheduleCommands($scheduledCommands);
        }
    }

    private function tryDispatchCommandAndSaveAggregate(Command $command)
    {
        $handlerAndAggregate = $this->loadCommandHandlerAndAggregate($command);

        $eventsWithMetaData = $this->applyCommandAndReturnEvents($command, $handlerAndAggregate);

        list($eventsForNow, $eventsForTheFuture, $scheduledCommands) = $this->splitMessagesByType($eventsWithMetaData);

        $this->aggregateRepository->saveAggregate($command->getAggregateId(), $handlerAndAggregate->getAggregate(), $eventsForNow);

        return [$eventsForNow, $eventsForTheFuture, $scheduledCommands];
    }

    /**
     * @param EventWithMetaData[]|ScheduledCommand[] $messages
     * @return array
     */
    private function splitMessagesByType($messages)
    {
        $nowEvents = [];
        $futureEvents = [];
        $scheduledCommands = [];

        foreach ($messages as $message) {
            if ($this->isScheduledCommand($message)) {
                $scheduledCommands[] = $message;
            } else if ($this->isScheduledEvent($message->getEvent())) {
class A
{
    public function foo() { }
}

class B extends A
{
    public function bar() { }
}

/**
 * @param A|B $x
 */
function someFunction($x)
{
    $x->foo(); // This call is fine as the method exists in A and B.
    $x->bar(); // This method only exists in B and might cause an error.
}
                $futureEvents[] = $message;
            } else {
                $nowEvents[] = $message;
            }
        }

        return [$nowEvents, $futureEvents, $scheduledCommands];
    }

    public function canExecuteCommand(Command $command): bool
    {
        try {
            $errors = $this->commandValidator->validateCommand($command);
            if (!empty($errors)) {
                return false;
            }
            $handlerAndAggregate = $this->loadCommandHandlerAndAggregate($command);
            $this->applyCommandAndReturnEvents($command, $handlerAndAggregate);
            return true;
        } catch (\Exception $exception) {
            return false;
        }
    }

    private function loadCommandHandlerAndAggregate(Command $command): CommandHandlerAndAggregate
    {
        $handler = $this->commandSubscriber->getHandlerForCommand($command);

        $aggregate = $this->aggregateRepository->loadAggregate($handler->getHandlerClass(), $command->getAggregateId());

        return new CommandHandlerAndAggregate($handler, $aggregate);
    }

    private function decorateEventWithMetaData(Event $event, MetaData $metaData): EventWithMetaData
    {
        return new EventWithMetaData($event, $metaData);
    }

    /**
     * @param Command $command
     * @param CommandHandlerAndAggregate $handlerAndAggregate
     * @return EventWithMetaData[]|ScheduledCommand[]
     */
    private function applyCommandAndReturnEvents(Command $command, CommandHandlerAndAggregate $handlerAndAggregate)
    {
        $aggregate = $handlerAndAggregate->getAggregate();
        $handler = $handlerAndAggregate->getCommandHandler();

        $metaData = $this->factoryMetadata($command, $aggregate);

        $newMessageGenerator = $this->commandApplier->applyCommand($aggregate, $command, $handler->getMethodName());

        /** @var EventWithMetaData[]|ScheduledCommand[] $eventsWithMetaData */
        $eventsWithMetaData = [];

        foreach ($newMessageGenerator as $message) {
            if ($this->isScheduledCommand($message)) {
                $eventsWithMetaData[] = $message;
            } else {
                $eventWithMetaData = $this->decorateEventWithMetaData($message, $metaData);
                if (!$this->isScheduledMessage($message)) {
                    $this->eventsApplierOnAggregate->applyEventsOnAggregate($aggregate, [$eventWithMetaData]);
                }
                $eventsWithMetaData[] = $eventWithMetaData;
            }
        }

        return $eventsWithMetaData;
    }

    private function isScheduledEvent($event): bool
    {
        return $event instanceof ScheduledEvent;
    }

    private function isScheduledCommand($message): bool
    {
        return $message instanceof ScheduledCommand;
    }

    private function isScheduledMessage($message): bool
    {
        return $message instanceof ScheduledMessage;
    }

    private function factoryMetadata(Command $command, $aggregate): MetaData
    {
        return new MetaData(
            $command->getAggregateId(),
            get_class($aggregate),
            new \DateTimeImmutable(),
            $this->authenticatedIdentityServiceReader->getAuthenticatedIdentityId());
    }
}

1		<?php
2		/******************************************************************************
3		* Copyright (c) 2016 Constantin Galbenu <[email protected]> *
4		******************************************************************************/
5
6		namespace Gica\Cqrs\Command;
7
8
9		use Gica\Cqrs\Aggregate\AggregateRepository;
10		use Gica\Cqrs\Command;
11		use Gica\Cqrs\Command\CommandDispatcher\AuthenticatedIdentityReaderService;
12		use Gica\Cqrs\Command\CommandDispatcher\ConcurrentProofFunctionCaller;
13		use Gica\Cqrs\Command\Exception\CommandValidationFailed;
14		use Gica\Cqrs\Command\ValueObject\CommandHandlerAndAggregate;
15		use Gica\Cqrs\Event;
16		use Gica\Cqrs\Event\EventDispatcher;
17		use Gica\Cqrs\Event\EventsApplier\EventsApplierOnAggregate;
18		use Gica\Cqrs\Event\EventWithMetaData;
19		use Gica\Cqrs\Event\MetaData;
20		use Gica\Cqrs\Event\ScheduledEvent;
21		use Gica\Cqrs\FutureEventsStore;
22		use Gica\Cqrs\ScheduledCommandStore;
23		use Gica\Cqrs\Scheduling\ScheduledCommand;
24		use Gica\Cqrs\Scheduling\ScheduledMessage;
25
26		class CommandDispatcher
27		{
28		const MAXIMUM_SAVE_RETRIES = 50;
29
30		/**
31		* @var CommandSubscriber
32		*/
33		private $commandSubscriber;
34		/**
35		* @var EventDispatcher
36		*/
37		private $eventDispatcher;
38		/**
39		* @var CommandApplier
40		*/
41		private $commandApplier;
42		/**
43		* @var AggregateRepository
44		*/
45		private $aggregateRepository;
46		/**
47		* @var ConcurrentProofFunctionCaller
48		*/
49		private $concurrentProofFunctionCaller;
50		/**
51		* @var CommandValidator
52		*/
53		private $commandValidator;
54		/**
55		* @var AuthenticatedIdentityReaderService
56		*/
57		private $authenticatedIdentityServiceReader;
58		/**
59		* @var FutureEventsStore
60		*/
61		private $futureEventsStore;
62		/**
63		* @var EventsApplierOnAggregate
64	5	*/
65		private $eventsApplierOnAggregate;
66		/**
67		* @var ScheduledCommandStore\|null
68		*/
69		private $scheduledCommandStore;
70
71		/**
72		* @param CommandSubscriber $commandSubscriber
73		* @param EventDispatcher $eventDispatcher
74		* @param CommandApplier $commandApplier
75		* @param AggregateRepository $aggregateRepository
76	5	* @param ConcurrentProofFunctionCaller $functionCaller
77	5	* @param CommandValidator $commandValidator
78	5	* @param AuthenticatedIdentityReaderService $authService
79	5	* @param FutureEventsStore\|null $futureEventsStore
80	5	* @param ScheduledCommandStore\|null $commandStore
81	5	* @param EventsApplierOnAggregate $eventsApplier
82	5	*/
83	5	public function __construct(
84	5	CommandSubscriber $commandSubscriber,
85	5	EventDispatcher $eventDispatcher,
86		CommandApplier $commandApplier,
87	4	AggregateRepository $aggregateRepository,
88		ConcurrentProofFunctionCaller $functionCaller,
89	4	CommandValidator $commandValidator,
90		AuthenticatedIdentityReaderService $authService,
91	4	?FutureEventsStore $futureEventsStore = null,
92	1	EventsApplierOnAggregate $eventsApplier,
93		?ScheduledCommandStore $commandStore = null
94		)
95		{
96	3	$this->commandSubscriber = $commandSubscriber;
97	3	$this->eventDispatcher = $eventDispatcher;
98	3	$this->commandApplier = $commandApplier;
99		$this->aggregateRepository = $aggregateRepository;
100	2	$this->concurrentProofFunctionCaller = $functionCaller;
101	2	$this->commandValidator = $commandValidator;
102		$this->authenticatedIdentityServiceReader = $authService;
103		$this->futureEventsStore = $futureEventsStore;
104	2	$this->eventsApplierOnAggregate = $eventsApplier;
105	2	$this->scheduledCommandStore = $commandStore;
106		}
107	3
108		public function dispatchCommand(Command $command)
109	3	{
110		$errors = $this->commandValidator->validateCommand($command);
111	3
112		if (!empty($errors)) {
113	2	throw new CommandValidationFailed($errors);
114		}
115	2
116		/** @var EventWithMetaData[] $eventsWithMetaData */
117	2	/** @var ScheduledCommand[] $scheduledCommands */
118
119		list($eventsWithMetaData, $futureEventsWithMeta, $scheduledCommands) = $this->concurrentProofFunctionCaller->executeFunction(function () use ($command) {
120		return $this->tryDispatchCommandAndSaveAggregate($command);
121		}, self::MAXIMUM_SAVE_RETRIES);
122
123		foreach ($eventsWithMetaData as $eventWithMetaData) {
124	2	$this->eventDispatcher->dispatchEvent($eventWithMetaData);
125		}
126	2
127	2	if ($this->futureEventsStore) {
128		$this->futureEventsStore->scheduleEvents($futureEventsWithMeta);
129	2	}
130	2
131	1	if ($this->scheduledCommandStore && $scheduledCommands) {
		0 ignored issues – show Bug Best Practice introduced 2017-03-17 08:34 UTC by Report Bug Copy Issue Report The expression `$scheduledCommands` of type `Gica\Cqrs\Scheduling\ScheduledCommand[]` is implicitly converted to a boolean; are you sure this is intended? If so, consider using `! empty($expr)` instead to make it clear that you intend to check for an array without elements. This check marks implicit conversions of arrays to boolean values in a comparison. While in PHP an empty array is considered to be equal (but not identical) to false, this is not always apparent. Consider making the comparison explicit by using `empty(..)` or `! empty(...)` instead. Loading history...
132		$this->scheduledCommandStore->scheduleCommands($scheduledCommands);
133	2	}
134		}
135
136		private function tryDispatchCommandAndSaveAggregate(Command $command)
137	2	{
138		$handlerAndAggregate = $this->loadCommandHandlerAndAggregate($command);
139
140	4	$eventsWithMetaData = $this->applyCommandAndReturnEvents($command, $handlerAndAggregate);
141
142		list($eventsForNow, $eventsForTheFuture, $scheduledCommands) = $this->splitMessagesByType($eventsWithMetaData);
143	4
144	4	$this->aggregateRepository->saveAggregate($command->getAggregateId(), $handlerAndAggregate->getAggregate(), $eventsForNow);
145	1
146		return [$eventsForNow, $eventsForTheFuture, $scheduledCommands];
147	3	}
148	3
149	2	/**
150	1	* @param EventWithMetaData[]\|ScheduledCommand[] $messages
151	1	* @return array
152		*/
153		private function splitMessagesByType($messages)
154		{
155	4	$nowEvents = [];
156		$futureEvents = [];
157	4	$scheduledCommands = [];
158
159	4	foreach ($messages as $message) {
160		if ($this->isScheduledCommand($message)) {
161	4	$scheduledCommands[] = $message;
162		} else if ($this->isScheduledEvent($message->getEvent())) {
		0 ignored issues – show Bug introduced 2017-03-17 08:34 UTC by Report Bug Copy Issue Report The method `getEvent` does only exist in `Gica\Cqrs\Event\EventWithMetaData`, but not in `Gica\Cqrs\Scheduling\ScheduledCommand`. It seems like the method you are trying to call exists only in some of the possible types. Let’s take a look at an example: class A { public function foo() { } } class B extends A { public function bar() { } } /** * @param A\|B $x / function someFunction($x) { $x->foo(); // This call is fine as the method exists in A and B. $x->bar(); // This method only exists in B and might cause an error. } Available Fixes Add an additional type-check: /* * @param A\|B $x / function someFunction($x) { $x->foo(); if ($x instanceof B) { $x->bar(); } } Only allow a single type to be passed if the variable comes from a parameter: function someFunction(B $x) { /* ... */ } Loading history...
163		$futureEvents[] = $message;
164	2	} else {
165		$nowEvents[] = $message;
166	2	}
167		}
168
169		return [$nowEvents, $futureEvents, $scheduledCommands];
170		}
171
172		public function canExecuteCommand(Command $command): bool
173		{
174	4	try {
175		$errors = $this->commandValidator->validateCommand($command);
176	4	if (!empty($errors)) {
177	4	return false;
178		}
179	4	$handlerAndAggregate = $this->loadCommandHandlerAndAggregate($command);
180		$this->applyCommandAndReturnEvents($command, $handlerAndAggregate);
181	4	return true;
182		} catch (\Exception $exception) {
183		return false;
184	4	}
185		}
186	4
187	2	private function loadCommandHandlerAndAggregate(Command $command): CommandHandlerAndAggregate
188		{
189	2	$handler = $this->commandSubscriber->getHandlerForCommand($command);
190	2
191		$aggregate = $this->aggregateRepository->loadAggregate($handler->getHandlerClass(), $command->getAggregateId());
192
193	2	return new CommandHandlerAndAggregate($handler, $aggregate);
194		}
195
196	2	private function decorateEventWithMetaData(Event $event, MetaData $metaData): EventWithMetaData
197		{
198		return new EventWithMetaData($event, $metaData);
199	2	}
200
201	2	/**
202		* @param Command $command
203		* @param CommandHandlerAndAggregate $handlerAndAggregate
204	4	* @return EventWithMetaData[]\|ScheduledCommand[]
205		*/
206	4	private function applyCommandAndReturnEvents(Command $command, CommandHandlerAndAggregate $handlerAndAggregate)
207	4	{
208		$aggregate = $handlerAndAggregate->getAggregate();
209	4	$handler = $handlerAndAggregate->getCommandHandler();
210	4
211		$metaData = $this->factoryMetadata($command, $aggregate);
212
213		$newMessageGenerator = $this->commandApplier->applyCommand($aggregate, $command, $handler->getMethodName());
214
215		/** @var EventWithMetaData[]\|ScheduledCommand[] $eventsWithMetaData */
216		$eventsWithMetaData = [];
217
218		foreach ($newMessageGenerator as $message) {
219		if ($this->isScheduledCommand($message)) {
220		$eventsWithMetaData[] = $message;
221		} else {
222		$eventWithMetaData = $this->decorateEventWithMetaData($message, $metaData);
223		if (!$this->isScheduledMessage($message)) {
224		$this->eventsApplierOnAggregate->applyEventsOnAggregate($aggregate, [$eventWithMetaData]);
225		}
226		$eventsWithMetaData[] = $eventWithMetaData;
227		}
228		}
229
230		return $eventsWithMetaData;
231		}
232
233		private function isScheduledEvent($event): bool
234		{
235		return $event instanceof ScheduledEvent;
236		}
237
238		private function isScheduledCommand($message): bool
239		{
240		return $message instanceof ScheduledCommand;
241		}
242
243		private function isScheduledMessage($message): bool
244		{
245		return $message instanceof ScheduledMessage;
246		}
247
248		private function factoryMetadata(Command $command, $aggregate): MetaData
249		{
250		return new MetaData(
251		$command->getAggregateId(),
252		get_class($aggregate),
253		new \DateTimeImmutable(),
254		$this->authenticatedIdentityServiceReader->getAuthenticatedIdentityId());
255		}
256		}

xprt64 / cqrs-es

Push — master ( 4321c6...27997e )

CommandDispatcher::dispatchCommand() B

Complexity

Size

Duplication

Code Coverage

Importance

Available Fixes

Duplication Side-by-Side

Filter issues like