Allow Error Handler to execute asynchronously

packages/Dbal/tests/Integration/DbalErrorChannelCommandBusTest.php

@@ -12,6 +12,7 @@
 use Ecotone\Messaging\Config\ModulePackageList;
 use Ecotone\Messaging\Config\ServiceConfiguration;
 use Ecotone\Messaging\Endpoint\ExecutionPollingMetadata;
+use Ecotone\Messaging\Handler\MessageHandlingException;
 use Ecotone\Messaging\Handler\Recoverability\ErrorContext;
 use Ecotone\Modelling\Attribute\InstantRetry;
 use Ecotone\Test\LicenceTesting;
@@ -189,7 +190,7 @@ public function test_it_fails_on_using_asynchronous_retry_on_synchronous_dead_le
         $exception = false;
         try {
             $commandBus->sendWithRouting('order.place', 'coffee');
-        } catch (InvalidArgumentException) {
+        } catch (MessageHandlingException) {
             $exception = true;
         }
 

packages/Ecotone/src/Messaging/Config/Annotation/ModuleConfiguration/ErrorHandlerModule.php

@@ -16,7 +16,7 @@
 use Ecotone\Messaging\Handler\InterfaceToCallRegistry;
 use Ecotone\Messaging\Handler\Logger\LoggingGateway;
 use Ecotone\Messaging\Handler\Processor\MethodInvoker\Converter\ReferenceBuilder;
-use Ecotone\Messaging\Handler\Recoverability\ErrorHandler;
+use Ecotone\Messaging\Handler\Recoverability\DelayedRetryErrorHandler;
 use Ecotone\Messaging\Handler\Recoverability\ErrorHandlerConfiguration;
 use Ecotone\Messaging\Handler\Router\HeaderRouter;
 use Ecotone\Messaging\Handler\Router\RouterBuilder;
@@ -57,7 +57,7 @@ public function prepare(Configuration $messagingConfiguration, array $extensionO
             }
 
             $errorHandler = ServiceActivatorBuilder::createWithDefinition(
-                new Definition(ErrorHandler::class, [
+                new Definition(DelayedRetryErrorHandler::class, [
                     $extensionObject->getDelayedRetryTemplate(),
                     (bool)$extensionObject->getDeadLetterQueueChannel(),
                     Reference::to(LoggingGateway::class),

packages/Ecotone/src/Messaging/Handler/Gateway/GatewayInternalProcessor.php

@@ -4,6 +4,7 @@
 
 use Ecotone\Messaging\Channel\QueueChannel;
 use Ecotone\Messaging\Future;
+use Ecotone\Messaging\Handler\MessageHandlingException;
 use Ecotone\Messaging\Handler\MessageProcessor;
 use Ecotone\Messaging\Handler\Processor\MethodInvoker\AroundInterceptable;
 use Ecotone\Messaging\Handler\Type;
@@ -104,7 +105,10 @@ private function getReply(PollableChannel $replyChannel): callable
                 throw InvalidArgumentException::create("{$this->interfaceToCallName} expects value, but null was returned. Have you consider changing return value to nullable?");
             }
             if ($replyMessage instanceof ErrorMessage) {
-                throw $replyMessage->getException();
+                throw MessageHandlingException::create(
+                    $replyMessage->getExceptionMessage(),
+                    $replyMessage->getExceptionCode(),
+                );
             }
 
             return $replyMessage;

packages/Ecotone/src/Messaging/Handler/Processor/MethodInvoker/Converter/MessageConverter.php

@@ -5,7 +5,9 @@
 namespace Ecotone\Messaging\Handler\Processor\MethodInvoker\Converter;
 
 use Ecotone\Messaging\Handler\ParameterConverter;
+use Ecotone\Messaging\Handler\Recoverability\ErrorContext;
 use Ecotone\Messaging\Message;
+use Ecotone\Messaging\Support\ErrorMessage;
 
 /**
  * Class MessageArgument
@@ -28,6 +30,10 @@ public static function create(): self
      */
     public function getArgumentFrom(Message $message): Message
     {
+        if (ErrorMessage::isErrorMessage($message)) {
+            return ErrorMessage::createFromMessage($message);
+        }
+
         return $message;
     }
 }

packages/Ecotone/src/Messaging/Handler/Recoverability/ErrorHandler.php → packages/Ecotone/src/Messaging/Handler/Recoverability/DelayedRetryErrorHandler.php

@@ -7,22 +7,19 @@
 use Ecotone\Messaging\Attribute\Parameter\Reference;
 use Ecotone\Messaging\Handler\ChannelResolver;
 use Ecotone\Messaging\Handler\Logger\LoggingGateway;
+use Ecotone\Messaging\Handler\MessageHandlingException;
 use Ecotone\Messaging\Message;
 use Ecotone\Messaging\MessageChannel;
 use Ecotone\Messaging\MessageHeaders;
+use Ecotone\Messaging\Support\ErrorMessage;
 use Ecotone\Messaging\Support\MessageBuilder;
 
 /**
  * licence Apache-2.0
  */
-class ErrorHandler
+class DelayedRetryErrorHandler
 {
     public const ECOTONE_RETRY_HEADER = 'ecotone_retry_number';
-    public const EXCEPTION_STACKTRACE = 'exception-stacktrace';
-    public const EXCEPTION_FILE = 'exception-file';
-    public const EXCEPTION_LINE = 'exception-line';
-    public const EXCEPTION_CODE = 'exception-code';
-    public const EXCEPTION_MESSAGE = 'exception-message';
 
     public function __construct(
         private RetryTemplate $delayedRetryTemplate,
@@ -32,23 +29,21 @@ public function __construct(
     }
 
     public function handle(
-        Message $errorMessage,
+        ErrorMessage $errorMessage,
         ChannelResolver $channelResolver,
         #[Reference] LoggingGateway $logger
     ): ?Message {
         $failedMessage = $errorMessage;
-        $cause = $errorMessage->getHeaders()->get(ErrorContext::EXCEPTION);
         $retryNumber = $failedMessage->getHeaders()->containsKey(self::ECOTONE_RETRY_HEADER) ? $failedMessage->getHeaders()->get(self::ECOTONE_RETRY_HEADER) + 1 : 1;
 
         if (! $failedMessage->getHeaders()->containsKey(MessageHeaders::POLLED_CHANNEL_NAME)) {
             $this->loggingGateway->error(
                 'Failed to handle Error Message via your Retry Configuration, as it does not contain information about origination channel from which it was polled.
                     This means that most likely Synchronous Dead Letter is configured with Retry Configuration which works only for Asynchronous configuration.',
                 $failedMessage,
-                ['exception' => $cause],
             );
 
-            throw $cause;
+            throw MessageHandlingException::create('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: ' . $failedMessage->getExceptionMessage());
         }
         /** @var MessageChannel $messageChannel */
         $messageChannel = $channelResolver->resolve($failedMessage->getHeaders()->get(MessageHeaders::POLLED_CHANNEL_NAME));
@@ -62,7 +57,6 @@ public function handle(
             MessageHeaders::DELIVERY_DELAY,
             MessageHeaders::TIME_TO_LIVE,
             MessageHeaders::CONSUMER_ACK_HEADER_LOCATION,
-            ErrorContext::EXCEPTION,
             self::ECOTONE_RETRY_HEADER,
         ]);
 
@@ -73,10 +67,10 @@ public function handle(
                         'Discarding message %s as no dead letter channel was defined. Retried maximum number of `%s` times. Due to: %s',
                         $failedMessage->getHeaders()->getMessageId(),
                         $retryNumber,
-                        $cause->getMessage()
+                        $errorMessage->getExceptionMessage()
                     ),
                     $failedMessage,
-                    ['exception' => $cause],
+                    $errorMessage->getErrorContext()->toArray(),
                 );
 
                 return null;
@@ -87,10 +81,10 @@ public function handle(
                     'Sending message `%s` to dead letter channel, as retried maximum number of `%s` times. Due to: %s',
                     $failedMessage->getHeaders()->getMessageId(),
                     $retryNumber,
-                    $cause->getMessage()
+                    $errorMessage->getExceptionMessage()
                 ),
                 $failedMessage,
-                ['exception' => $cause],
+                [ErrorContext::EXCEPTION_CLASS => $errorMessage->getExceptionClass()],
             );
 
             return $messageBuilder->build();
@@ -105,10 +99,10 @@ public function handle(
                 $this->delayedRetryTemplate->getMaxAttempts()
                     ? sprintf('Try %d out of %s', $retryNumber, $this->delayedRetryTemplate->getMaxAttempts())
                     : '',
-                $cause->getMessage()
+                $errorMessage->getExceptionMessage()
             ),
             $failedMessage,
-            ['exception' => $cause],
+            [ErrorContext::EXCEPTION_CLASS => $errorMessage->getExceptionClass()],
         );
         $messageChannel->send(
             $messageBuilder

packages/Ecotone/src/Messaging/Handler/Recoverability/ErrorContext.php

@@ -10,7 +10,7 @@
 class ErrorContext
 {
     public const WHOLE_ERROR_CONTEXT = [
-        self::EXCEPTION,
+        self::EXCEPTION_CLASS,
         self::EXCEPTION_STACKTRACE,
         self::EXCEPTION_FILE,
         self::EXCEPTION_LINE,
@@ -19,13 +19,15 @@ class ErrorContext
     ];
 
     public const EXCEPTION = 'exception';
-    public const EXCEPTION_STACKTRACE = 'exception-stacktrace';
-    public const EXCEPTION_FILE = 'exception-file';
-    public const EXCEPTION_LINE = 'exception-line';
-    public const EXCEPTION_CODE = 'exception-code';
-    public const EXCEPTION_MESSAGE = 'exception-message';
+    public const EXCEPTION_MESSAGE = 'ecotone.exception.message';
+    public const EXCEPTION_CLASS = 'ecotone.exception.class';
+    public const EXCEPTION_STACKTRACE = 'ecotone.exception.stacktrace';
+    public const EXCEPTION_FILE = 'ecotone.exception.file';
+    public const EXCEPTION_LINE = 'ecotone.exception.line';
+    public const EXCEPTION_CODE = 'ecotone.exception.code';
     public const DLQ_MESSAGE_REPLIED = 'ecotone.dlq.message_replied';
 
+    private string $exceptionClass;
     private string $messageId;
     private int $failedTimestamp;
     private string $stackTrace;
@@ -34,10 +36,11 @@ class ErrorContext
     private string $code;
     private string $message;
 
-    private function __construct(string $messageId, int $failedTimestamp, string $message, string $stackTrace, string $code, string $file, string $line)
+    private function __construct(string $messageId, int $failedTimestamp, string $exceptionClass, string $message, string $stackTrace, string $code, string $file, string $line)
     {
         $this->messageId = $messageId;
         $this->failedTimestamp = $failedTimestamp;
+        $this->exceptionClass = $exceptionClass;
         $this->stackTrace = $stackTrace;
         $this->file       = $file;
         $this->line       = $line;
@@ -50,6 +53,7 @@ public static function fromHeaders(array $messageHeaders): self
         return new self(
             $messageHeaders[MessageHeaders::MESSAGE_ID],
             $messageHeaders[MessageHeaders::TIMESTAMP],
+            $messageHeaders[self::EXCEPTION_CLASS] ?? '',
             $messageHeaders[self::EXCEPTION_MESSAGE],
             $messageHeaders[self::EXCEPTION_STACKTRACE],
             $messageHeaders[self::EXCEPTION_CODE],
@@ -68,6 +72,11 @@ public function getFailedTimestamp(): int
         return $this->failedTimestamp;
     }
 
+    public function getExceptionClass(): string
+    {
+        return $this->exceptionClass;
+    }
+
     public function getStackTrace(): string
     {
         return $this->stackTrace;
@@ -92,4 +101,16 @@ public function getMessage(): string
     {
         return $this->message;
     }
+
+    public function toArray(): array
+    {
+        return [
+            self::EXCEPTION_CLASS => $this->exceptionClass,
+            self::EXCEPTION_MESSAGE => $this->message,
+            self::EXCEPTION_STACKTRACE => $this->stackTrace,
+            self::EXCEPTION_FILE => $this->file,
+            self::EXCEPTION_LINE => $this->line,
+            self::EXCEPTION_CODE => $this->code,
+        ];
+    }
 }

packages/Ecotone/src/Messaging/MessageHeaders.php

@@ -178,7 +178,6 @@ public static function getFrameworksHeaderNames(): array
             self::STREAM_BASED_SOURCED,
             MessagingEntrypoint::ENTRYPOINT,
             self::CHANNEL_SEND_RETRY_NUMBER,
-            ErrorContext::EXCEPTION,
         ];
     }
 
@@ -252,7 +251,7 @@ public static function unsetEnqueueMetadata(array $metadata): array
             $metadata[self::POLLED_CHANNEL_NAME],
             $metadata[self::CONSUMER_POLLING_METADATA],
             $metadata[self::REPLY_CHANNEL],
-            $metadata[self::TEMPORARY_SPAN_CONTEXT_HEADER]
+            $metadata[self::TEMPORARY_SPAN_CONTEXT_HEADER],
         );
 
         return $metadata;

packages/Ecotone/src/Messaging/Support/ErrorMessage.php

@@ -5,6 +5,7 @@
 use Ecotone\Messaging\Handler\Recoverability\ErrorContext;
 use Ecotone\Messaging\Message;
 use Ecotone\Messaging\MessageHeaders;
+use Ecotone\Messaging\MessagingException;
 use Throwable;
 
 /**
@@ -22,11 +23,31 @@ private function __construct(
     ) {
     }
 
+    public static function createFromMessage(Message $message): self
+    {
+        if (!self::isErrorMessage($message)) {
+            throw MessagingException::create('Trying to create error message from message that is not generic message.');
+        }
+
+        return new self($message);
+    }
+
+    public static function isErrorMessage(Message $message): bool
+    {
+        foreach (ErrorContext::WHOLE_ERROR_CONTEXT as $errorContextKey) {
+            if (!$message->getHeaders()->containsKey($errorContextKey)) {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
     public static function create(Message $message, Throwable $cause): self
     {
         return new self(
             MessageBuilder::fromMessage($message)
-                ->setHeader(ErrorContext::EXCEPTION, $cause)
+                ->setHeader(ErrorContext::EXCEPTION_CLASS, get_class($cause))
                 ->setHeader(ErrorContext::EXCEPTION_MESSAGE, $cause->getMessage())
                 ->setHeader(ErrorContext::EXCEPTION_STACKTRACE, $cause->getTraceAsString())
                 ->setHeader(ErrorContext::EXCEPTION_FILE, $cause->getFile())
@@ -52,8 +73,38 @@ public function getPayload(): mixed
         return $this->message->getPayload();
     }
 
-    public function getException(): Throwable
+    public function getErrorContext(): ErrorContext
+    {
+        return ErrorContext::fromHeaders($this->getHeaders()->headers());
+    }
+
+    public function getExceptionClass(): string
+    {
+        return $this->getHeaders()->get(ErrorContext::EXCEPTION_CLASS);
+    }
+
+    public function getExceptionMessage(): string
+    {
+        return $this->getHeaders()->get(ErrorContext::EXCEPTION_MESSAGE);
+    }
+
+    public function getExceptionStackTrace(): string
+    {
+        return $this->getHeaders()->get(ErrorContext::EXCEPTION_STACKTRACE);
+    }
+
+    public function getExceptionFile(): string
+    {
+        return $this->getHeaders()->get(ErrorContext::EXCEPTION_FILE);
+    }
+
+    public function getExceptionLine(): string
+    {
+        return $this->getHeaders()->get(ErrorContext::EXCEPTION_LINE);
+    }
+
+    public function getExceptionCode(): string
     {
-        return $this->getHeaders()->get(ErrorContext::EXCEPTION);
+        return $this->getHeaders()->get(ErrorContext::EXCEPTION_CODE);
     }
 }

packages/Ecotone/tests/Messaging/Fixture/Handler/FailureHandler/FailureErrorHandler.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Messaging\Fixture\Handler\FailureHandler;
+
+use Ecotone\Messaging\Attribute\Asynchronous;
+use Ecotone\Messaging\Attribute\InternalHandler;
+use Ecotone\Messaging\Support\ErrorMessage;
+use Monolog\ErrorHandler;
+
+final class FailureErrorHandler
+{
+    private ?ErrorMessage $message = null;
+
+    #[Asynchronous('async')]
+    #[InternalHandler('errorHandler', endpointId: 'errorHandlerEndpoint')]
+    public function handle(ErrorMessage $message): void
+    {
+        $this->message = $message;
+    }
+
+    public function getMessage(): ?ErrorMessage
+    {
+        return $this->message;
+    }
+}