fix(core): handle objects in headers

springwolf-core/src/main/java/io/github/springwolf/core/asyncapi/scanners/common/headers/HeaderClassExtractor.java

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 package io.github.springwolf.core.asyncapi.scanners.common.headers;
 
+import io.github.springwolf.asyncapi.v3.model.schema.Schema;
 import io.github.springwolf.asyncapi.v3.model.schema.SchemaObject;
 import io.github.springwolf.asyncapi.v3.model.schema.SchemaType;
 import io.github.springwolf.core.asyncapi.scanners.common.payload.PayloadSchemaObject;
@@ -33,12 +34,21 @@ public SchemaObject extractHeader(Method method, PayloadSchemaObject payload) {
                 Header headerAnnotation = argument.getAnnotation(Header.class);
                 String headerName = getHeaderAnnotationName(headerAnnotation);
 
-                SchemaObject schema = schemaService
-                        .extractSchema(argument.getType())
-                        .rootSchema()
-                        .getSchema();
+                SwaggerSchemaService.ExtractedSchemas extractedSchema = schemaService.extractSchema(argument.getType());
+                Schema schema = extractedSchema.rootSchema().getSchema();
+                if (schema == null && extractedSchema.referencedSchemas().size() == 1) {
+                    schema = extractedSchema
+                            .referencedSchemas()
+                            .values()
+                            .iterator()
+                            .next();
+                }
 
-                headers.getProperties().put(headerName, schema);
+                if (schema != null) {
+                    headers.getProperties().put(headerName, schema);
+                } else {
+                    log.debug("Unable to extract schema for header {} in method {}", headerName, methodName);
+                }
             }
         }
 

springwolf-core/src/main/java/io/github/springwolf/core/controller/dtos/MessageDto.java

@@ -1,11 +1,17 @@
 // SPDX-License-Identifier: Apache-2.0
 package io.github.springwolf.core.controller.dtos;
 
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.extern.jackson.Jacksonized;
 
+import java.io.IOException;
 import java.util.Map;
 
 @Data
@@ -17,11 +23,27 @@ public class MessageDto {
 
     private final Map<String, String> bindings;
 
-    private final Map<String, String> headers;
+    private final Map<String, HeaderValue> headers;
 
     @Builder.Default
     private final String payloadType = String.class.getCanonicalName();
 
     @Builder.Default
     private final String payload = EMPTY;
+
+    @JsonDeserialize(using = HeaderValueDeserializer.class)
+    public record HeaderValue(String stringValue) {}
+
+    public static class HeaderValueDeserializer extends JsonDeserializer<HeaderValue> {
+        @Override
+        public HeaderValue deserialize(JsonParser p, DeserializationContext ctxt) throws IOException {
+            JsonNode node = p.getCodec().readTree(p);
+            if (node.isNumber()) {
+                return new HeaderValue(node.numberValue().toString());
+            } else if (node.isTextual()) {
+                return new HeaderValue(node.textValue());
+            }
+            return new HeaderValue(node.toString());
+        }
+    }
 }

springwolf-core/src/test/java/io/github/springwolf/core/asyncapi/controller/dtos/MessageDtoDeserializationTest.java

@@ -24,7 +24,8 @@ void testCanBeSerialized() throws IOException, ClassNotFoundException {
         MessageDto value = objectMapper.readValue(content, MessageDto.class);
 
         assertThat(value).isNotNull();
-        assertThat(value.getHeaders()).isEqualTo(singletonMap("some-header-key", "some-header-value"));
+        assertThat(value.getHeaders())
+                .isEqualTo(singletonMap("some-header-key", new MessageDto.HeaderValue("some-header-value")));
         assertThat(value.getPayload())
                 .isEqualTo(
                         new ObjectMapper().writeValueAsString(singletonMap("some-payload-key", "some-payload-value")));

springwolf-core/src/test/java/io/github/springwolf/core/asyncapi/scanners/common/headers/HeaderClassExtractorIntegrationTest.java

@@ -0,0 +1,136 @@
+// SPDX-License-Identifier: Apache-2.0
+package io.github.springwolf.core.asyncapi.scanners.common.headers;
+
+import io.github.springwolf.asyncapi.v3.model.components.ComponentSchema;
+import io.github.springwolf.asyncapi.v3.model.schema.SchemaObject;
+import io.github.springwolf.asyncapi.v3.model.schema.SchemaType;
+import io.github.springwolf.core.asyncapi.scanners.common.payload.PayloadSchemaObject;
+import io.github.springwolf.core.asyncapi.schemas.SwaggerSchemaService;
+import io.github.springwolf.core.asyncapi.schemas.SwaggerSchemaUtil;
+import io.github.springwolf.core.configuration.properties.SpringwolfConfigProperties;
+import lombok.val;
+import org.junit.jupiter.api.Test;
+import org.springframework.messaging.handler.annotation.Header;
+
+import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class HeaderClassExtractorIntegrationTest {
+
+    private final SwaggerSchemaService schemaService =
+            new SwaggerSchemaService(List.of(), List.of(), new SwaggerSchemaUtil(), new SpringwolfConfigProperties());
+    private final HeaderClassExtractor headerClassExtractor = new HeaderClassExtractor(schemaService);
+
+    private final PayloadSchemaObject payloadSchemaName = new PayloadSchemaObject(
+            "payloadSchemaName", String.class.getSimpleName(), ComponentSchema.of(new SchemaObject()));
+    private final SchemaObject stringSchema =
+            SchemaObject.builder().type(SchemaType.STRING).build();
+
+    @Test
+    void getNoDocumentedHeaders() throws NoSuchMethodException {
+        // when
+        Method m = TestClass.class.getDeclaredMethod("consumeWithoutHeadersAnnotation", String.class);
+        val result = headerClassExtractor.extractHeader(m, payloadSchemaName);
+
+        // then
+        assertEquals(AsyncHeadersNotDocumented.NOT_DOCUMENTED, result);
+    }
+
+    @Test
+    void getHeaderWithSingleHeaderAnnotation() throws NoSuchMethodException {
+        // when
+        Method m = TestClass.class.getDeclaredMethod("consumeWithSingleHeaderAnnotation", String.class);
+        val result = headerClassExtractor.extractHeader(m, payloadSchemaName);
+
+        // then
+        SchemaObject expectedHeaders = SchemaObject.builder()
+                .type(SchemaType.OBJECT)
+                .title("payloadSchemaNameHeaders")
+                .properties(new HashMap<>())
+                .build();
+        expectedHeaders.getProperties().put("kafka_receivedMessageKey", stringSchema);
+
+        assertEquals(expectedHeaders, result);
+    }
+
+    @Test
+    void getHeaderWithMultipleHeaderAnnotation() throws NoSuchMethodException {
+        // when
+        Method m = TestClass.class.getDeclaredMethod("consumeWithMultipleHeaderAnnotation", String.class, String.class);
+        val result = headerClassExtractor.extractHeader(m, payloadSchemaName);
+
+        // then
+        SchemaObject expectedHeaders = SchemaObject.builder()
+                .type(SchemaType.OBJECT)
+                .title("payloadSchemaNameHeaders")
+                .properties(new HashMap<>())
+                .build();
+        expectedHeaders.getProperties().put("kafka_receivedMessageKey", stringSchema);
+        expectedHeaders.getProperties().put("non-exist", stringSchema);
+
+        assertEquals(expectedHeaders, result);
+    }
+
+    @Test
+    void getHeaderWithObjectHeaderAnnotation() throws NoSuchMethodException {
+        // when
+        Method m = TestClass.class.getDeclaredMethod("consumeWithObjectHeaderAnnotation", TestClass.MyHeader.class);
+        val result = headerClassExtractor.extractHeader(m, payloadSchemaName);
+
+        // then
+        SchemaObject expectedHeaders = SchemaObject.builder()
+                .type(SchemaType.OBJECT)
+                .title("payloadSchemaNameHeaders")
+                .properties(new HashMap<>())
+                .build();
+        expectedHeaders
+                .getProperties()
+                .put(
+                        "myHeader",
+                        SchemaObject.builder()
+                                .type(SchemaType.OBJECT)
+                                .title("MyHeader")
+                                .properties(Map.of("key", ComponentSchema.of(stringSchema)))
+                                .build());
+
+        assertEquals(expectedHeaders, result);
+    }
+
+    @Test
+    void getHeaderWithNestedHeaderAnnotation() throws NoSuchMethodException {
+        // when
+        Method m = TestClass.class.getDeclaredMethod(
+                "consumeWithNestedObjectHeaderAnnotation", TestClass.MyNestedHeader.class);
+        val result = headerClassExtractor.extractHeader(m, payloadSchemaName);
+
+        // then
+        assertEquals(AsyncHeadersNotDocumented.NOT_DOCUMENTED, result); // currently not supported
+    }
+
+    public static class TestClass {
+
+        public void consumeWithoutHeadersAnnotation(String simpleValue) {}
+
+        public void consumeWithSingleHeaderAnnotation(@Header("kafka_receivedMessageKey") String key) {}
+
+        public void consumeWithMultipleHeaderAnnotation(
+                @Header("kafka_receivedMessageKey") String key,
+                @Header(name = "non-exist", defaultValue = "default") String nonExistingHeader) {}
+
+        public void consumeWithObjectHeaderAnnotation(@Header("myHeader") MyHeader key) {}
+
+        public void consumeWithNestedObjectHeaderAnnotation(@Header("myNestedHeader") MyNestedHeader key) {}
+
+        static class MyHeader {
+            public String key;
+        }
+
+        static class MyNestedHeader {
+            public MyHeader header;
+        }
+    }
+}

springwolf-examples/springwolf-kafka-example/src/main/java/io/github/springwolf/examples/kafka/consumers/ExampleClassLevelKafkaListener.java

@@ -10,6 +10,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.kafka.annotation.KafkaHandler;
 import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.listener.adapter.ConsumerRecordMetadata;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.handler.annotation.Payload;
@@ -30,6 +31,7 @@ public class ExampleClassLevelKafkaListener {
     public void receiveExamplePayload(
             @Header(KafkaHeaders.RECEIVED_KEY) String key,
             @Header(KafkaHeaders.OFFSET) Integer offset,
+            @Header(KafkaHeaders.RECORD_METADATA) ConsumerRecordMetadata meta,
             @Payload ExamplePayloadDto payload) {
         log.info("Received new message in {}: {}", TOPIC, payload.toString());
     }

springwolf-examples/springwolf-kafka-example/src/main/java/io/github/springwolf/examples/kafka/consumers/ExampleConsumer.java

@@ -7,6 +7,7 @@
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.listener.adapter.ConsumerRecordMetadata;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.handler.annotation.Payload;
@@ -25,6 +26,7 @@ public class ExampleConsumer {
     public void receiveExamplePayload(
             @Header(KafkaHeaders.RECEIVED_KEY) String key,
             @Header(KafkaHeaders.OFFSET) Integer offset,
+            @Header(KafkaHeaders.RECORD_METADATA) ConsumerRecordMetadata meta,
             @Payload ExamplePayloadDto payload) {
         log.info("Received new message in example-topic: {}", payload.toString());
 

springwolf-examples/springwolf-kafka-example/src/test/resources/asyncapi.json

@@ -444,8 +444,8 @@
           "type": "object"
         }
       },
-      "SpringKafkaDefaultHeaders-ExamplePayloadDto-1316119953": {
-        "title": "SpringKafkaDefaultHeaders-ExamplePayloadDto-1316119953",
+      "SpringKafkaDefaultHeaders-ExamplePayloadDto-1119017590": {
+        "title": "SpringKafkaDefaultHeaders-ExamplePayloadDto-1119017590",
         "type": "object",
         "properties": {
           "__TypeId__": {
@@ -469,13 +469,20 @@
             "examples": [
               "\"string\""
             ]
+          },
+          "kafka_recordMetadata": {
+            "type": "object",
+            "examples": [
+              { }
+            ]
           }
         },
         "examples": [
           {
             "__TypeId__": "io.github.springwolf.examples.kafka.dtos.ExamplePayloadDto",
             "kafka_offset": 0,
-            "kafka_receivedMessageKey": "string"
+            "kafka_receivedMessageKey": "string",
+            "kafka_recordMetadata": { }
           }
         ],
         "x-json-schema": {
@@ -493,9 +500,12 @@
             },
             "kafka_receivedMessageKey": {
               "type": "string"
+            },
+            "kafka_recordMetadata": {
+              "type": "object"
             }
           },
-          "title": "SpringKafkaDefaultHeaders-ExamplePayloadDto-1316119953",
+          "title": "SpringKafkaDefaultHeaders-ExamplePayloadDto-1119017590",
           "type": "object"
         }
       },
@@ -1723,7 +1733,7 @@
       },
       "io.github.springwolf.examples.kafka.dtos.ExamplePayloadDto": {
         "headers": {
-          "$ref": "#/components/schemas/SpringKafkaDefaultHeaders-ExamplePayloadDto-1316119953"
+          "$ref": "#/components/schemas/SpringKafkaDefaultHeaders-ExamplePayloadDto-1119017590"
         },
         "payload": {
           "schemaFormat": "application/vnd.aai.asyncapi+json;version=3.0.0",
@@ -2147,4 +2157,4 @@
       ]
     }
   }
-}
+}

springwolf-examples/springwolf-kafka-example/src/test/resources/asyncapi.yaml

@@ -304,8 +304,8 @@ components:
             type: string
         title: SpringKafkaDefaultHeaders-AnotherPayloadDto
         type: object
-    SpringKafkaDefaultHeaders-ExamplePayloadDto-1316119953:
-      title: SpringKafkaDefaultHeaders-ExamplePayloadDto-1316119953
+    SpringKafkaDefaultHeaders-ExamplePayloadDto-1119017590:
+      title: SpringKafkaDefaultHeaders-ExamplePayloadDto-1119017590
       type: object
       properties:
         __TypeId__:
@@ -323,10 +323,15 @@ components:
           type: string
           examples:
             - '"string"'
+        kafka_recordMetadata:
+          type: object
+          examples:
+            - {}
       examples:
         - __TypeId__: io.github.springwolf.examples.kafka.dtos.ExamplePayloadDto
           kafka_offset: 0
           kafka_receivedMessageKey: string
+          kafka_recordMetadata: {}
       x-json-schema:
         $schema: https://json-schema.org/draft-04/schema#
         properties:
@@ -339,7 +344,9 @@ components:
             type: integer
           kafka_receivedMessageKey:
             type: string
-        title: SpringKafkaDefaultHeaders-ExamplePayloadDto-1316119953
+          kafka_recordMetadata:
+            type: object
+        title: SpringKafkaDefaultHeaders-ExamplePayloadDto-1119017590
         type: object
     SpringKafkaDefaultHeaders-Message:
       title: SpringKafkaDefaultHeaders-Message
@@ -1237,7 +1244,7 @@ components:
           bindingVersion: 0.5.0
     io.github.springwolf.examples.kafka.dtos.ExamplePayloadDto:
       headers:
-        $ref: "#/components/schemas/SpringKafkaDefaultHeaders-ExamplePayloadDto-1316119953"
+        $ref: "#/components/schemas/SpringKafkaDefaultHeaders-ExamplePayloadDto-1119017590"
       payload:
         schemaFormat: application/vnd.aai.asyncapi+json;version=3.0.0
         schema:
@@ -1501,4 +1508,4 @@ operations:
       kafka:
         bindingVersion: 0.5.0
     messages:
-      - $ref: "#/channels/yaml-topic/messages/io.github.springwolf.examples.kafka.dtos.YamlPayloadDto"
+      - $ref: "#/channels/yaml-topic/messages/io.github.springwolf.examples.kafka.dtos.YamlPayloadDto"