Skip to content

Commit d211546

Browse files
dgafkadgafka
authored
Refactor aggregate module (#424)
* Simplify aggregate module
1 parent 0d2a1aa commit d211546

File tree

72 files changed

+1645
-2308
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

72 files changed

+1645
-2308
lines changed

‎packages/Amqp/tests/Integration/CallAggregateAsynchronousEndpointTest.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,6 @@ classesToResolve: [Calendar::class],
4242
->sendCommand(new ScheduleMeeting('1', '2'))
4343
;
4444

45-
self::assertFalse($ecotone->getMessageChannel('calendar')->receive()->getHeaders()->containsKey(AggregateMessage::CALLED_AGGREGATE_OBJECT));
45+
self::assertFalse($ecotone->getMessageChannel('calendar')->receive()->getHeaders()->containsKey(AggregateMessage::CALLED_AGGREGATE_INSTANCE));
4646
}
4747
}

‎packages/Ecotone/src/Lite/Test/ConfiguredMessagingSystemWithTestSupport.php‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,9 +11,11 @@
1111
use Ecotone\Messaging\Handler\Gateway\Gateway;
1212
use Ecotone\Messaging\MessageChannel;
1313
use Ecotone\Messaging\MessagePublisher;
14+
use Ecotone\Modelling\AggregateFlow\SaveAggregate\AggregateResolver\AggregateDefinitionRegistry;
1415
use Ecotone\Modelling\CommandBus;
1516
use Ecotone\Modelling\DistributedBus;
1617
use Ecotone\Modelling\EventBus;
18+
use Ecotone\Modelling\EventSourcingExecutor\GroupedEventSourcingExecutor;
1719
use Ecotone\Modelling\QueryBus;
1820

1921
/**
@@ -84,12 +86,18 @@ public function getFlowTestSupport(): FlowTestSupport
8486
$this->getCommandBus(),
8587
$this->getEventBus(),
8688
$this->getQueryBus(),
89+
$this->getServiceFromContainer(AggregateDefinitionRegistry::class),
8790
$this->getMessagingTestSupport(),
8891
$this->getGatewayByName(MessagingEntrypoint::class),
8992
$this->configuredMessagingSystem
9093
);
9194
}
9295

96+
/**
97+
* @template T
98+
* @param class-string<T> $referenceName
99+
* @return T
100+
*/
93101
public function getServiceFromContainer(string $referenceName): object
94102
{
95103
return $this->configuredMessagingSystem->getServiceFromContainer($referenceName);

‎packages/Ecotone/src/Lite/Test/FlowTestSupport.php‎

Lines changed: 16 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
use Ecotone\Messaging\Conversion\MediaType;
1212
use Ecotone\Messaging\Endpoint\ExecutionPollingMetadata;
1313
use Ecotone\Messaging\Gateway\MessagingEntrypoint;
14+
use Ecotone\Messaging\Handler\TypeDescriptor;
1415
use Ecotone\Messaging\Message;
1516
use Ecotone\Messaging\MessageChannel;
1617
use Ecotone\Messaging\MessageHeaders;
@@ -19,13 +20,16 @@
1920
use Ecotone\Messaging\Scheduling\TimeSpan;
2021
use Ecotone\Messaging\Support\Assert;
2122
use Ecotone\Messaging\Support\MessageBuilder;
23+
use Ecotone\Modelling\AggregateFlow\SaveAggregate\AggregateResolver\AggregateDefinitionRegistry;
2224
use Ecotone\Modelling\AggregateMessage;
2325
use Ecotone\Modelling\CommandBus;
2426
use Ecotone\Modelling\Config\MessageBusChannel;
2527
use Ecotone\Modelling\Config\AggregrateHandlerModule;
2628
use Ecotone\Modelling\Event;
2729
use Ecotone\Modelling\EventBus;
30+
use Ecotone\Modelling\EventSourcingExecutor\GroupedEventSourcingExecutor;
2831
use Ecotone\Modelling\QueryBus;
32+
use Ecotone\Test\StubEventSourcedAggregate;
2933

3034
/**
3135
* @template T
@@ -39,6 +43,7 @@ public function __construct(
3943
private CommandBus $commandBus,
4044
private EventBus $eventBus,
4145
private QueryBus $queryBus,
46+
private AggregateDefinitionRegistry $aggregateDefinitionRegistry,
4247
private MessagingTestSupport $testSupportGateway,
4348
private MessagingEntrypoint $messagingEntrypoint,
4449
private ConfiguredMessagingSystem $configuredMessagingSystem
@@ -179,13 +184,17 @@ public function getEventStreamEvents(string $streamName): array
179184
*/
180185
public function withEventsFor(string|object|array $identifiers, string $aggregateClass, array $events, int $aggregateVersion = 0): self
181186
{
187+
$aggregateDefinition = $this->aggregateDefinitionRegistry->getFor(TypeDescriptor::create($aggregateClass));
188+
Assert::isTrue($aggregateDefinition->isEventSourced(), "Aggregate {$aggregateClass} is not event sourced. Can't store events for it.");
189+
182190
$this->messagingEntrypoint->sendWithHeaders(
183-
$events,
191+
[],
184192
[
185-
AggregateMessage::OVERRIDE_AGGREGATE_IDENTIFIER => is_object($identifiers) ? (string)$identifiers : $identifiers,
193+
AggregateMessage::AGGREGATE_ID => is_object($identifiers) ? (string)$identifiers : $identifiers,
186194
AggregateMessage::TARGET_VERSION => $aggregateVersion,
187-
AggregateMessage::RESULT_AGGREGATE_OBJECT => $aggregateClass,
188-
AggregateMessage::RESULT_AGGREGATE_EVENTS => $events,
195+
AggregateMessage::CALLED_AGGREGATE_CLASS => $aggregateClass,
196+
AggregateMessage::CALLED_AGGREGATE_INSTANCE => new $aggregateClass(),
197+
AggregateMessage::RECORDED_AGGREGATE_EVENTS => $events,
189198
],
190199
AggregrateHandlerModule::getRegisterAggregateSaveRepositoryInputChannel($aggregateClass). '.test_setup_state'
191200
);
@@ -198,7 +207,8 @@ public function withStateFor(object $aggregate): self
198207
$this->messagingEntrypoint->sendWithHeaders(
199208
$aggregate,
200209
[
201-
AggregateMessage::RESULT_AGGREGATE_OBJECT => $aggregate,
210+
AggregateMessage::CALLED_AGGREGATE_INSTANCE => $aggregate,
211+
AggregateMessage::CALLED_AGGREGATE_CLASS => $aggregate::class,
202212
],
203213
AggregrateHandlerModule::getRegisterAggregateSaveRepositoryInputChannel($aggregate::class). '.test_setup_state'
204214
);
@@ -397,6 +407,7 @@ public function sendMessageDirectToChannelWithMessageReply(string $targetChannel
397407
}
398408

399409
/**
410+
* @template T
400411
* @param class-string<T> $referenceName
401412
* @return T
402413
*/

‎packages/Ecotone/src/Messaging/Config/ConfiguredMessagingSystem.php‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,9 @@ public function getMessagePublisher(string $referenceName = MessagePublisher::cl
5757

5858
/**
5959
* @throws InvalidArgumentException if trying to find not existing service reference
60+
* @template T
61+
* @param class-string<T> $referenceName
62+
* @return T
6063
*/
6164
public function getServiceFromContainer(string $referenceName): object;
6265

‎packages/Ecotone/src/Messaging/Handler/Enricher/PropertyEditorAccessor.php‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,8 @@
44

55
namespace Ecotone\Messaging\Handler\Enricher;
66

7+
use Ecotone\Messaging\Config\Container\Definition;
8+
use Ecotone\Messaging\Config\Container\Reference;
79
use Ecotone\Messaging\Handler\ExpressionEvaluationService;
810
use Ecotone\Messaging\Message;
911
use Ecotone\Messaging\Support\Assert;
@@ -35,6 +37,17 @@ public static function create(ExpressionEvaluationService $expressionEvaluationS
3537
return self::createWithMapping($expressionEvaluationService, '');
3638
}
3739

40+
public static function getDefinition(): Definition
41+
{
42+
return new Definition(
43+
self::class,
44+
[
45+
Reference::to(ExpressionEvaluationService::REFERENCE)
46+
],
47+
[self::class, 'create']
48+
);
49+
}
50+
3851
/**
3952
* @param PropertyPath $propertyNamePath
4053
* @param mixed $dataToEnrich

‎packages/Ecotone/src/Messaging/Handler/Enricher/PropertyReaderAccessor.php‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44

55
namespace Ecotone\Messaging\Handler\Enricher;
66

7+
use Ecotone\Messaging\Config\Container\Definition;
78
use Ecotone\Messaging\MessagingException;
89
use Ecotone\Messaging\Support\InvalidArgumentException;
910
use ReflectionClass;
@@ -39,6 +40,14 @@ public function hasPropertyValue(PropertyPath $propertyPath, $fromData): bool
3940
return true;
4041
}
4142

43+
public static function getDefinition(): Definition
44+
{
45+
return new Definition(
46+
self::class,
47+
[],
48+
);
49+
}
50+
4251
/**
4352
* @param PropertyPath $propertyPath
4453
* @param mixed $fromData

‎packages/Ecotone/src/Messaging/Handler/Processor/MethodInvoker/MethodInvokerAggregateObjectResolver.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,6 @@ class MethodInvokerAggregateObjectResolver implements MethodInvokerObjectResolve
1313
{
1414
public function resolveFor(Message $message): object
1515
{
16-
return $message->getHeaders()->get(AggregateMessage::CALLED_AGGREGATE_OBJECT);
16+
return $message->getHeaders()->get(AggregateMessage::CALLED_AGGREGATE_INSTANCE);
1717
}
1818
}

‎packages/Ecotone/src/Messaging/Handler/Router/HeaderExistsRouter.php‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
<?php
2+
3+
namespace Ecotone\Messaging\Handler\Router;
4+
5+
use Ecotone\Messaging\Config\Container\DefinedObject;
6+
use Ecotone\Messaging\Config\Container\Definition;
7+
use Ecotone\Messaging\Message;
8+
9+
/**
10+
* Class HeaderValueRouter
11+
* @package Ecotone\Messaging\Handler\Router
12+
* @author Dariusz Gafka <support@simplycodedsoftware.com>
13+
* @internal
14+
*/
15+
/**
16+
* licence Apache-2.0
17+
*/
18+
final class HeaderExistsRouter implements RouteSelector, DefinedObject
19+
{
20+
21+
private function __construct(private string $headerName, private string $routeToChannel, private string $fallbackRoute)
22+
{
23+
}
24+
25+
public static function create(string $headerName, string $routeToChannel, string $fallbackRoute): self
26+
{
27+
return new self($headerName, $routeToChannel, $fallbackRoute);
28+
}
29+
30+
/**
31+
* @param Message $message
32+
* @return array
33+
*/
34+
public function route(Message $message): array
35+
{
36+
return $message->getHeaders()->containsKey($this->headerName)
37+
? [$this->routeToChannel]
38+
: [$this->fallbackRoute];
39+
}
40+
41+
public function getDefinition(): Definition
42+
{
43+
return new Definition(self::class, [
44+
$this->headerName,
45+
$this->routeToChannel,
46+
$this->fallbackRoute
47+
], 'create');
48+
}
49+
}

‎packages/Ecotone/src/Messaging/Handler/Router/RouterProcessorBuilder.php‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,21 @@ public static function createRecipientListRouter(array $recipientList): self
4141
);
4242
}
4343

44+
public static function createHeaderExistsRouter(string $headerName, string $routeToChannel, string $fallbackRoute): self
45+
{
46+
return new self(
47+
HeaderExistsRouter::create($headerName, $routeToChannel, $fallbackRoute)->getDefinition(),
48+
[
49+
$routeToChannel => new Definition(SendToChannelProcessor::class, [
50+
new ChannelReference($routeToChannel),
51+
]),
52+
$fallbackRoute => new Definition(SendToChannelProcessor::class, [
53+
new ChannelReference($fallbackRoute),
54+
]),
55+
]
56+
);
57+
}
58+
4459
public function route(string $routeName, CompilableBuilder $processor): self
4560
{
4661
$this->routeMap[$routeName] = $processor;

‎packages/Ecotone/src/Messaging/Handler/Transformer/TransformerBuilder.php‎

Lines changed: 36 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -133,24 +133,7 @@ public function getParameterConverters(): array
133133

134134
public function compile(MessagingContainerBuilder $builder): Definition
135135
{
136-
if ($this->expression) {
137-
$objectToInvokeOn = new Definition(ExpressionTransformer::class, [$this->expression, new Reference(ExpressionEvaluationService::REFERENCE), new Reference(ReferenceSearchService::class)]);
138-
$interfaceToCallReference = new InterfaceToCallReference(ExpressionTransformer::class, 'transform');
139-
} else {
140-
$objectToInvokeOn = $this->directObject ?: new Reference($this->objectToInvokeReferenceName);
141-
if ($this->methodNameOrInterface instanceof InterfaceToCall) {
142-
$interfaceToCallReference = InterfaceToCallReference::fromInstance($this->methodNameOrInterface);
143-
} else {
144-
$className = $this->directObject ? \get_class($objectToInvokeOn) : $this->objectToInvokeReferenceName;
145-
$interfaceToCallReference = new InterfaceToCallReference($className, $this->getMethodName());
146-
}
147-
}
148-
149-
$interfaceToCall = $builder->getInterfaceToCall($interfaceToCallReference);
150-
151-
if (! $interfaceToCall->canReturnValue()) {
152-
throw InvalidArgumentException::create("Can't create transformer for {$interfaceToCall}, because method has no return value");
153-
}
136+
list($objectToInvokeOn, $interfaceToCallReference, $interfaceToCall) = $this->prepare($builder);
154137

155138
$newImplementation = MessageProcessorActivatorBuilder::create()
156139
->withEndpointId($this->getEndpointId())
@@ -172,6 +155,17 @@ public function compile(MessagingContainerBuilder $builder): Definition
172155
return $newImplementation->compile($builder);
173156
}
174157

158+
public function compileProcessor(MessagingContainerBuilder $builder): Definition
159+
{
160+
list($objectToInvokeOn, $interfaceToCallReference, $interfaceToCall) = $this->prepare($builder);
161+
162+
return MethodInvokerBuilder::create(
163+
$objectToInvokeOn,
164+
$interfaceToCallReference,
165+
$this->methodParameterConverterBuilders
166+
)->compile($builder);
167+
}
168+
175169
private function setDirectObjectToInvoke(DefinedObject $objectToInvoke): void
176170
{
177171
$this->directObject = $objectToInvoke;
@@ -202,4 +196,28 @@ private function getMethodName(): string|InterfaceToCall
202196
? $this->methodNameOrInterface->getMethodName()
203197
: $this->methodNameOrInterface;
204198
}
199+
200+
public function prepare(MessagingContainerBuilder $builder): array
201+
{
202+
if ($this->expression) {
203+
$objectToInvokeOn = new Definition(ExpressionTransformer::class, [$this->expression, new Reference(ExpressionEvaluationService::REFERENCE), new Reference(ReferenceSearchService::class)]);
204+
$interfaceToCallReference = new InterfaceToCallReference(ExpressionTransformer::class, 'transform');
205+
} else {
206+
$objectToInvokeOn = $this->directObject ?: new Reference($this->objectToInvokeReferenceName);
207+
if ($this->methodNameOrInterface instanceof InterfaceToCall) {
208+
$interfaceToCallReference = InterfaceToCallReference::fromInstance($this->methodNameOrInterface);
209+
} else {
210+
$className = $this->directObject ? \get_class($objectToInvokeOn) : $this->objectToInvokeReferenceName;
211+
$interfaceToCallReference = new InterfaceToCallReference($className, $this->getMethodName());
212+
}
213+
}
214+
215+
$interfaceToCall = $builder->getInterfaceToCall($interfaceToCallReference);
216+
217+
if (!$interfaceToCall->canReturnValue()) {
218+
throw InvalidArgumentException::create("Can't create transformer for {$interfaceToCall}, because method has no return value");
219+
}
220+
221+
return array($objectToInvokeOn, $interfaceToCallReference, $interfaceToCall);
222+
}
205223
}

0 commit comments

Comments
 (0)