feat: add ContentType attribute for default message content type

packages/Ecotone/src/Messaging/Attribute/Endpoint/ContentType.php

@@ -0,0 +1,28 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ecotone\Messaging\Attribute\Endpoint;
+
+use Attribute;
+use Ecotone\Messaging\Conversion\MediaType;
+use Ecotone\Messaging\MessageHeaders;
+
+#[Attribute(Attribute::TARGET_CLASS | Attribute::TARGET_METHOD)]
+/**
+ * licence Apache-2.0
+ */
+class ContentType extends AddHeader
+{
+    public function __construct(string $contentType, private bool $replaceIfExists = false)
+    {
+        MediaType::parseMediaType($contentType);
+
+        parent::__construct(MessageHeaders::CONTENT_TYPE, $contentType);
+    }
+
+    public function shouldReplaceExistingHeader(): bool
+    {
+        return $this->replaceIfExists;
+    }
+}

packages/Ecotone/src/Messaging/Config/Annotation/ModuleConfiguration/EndpointHeaders/EndpointHeadersInterceptor.php

@@ -7,6 +7,7 @@
 use DateTimeImmutable;
 use DateTimeInterface;
 use Ecotone\Messaging\Attribute\Endpoint\AddHeader;
+use Ecotone\Messaging\Attribute\Endpoint\ContentType;
 use Ecotone\Messaging\Attribute\Endpoint\Delayed;
 use Ecotone\Messaging\Attribute\Endpoint\Priority;
 use Ecotone\Messaging\Attribute\Endpoint\RemoveHeader;
@@ -41,7 +42,7 @@ public function __construct(private ExpressionEvaluationService $expressionEvalu
 
     }
 
-    public function addMetadata(Message $message, ?AddHeader $addHeader, ?Delayed $delayed, ?Priority $priority, ?TimeToLive $timeToLive, ?RemoveHeader $removeHeader): array
+    public function addMetadata(Message $message, ?AddHeader $addHeader, ?Delayed $delayed, ?Priority $priority, ?TimeToLive $timeToLive, ?RemoveHeader $removeHeader, ?ContentType $contentType): array
     {
         $metadata = [];
 
@@ -56,6 +57,15 @@ public function addMetadata(Message $message, ?AddHeader $addHeader, ?Delayed $d
             }
         }
 
+        $isContentTypeHeaderExists = $message->getHeaders()->containsKey(MessageHeaders::CONTENT_TYPE);
+        if ($contentType) {
+            if ($contentType->shouldReplaceExistingHeader() || ! $isContentTypeHeaderExists) {
+                $metadata[MessageHeaders::CONTENT_TYPE] = $contentType->getHeaderValue();
+            } else {
+                $metadata[MessageHeaders::CONTENT_TYPE] = $message->getHeaders()->get(MessageHeaders::CONTENT_TYPE);
+            }
+        }
+
         $isDeliveryDelayHeaderExists = $message->getHeaders()->containsKey(MessageHeaders::DELIVERY_DELAY);
         if ($delayed && ($delayed->shouldReplaceExistingHeader() || ! $isDeliveryDelayHeaderExists)) {
             $metadata[MessageHeaders::DELIVERY_DELAY] = $delayed->getHeaderValue();

packages/Ecotone/tests/Messaging/Unit/Config/Annotation/ModuleConfiguration/ContentTypeAttributeTest.php

@@ -0,0 +1,120 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Messaging\Unit\Config\Annotation\ModuleConfiguration;
+
+use Ecotone\Lite\EcotoneLite;
+use Ecotone\Messaging\Attribute\Asynchronous;
+use Ecotone\Messaging\Attribute\Endpoint\ContentType;
+use Ecotone\Messaging\Channel\SimpleMessageChannelBuilder;
+use Ecotone\Messaging\Conversion\MediaType;
+use Ecotone\Messaging\Endpoint\ExecutionPollingMetadata;
+use Ecotone\Messaging\Support\MessageBuilder;
+use Ecotone\Modelling\Attribute\CommandHandler;
+use PHPUnit\Framework\TestCase;
+use Test\Ecotone\Messaging\Fixture\Handler\HeaderConversion\JsonConverter;
+
+/**
+ * licence Apache-2.0
+ * @internal
+ */
+final class ContentTypeAttributeTest extends TestCase
+{
+    public function test_using_provided_content_type_when_message_has_none(): void
+    {
+        $orderService = new class () {
+            public ?array $order = null;
+
+            #[ContentType('application/json')]
+            #[Asynchronous('async')]
+            #[CommandHandler('order.place', endpointId: 'orderPlaceEndpoint')]
+            public function place(array $order): void
+            {
+                $this->order = $order;
+            }
+        };
+
+        $ecotoneLite = EcotoneLite::bootstrapFlowTesting(
+            [get_class($orderService), JsonConverter::class],
+            [$orderService, new JsonConverter()],
+            enableAsynchronousProcessing: [
+                SimpleMessageChannelBuilder::createQueueChannel('async'),
+            ],
+        );
+
+        $ecotoneLite->sendMessageDirectToChannel(
+            'order.place',
+            MessageBuilder::withPayload('{"product":"Book"}')->build(),
+        );
+        $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup());
+
+        $this->assertSame(['product' => 'Book'], $orderService->order);
+    }
+
+    public function test_keeping_content_type_from_message_when_already_defined(): void
+    {
+        $orderService = new class () {
+            public ?array $order = null;
+
+            #[ContentType('application/json')]
+            #[Asynchronous('async')]
+            #[CommandHandler('order.place', endpointId: 'orderPlaceEndpoint')]
+            public function place(array $order): void
+            {
+                $this->order = $order;
+            }
+        };
+
+        $ecotoneLite = EcotoneLite::bootstrapFlowTesting(
+            [get_class($orderService), JsonConverter::class],
+            [$orderService, new JsonConverter()],
+            enableAsynchronousProcessing: [
+                SimpleMessageChannelBuilder::createQueueChannel('async'),
+            ],
+        );
+
+        $ecotoneLite->sendMessageDirectToChannel(
+            'order.place',
+            MessageBuilder::withPayload(serialize(['product' => 'Chair']))
+                ->setContentType(MediaType::createApplicationXPHPSerialized())
+                ->build(),
+        );
+        $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup());
+
+        $this->assertSame(['product' => 'Chair'], $orderService->order);
+    }
+
+    public function test_replacing_content_type_from_message_when_marked_to_replace_existing_one(): void
+    {
+        $orderService = new class () {
+            public ?array $order = null;
+
+            #[ContentType('application/json', replaceIfExists: true)]
+            #[Asynchronous('async')]
+            #[CommandHandler('order.place', endpointId: 'orderPlaceEndpoint')]
+            public function place(array $order): void
+            {
+                $this->order = $order;
+            }
+        };
+
+        $ecotoneLite = EcotoneLite::bootstrapFlowTesting(
+            [get_class($orderService), JsonConverter::class],
+            [$orderService, new JsonConverter()],
+            enableAsynchronousProcessing: [
+                SimpleMessageChannelBuilder::createQueueChannel('async'),
+            ],
+        );
+
+        $ecotoneLite->sendMessageDirectToChannel(
+            'order.place',
+            MessageBuilder::withPayload('{"product":"Table"}')
+                ->setContentType(MediaType::createApplicationXPHPSerialized())
+                ->build(),
+        );
+        $ecotoneLite->run('async', ExecutionPollingMetadata::createWithTestingSetup());
+
+        $this->assertSame(['product' => 'Table'], $orderService->order);
+    }
+}