diff --git a/packages/Dbal/Api/ExtensionObject/DbalDeadLetterBuilder.php b/packages/Dbal/Api/ExtensionObject/DbalDeadLetterBuilder.php index bb437a9814..2a6dd0213e 100644 --- a/packages/Dbal/Api/ExtensionObject/DbalDeadLetterBuilder.php +++ b/packages/Dbal/Api/ExtensionObject/DbalDeadLetterBuilder.php @@ -11,6 +11,7 @@ use Ecotone\Messaging\Config\Container\MessagingContainerBuilder; use Ecotone\Messaging\Config\Container\Reference; use Ecotone\Messaging\Gateway\MessagingEntrypointService; +use Ecotone\Messaging\Handler\ChannelResolver; use Ecotone\Messaging\Handler\InputOutputMessageHandlerBuilder; use Ecotone\Messaging\Handler\InterfaceToCall; use Ecotone\Messaging\Handler\InterfaceToCallRegistry; @@ -27,6 +28,7 @@ class DbalDeadLetterBuilder extends InputOutputMessageHandlerBuilder { public const LIMIT_HEADER = 'ecotone_dbal_deadletter_limit'; public const OFFSET_HEADER = 'ecotone_dbal_deadletter_offset'; + public const TARGET_CHANNEL_HEADER = 'ecotone_dbal_deadletter_target_channel'; public const LIST_CHANNEL = 'ecotone.dbal.deadletter.list'; public const SHOW_CHANNEL = 'ecotone.dbal.deadletter.show'; @@ -101,7 +103,11 @@ public static function createReplay(string $referenceName): self 'replay', $referenceName, self::getChannelName($referenceName, self::REPLAY_CHANNEL), - [] + [ + PayloadBuilder::create('messageId'), + ReferenceBuilder::create('messagingEntrypoint', MessagingEntrypointService::class), + HeaderBuilder::createOptional('targetChannel', self::TARGET_CHANNEL_HEADER), + ] ); } @@ -113,6 +119,7 @@ public static function createReplayAll(string $referenceName): self self::getChannelName($referenceName, self::REPLAY_ALL_CHANNEL), [ ReferenceBuilder::create('messagingEntrypoint', MessagingEntrypointService::class), + HeaderBuilder::createOptional('targetChannel', self::TARGET_CHANNEL_HEADER), ] ); } @@ -160,6 +167,7 @@ public function compile(MessagingContainerBuilder $builder): Definition Reference::to(self::storeReferenceFor($this->referenceName)), Reference::to(MessageBuilder::class), Reference::to(FlowTrackingScope::class), + Reference::to(ChannelResolver::class), ])); } diff --git a/packages/Dbal/Api/Gateway/DeadLetterGateway.php b/packages/Dbal/Api/Gateway/DeadLetterGateway.php index 3e3f87ecad..0bbade2413 100644 --- a/packages/Dbal/Api/Gateway/DeadLetterGateway.php +++ b/packages/Dbal/Api/Gateway/DeadLetterGateway.php @@ -21,9 +21,9 @@ public function show(string $messageId): Message; public function count(): int; - public function replay(string|array $messageId): void; + public function replay(string|array $messageId, ?string $targetChannel = null): void; - public function replayAll(): void; + public function replayAll(?string $targetChannel = null): void; public function delete(string|array $messageId): void; diff --git a/packages/Dbal/src/Recoverability/DbalDeadLetterConsoleCommand.php b/packages/Dbal/src/Recoverability/DbalDeadLetterConsoleCommand.php index 8e12dfaab7..6fe9fae3ad 100644 --- a/packages/Dbal/src/Recoverability/DbalDeadLetterConsoleCommand.php +++ b/packages/Dbal/src/Recoverability/DbalDeadLetterConsoleCommand.php @@ -21,8 +21,8 @@ public function help(): ConsoleCommandResultSet [ [DbalDeadLetterModule::LIST_COMMAND_NAME . ' {page:int}', 'List all dead messages'], [DbalDeadLetterModule::SHOW_COMMAND_NAME . ' {messageId:string} {fullDetails:bool}', 'Show more details about specific message'], - [DbalDeadLetterModule::REPLAY_COMMAND_NAME . ' {messageId:string}', 'Replay chosen dead message'], - [DbalDeadLetterModule::REPLAY_ALL_COMMAND_NAME, 'Replay ALL dead messages'], + [DbalDeadLetterModule::REPLAY_COMMAND_NAME . ' {messageId:string} {targetChannel:?string}', 'Replay chosen dead message, to its source channel or to the target channel'], + [DbalDeadLetterModule::REPLAY_ALL_COMMAND_NAME . ' {targetChannel:?string}', 'Replay ALL dead messages, to their source channels or to the target channel'], [DbalDeadLetterModule::DELETE_COMMAND_NAME . ' {messageId:string}', 'Delete chosen dead message'], ] ); @@ -65,9 +65,9 @@ public function show(DeadLetterGateway $deadLetterGateway, string $messageId, bo ); } - public function replay(DeadLetterGateway $deadLetterGateway, string $messageId): void + public function replay(DeadLetterGateway $deadLetterGateway, string $messageId, ?string $targetChannel = null): void { - $deadLetterGateway->replay($messageId); + $deadLetterGateway->replay($messageId, $targetChannel); } public function delete(DeadLetterGateway $deadLetterGateway, string $messageId): void @@ -75,9 +75,9 @@ public function delete(DeadLetterGateway $deadLetterGateway, string $messageId): $deadLetterGateway->delete($messageId); } - public function replayAll(DeadLetterGateway $deadLetterGateway): void + public function replayAll(DeadLetterGateway $deadLetterGateway, ?string $targetChannel = null): void { - $deadLetterGateway->replayAll(); + $deadLetterGateway->replayAll($targetChannel); } private function convertTimestampToReadableFormat(int $timestamp) diff --git a/packages/Dbal/src/Recoverability/DbalDeadLetterModule.php b/packages/Dbal/src/Recoverability/DbalDeadLetterModule.php index 9ecf0fc906..ba78fb3bdd 100644 --- a/packages/Dbal/src/Recoverability/DbalDeadLetterModule.php +++ b/packages/Dbal/src/Recoverability/DbalDeadLetterModule.php @@ -210,6 +210,9 @@ private function registerGateway(string $referenceName, string $connectionFactor 'replay', DbalDeadLetterBuilder::getChannelName($referenceName, DbalDeadLetterBuilder::REPLAY_CHANNEL) ) + ->withParameterConverters([ + GatewayHeaderBuilder::create('targetChannel', DbalDeadLetterBuilder::TARGET_CHANNEL_HEADER), + ]) ) ->registerGatewayBuilder( GatewayProxyBuilder::create( @@ -218,6 +221,9 @@ private function registerGateway(string $referenceName, string $connectionFactor 'replayAll', DbalDeadLetterBuilder::getChannelName($referenceName, DbalDeadLetterBuilder::REPLAY_ALL_CHANNEL) ) + ->withParameterConverters([ + GatewayHeaderBuilder::create('targetChannel', DbalDeadLetterBuilder::TARGET_CHANNEL_HEADER), + ]) ) ->registerGatewayBuilder( GatewayProxyBuilder::create( diff --git a/packages/Dbal/src/Recoverability/DeadLetterHandler.php b/packages/Dbal/src/Recoverability/DeadLetterHandler.php index 9ea22e7544..cdc5176c0a 100644 --- a/packages/Dbal/src/Recoverability/DeadLetterHandler.php +++ b/packages/Dbal/src/Recoverability/DeadLetterHandler.php @@ -11,6 +11,7 @@ use Ecotone\Dbal\Tracing\FlowTrackingScope; use Ecotone\Messaging\Config\Annotation\ModuleConfiguration\ErrorChannelExceptionMessages; use Ecotone\Messaging\Gateway\MessagingEntrypointService; +use Ecotone\Messaging\Handler\ChannelResolver; use Ecotone\Messaging\Support\InvalidArgumentException; use Ecotone\Messaging\Support\MessageBuilder; @@ -23,6 +24,7 @@ public function __construct( private DeadLetterStore $deadLetterStore, private MessageBuilder $messageBuilder, private FlowTrackingScope $flowTracking, + private ChannelResolver $channelResolver, ) { } @@ -60,18 +62,22 @@ public function count(): int return $this->deadLetterStore->count(); } - public function replay(string|array $messageId, MessagingEntrypointService $messagingEntrypoint): void + public function replay(string|array $messageId, MessagingEntrypointService $messagingEntrypoint, ?string $targetChannel = null): void { + $this->assertTargetChannelExists($targetChannel); + foreach ((array) $messageId as $id) { - $this->replayMessage($id, $messagingEntrypoint); + $this->replayMessage($id, $messagingEntrypoint, $targetChannel); } } - public function replayAll(MessagingEntrypointService $messagingEntrypoint): void + public function replayAll(MessagingEntrypointService $messagingEntrypoint, ?string $targetChannel = null): void { + $this->assertTargetChannelExists($targetChannel); + while ($errorContexts = $this->deadLetterStore->list(100, 0)) { foreach ($errorContexts as $errorContext) { - $this->replayMessage($errorContext->getMessageId(), $messagingEntrypoint); + $this->replayMessage($errorContext->getMessageId(), $messagingEntrypoint, $targetChannel); } } } @@ -93,18 +99,31 @@ public function store(Message $message): void $this->deadLetterStore->store($message); } - private function replayMessage(string $messageId, MessagingEntrypointService $messagingEntrypoint): void + private function assertTargetChannelExists(?string $targetChannel): void + { + if ($targetChannel !== null && ! $this->channelResolver->hasChannelWithName($targetChannel)) { + throw InvalidArgumentException::create(ErrorChannelExceptionMessages::cannotReplayToUnknownChannel($targetChannel)); + } + } + + private function replayMessage(string $messageId, MessagingEntrypointService $messagingEntrypoint, ?string $targetChannel): void { $message = $this->show($messageId); $hasPolledChannel = $message->getHeaders()->containsKey(MessageHeaders::POLLED_CHANNEL_NAME); $hasInboundRequestChannel = $message->getHeaders()->containsKey(MessageHeaders::INBOUND_REQUEST_CHANNEL); $hasRoutingSlip = $message->getHeaders()->containsKey(MessageHeaders::ROUTING_SLIP); - if (! $hasPolledChannel && ! $hasInboundRequestChannel && ! $hasRoutingSlip) { + if ($targetChannel === null && ! $hasPolledChannel && ! $hasInboundRequestChannel && ! $hasRoutingSlip) { throw InvalidArgumentException::create(ErrorChannelExceptionMessages::cannotReplyToDeadLetterMessage($messageId)); } - if ($hasPolledChannel) { + if ($targetChannel !== null) { + $entrypoint = $targetChannel; + + $message = $this->messageBuilder->fromMessage($message) + ->removeHeader(MessageHeaders::ROUTING_SLIP) + ->build(); + } elseif ($hasPolledChannel) { $entrypoint = $message->getHeaders()->get(MessageHeaders::POLLED_CHANNEL_NAME); } elseif ($hasInboundRequestChannel) { $entrypoint = $message->getHeaders()->get(MessageHeaders::INBOUND_REQUEST_CHANNEL); diff --git a/packages/Dbal/tests/Integration/DeadLetterTest.php b/packages/Dbal/tests/Integration/DeadLetterTest.php index d3f8bd438a..36ee7f5e8c 100644 --- a/packages/Dbal/tests/Integration/DeadLetterTest.php +++ b/packages/Dbal/tests/Integration/DeadLetterTest.php @@ -287,6 +287,257 @@ classesToResolve: [$handler::class], $this->assertSame(['first-payload'], $handler->processedPayloads); } + public function test_dead_letter_message_is_replayed_to_the_chosen_channel_instead_of_the_source_one(): void + { + $handler = new class () { + public const REQUEST_CHANNEL = 'sourceRequestChannel'; + public const TARGET_CHANNEL = 'targetRequestChannel'; + + /** @var string[] */ + public array $handledBySource = []; + /** @var string[] */ + public array $handledByTarget = []; + private bool $hasEmitted = false; + + #[\Ecotone\Api\Attribute\Scheduled(self::REQUEST_CHANNEL, 'emittingAdapter')] + #[\Ecotone\Api\Attribute\Poller(executionTimeLimitInMilliseconds: 1, handledMessageLimit: 1)] + public function emit(): ?string + { + if ($this->hasEmitted) { + return null; + } + $this->hasEmitted = true; + + return 'first-payload'; + } + + #[\Ecotone\Api\Attribute\InternalHandler(self::REQUEST_CHANNEL)] + public function handleBySource(string $payload): void + { + $this->handledBySource[] = $payload; + + throw new RuntimeException('simulated'); + } + + #[\Ecotone\Api\Attribute\InternalHandler(self::TARGET_CHANNEL)] + public function handleByTarget(string $payload): void + { + $this->handledByTarget[] = $payload; + } + }; + $connectionFactory = $this->getConnectionFactory(); + + $ecotone = $this->bootstrapFlowTesting( + classesToResolve: [$handler::class], + containerOrAvailableServices: [ + $handler, + DbalConnectionFactory::class => $connectionFactory, + 'managerRegistry' => $connectionFactory, + ], + extensionObjects: [ + \Ecotone\Api\ExtensionObject\InstantRetryConfiguration::createWithDefaults()->withAsynchronousEndpointsRetry(false), + ServiceConfiguration::createWithDefaults()->withDefaultErrorChannel(DbalDeadLetterBuilder::STORE_CHANNEL), + ], + enabledModules: [ModulePackageList::DBAL_PACKAGE, ], + pathToRootCatalog: __DIR__ . '/../../', + environment: 'prod', + ); + + $ecotone->run('emittingAdapter', ExecutionPollingMetadata::createWithTestingSetup(handledMessageLimit: 1, stopOnError: false)); + $this->assertErrorMessageCount($ecotone, 1); + + $gateway = $ecotone->getGateway(DeadLetterGateway::class); + $gateway->replay($gateway->list(10, 0)[0]->getMessageId(), $handler::TARGET_CHANNEL); + + $this->assertErrorMessageCount($ecotone, 0); + $this->assertSame(['first-payload'], $handler->handledBySource); + $this->assertSame(['first-payload'], $handler->handledByTarget); + } + + public function test_all_dead_letter_messages_are_replayed_to_the_chosen_channel(): void + { + $handler = new class () { + public const TARGET_CHANNEL = 'allTargetChannel'; + + /** @var string[] */ + public array $handledByTarget = []; + private bool $hasEmitted = false; + + #[\Ecotone\Api\Attribute\Scheduled('allSourceChannel', 'allEmittingAdapter')] + #[\Ecotone\Api\Attribute\Poller(executionTimeLimitInMilliseconds: 1, handledMessageLimit: 1)] + public function emit(): ?string + { + if ($this->hasEmitted) { + return null; + } + $this->hasEmitted = true; + + return 'payload'; + } + + #[\Ecotone\Api\Attribute\InternalHandler('allSourceChannel')] + public function handleBySource(string $payload): void + { + throw new RuntimeException('simulated'); + } + + #[\Ecotone\Api\Attribute\InternalHandler(self::TARGET_CHANNEL)] + public function handleByTarget(string $payload): void + { + $this->handledByTarget[] = $payload; + } + }; + $connectionFactory = $this->getConnectionFactory(); + + $ecotone = $this->bootstrapFlowTesting( + classesToResolve: [$handler::class], + containerOrAvailableServices: [ + $handler, + DbalConnectionFactory::class => $connectionFactory, + 'managerRegistry' => $connectionFactory, + ], + extensionObjects: [ + \Ecotone\Api\ExtensionObject\InstantRetryConfiguration::createWithDefaults()->withAsynchronousEndpointsRetry(false), + ServiceConfiguration::createWithDefaults()->withDefaultErrorChannel(DbalDeadLetterBuilder::STORE_CHANNEL), + ], + enabledModules: [ModulePackageList::DBAL_PACKAGE, ], + pathToRootCatalog: __DIR__ . '/../../', + environment: 'prod', + ); + + $ecotone->run('allEmittingAdapter', ExecutionPollingMetadata::createWithTestingSetup(handledMessageLimit: 1, stopOnError: false)); + + $ecotone->getGateway(DeadLetterGateway::class)->replayAll($handler::TARGET_CHANNEL); + + $this->assertErrorMessageCount($ecotone, 0); + $this->assertSame(['payload'], $handler->handledByTarget); + } + + public function test_replaying_to_unknown_channel_is_refused_and_keeps_the_dead_letter_message(): void + { + $handler = new class () { + public const TARGET_CHANNEL = 'unknownTargetCheckChannel'; + + /** @var string[] */ + public array $handledByTarget = []; + private bool $hasEmitted = false; + + #[\Ecotone\Api\Attribute\Scheduled('unknownSourceChannel', 'unknownEmittingAdapter')] + #[\Ecotone\Api\Attribute\Poller(executionTimeLimitInMilliseconds: 1, handledMessageLimit: 1)] + public function emit(): ?string + { + if ($this->hasEmitted) { + return null; + } + $this->hasEmitted = true; + + return 'payload'; + } + + #[\Ecotone\Api\Attribute\InternalHandler('unknownSourceChannel')] + public function handleBySource(string $payload): void + { + throw new RuntimeException('simulated'); + } + + #[\Ecotone\Api\Attribute\InternalHandler(self::TARGET_CHANNEL)] + public function handleByTarget(string $payload): void + { + $this->handledByTarget[] = $payload; + } + }; + $connectionFactory = $this->getConnectionFactory(); + + $ecotone = $this->bootstrapFlowTesting( + classesToResolve: [$handler::class], + containerOrAvailableServices: [ + $handler, + DbalConnectionFactory::class => $connectionFactory, + 'managerRegistry' => $connectionFactory, + ], + extensionObjects: [ + \Ecotone\Api\ExtensionObject\InstantRetryConfiguration::createWithDefaults()->withAsynchronousEndpointsRetry(false), + ServiceConfiguration::createWithDefaults()->withDefaultErrorChannel(DbalDeadLetterBuilder::STORE_CHANNEL), + ], + enabledModules: [ModulePackageList::DBAL_PACKAGE, ], + pathToRootCatalog: __DIR__ . '/../../', + environment: 'prod', + ); + + $ecotone->run('unknownEmittingAdapter', ExecutionPollingMetadata::createWithTestingSetup(handledMessageLimit: 1, stopOnError: false)); + + try { + $ecotone->getGateway(DeadLetterGateway::class)->replayAll('missingChannel'); + $this->fail('Replay to an unknown channel should be refused'); + } catch (\Ecotone\Messaging\Support\InvalidArgumentException $exception) { + $this->assertStringContainsString('missingChannel', $exception->getMessage()); + $this->assertStringContainsString('omit the target channel', $exception->getMessage()); + } + + $this->assertErrorMessageCount($ecotone, 1); + $this->assertSame([], $handler->handledByTarget); + } + + public function test_console_command_replays_dead_letter_message_to_the_chosen_channel(): void + { + $handler = new class () { + public const TARGET_CHANNEL = 'consoleTargetChannel'; + + /** @var string[] */ + public array $handledByTarget = []; + private bool $hasEmitted = false; + + #[\Ecotone\Api\Attribute\Scheduled('consoleSourceChannel', 'consoleEmittingAdapter')] + #[\Ecotone\Api\Attribute\Poller(executionTimeLimitInMilliseconds: 1, handledMessageLimit: 1)] + public function emit(): ?string + { + if ($this->hasEmitted) { + return null; + } + $this->hasEmitted = true; + + return 'payload'; + } + + #[\Ecotone\Api\Attribute\InternalHandler('consoleSourceChannel')] + public function handleBySource(string $payload): void + { + throw new RuntimeException('simulated'); + } + + #[\Ecotone\Api\Attribute\InternalHandler(self::TARGET_CHANNEL)] + public function handleByTarget(string $payload): void + { + $this->handledByTarget[] = $payload; + } + }; + $connectionFactory = $this->getConnectionFactory(); + + $ecotone = $this->bootstrapFlowTesting( + classesToResolve: [$handler::class], + containerOrAvailableServices: [ + $handler, + DbalConnectionFactory::class => $connectionFactory, + 'managerRegistry' => $connectionFactory, + ], + extensionObjects: [ + \Ecotone\Api\ExtensionObject\InstantRetryConfiguration::createWithDefaults()->withAsynchronousEndpointsRetry(false), + ServiceConfiguration::createWithDefaults()->withDefaultErrorChannel(DbalDeadLetterBuilder::STORE_CHANNEL), + ], + enabledModules: [ModulePackageList::DBAL_PACKAGE, ], + pathToRootCatalog: __DIR__ . '/../../', + environment: 'prod', + ); + + $ecotone->run('consoleEmittingAdapter', ExecutionPollingMetadata::createWithTestingSetup(handledMessageLimit: 1, stopOnError: false)); + + $messageId = $ecotone->getGateway(DeadLetterGateway::class)->list(10, 0)[0]->getMessageId(); + $ecotone->runConsoleCommand('ecotone:deadletter:replay', ['messageId' => $messageId, 'targetChannel' => $handler::TARGET_CHANNEL]); + + $this->assertErrorMessageCount($ecotone, 0); + $this->assertSame(['payload'], $handler->handledByTarget); + } + private function assertErrorMessageCount(FlowTestSupport $ecotone, int $amount, string $deadLetterReference = DeadLetterGateway::class): void { $gateway = $ecotone->getGateway(DeadLetterGateway::class); diff --git a/packages/Ecotone/src/Messaging/Config/Annotation/ModuleConfiguration/ErrorChannelExceptionMessages.php b/packages/Ecotone/src/Messaging/Config/Annotation/ModuleConfiguration/ErrorChannelExceptionMessages.php index 9a6e319a14..eafaa6c68b 100644 --- a/packages/Ecotone/src/Messaging/Config/Annotation/ModuleConfiguration/ErrorChannelExceptionMessages.php +++ b/packages/Ecotone/src/Messaging/Config/Annotation/ModuleConfiguration/ErrorChannelExceptionMessages.php @@ -48,6 +48,13 @@ public static function cannotReplyToDeadLetterMessage(string $messageId): string . 'Please add one of them, so Message can be routed back to the original channel.'; } + public static function cannotReplayToUnknownChannel(string $targetChannel): string + { + return "Can not replay dead letter messages to channel `{$targetChannel}`, as no channel or handler with that name exists. " + . 'Pass the name of an existing message channel or of a handler input channel (e.g. the routing key of a #[CommandHandler] or the inputChannelName of an #[InternalHandler]), ' + . 'or omit the target channel to replay each message to the channel it was polled from.'; + } + public static function delayedRetryRequiresPolledChannelName(string $originalErrorMessage): string { return 'Failed to handle Error Message via Retry Configuration, as it does not contain information about origination channel from which it was polled. Original error message: '