Implemented parsing of messages with array payloads (into schema obje…

…ct or list of schemas) in v3.

asyncapi-core/src/main/java/com/asyncapi/v3/jackson/SchemaItemsDeserializer.java

@@ -0,0 +1,51 @@
+package com.asyncapi.v3.jackson;
+
+import com.asyncapi.v3.schema.Schema;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.ObjectCodec;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.JsonNodeType;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+public class SchemaItemsDeserializer extends JsonDeserializer<Object> {
+
+    @Override
+    public Object deserialize(JsonParser jsonParser, DeserializationContext deserializationContext) throws IOException {
+        ObjectCodec objectCodec = jsonParser.getCodec();
+        JsonNode node = objectCodec.readTree(jsonParser);
+        JsonNodeType nodeType = node.getNodeType();
+        if (nodeType == JsonNodeType.OBJECT) {
+            return readAsSchema(node, objectCodec);
+        }
+        if (nodeType == JsonNodeType.ARRAY) {
+            return readAsListOfSchemas((ArrayNode) node, objectCodec);
+        }
+        return readAsObject(node, objectCodec);
+    }
+
+    private List<Schema> readAsListOfSchemas(ArrayNode arrayNode, ObjectCodec objectCodec) throws IOException {
+        List<Schema> schemaList = new ArrayList<>();
+        for (JsonNode childNode : arrayNode) {
+            schemaList.add(readAsSchema(childNode, objectCodec));
+        }
+        return schemaList;
+    }
+
+    private Schema readAsSchema(JsonNode jsonNode, ObjectCodec objectCodec) throws IOException {
+        try (JsonParser parser = jsonNode.traverse(objectCodec)) {
+            return parser.readValueAs(Schema.class);
+        }
+    }
+
+    private Object readAsObject(JsonNode jsonNode, ObjectCodec objectCodec) throws IOException {
+        try (JsonParser jsonParser = jsonNode.traverse(objectCodec)) {
+            return jsonParser.readValueAs(Object.class);
+        }
+    }
+}

asyncapi-core/src/main/java/com/asyncapi/v3/schema/Schema.java

@@ -1,5 +1,6 @@
 package com.asyncapi.v3.schema;
 
+import com.asyncapi.v3.jackson.SchemaItemsDeserializer;
 import com.asyncapi.v3.ExtendableObject;
 import com.asyncapi.v3.jackson.schema.SchemasAdditionalPropertiesDeserializer;
 import com.asyncapi.v3._0_0.model.ExternalDocumentation;
@@ -358,7 +359,7 @@ Validation Keywords for Numeric Instances (number and integer)
      * Omitting this keyword has the same behavior as an empty schema.
      */
     @Nullable
-    @JsonProperty
+    @JsonDeserialize(using = SchemaItemsDeserializer.class)
     public Object items;
 
     /**

asyncapi-core/src/test/kotlin/com/asyncapi/v3/_0_0/model/channel/message/MessageWithArrayPayloadTest.kt

@@ -0,0 +1,30 @@
+package com.asyncapi.v3._0_0.model.channel.message
+
+import com.asyncapi.v3.ClasspathUtils
+import com.asyncapi.v3.schema.Schema
+import com.fasterxml.jackson.databind.ObjectMapper
+import org.junit.jupiter.api.DisplayName
+import org.junit.jupiter.api.Test
+import kotlin.test.assertTrue
+
+class MessageWithArrayPayloadTest {
+    private val objectMapper = ObjectMapper()
+
+    @Test
+    @DisplayName("Test array items property is parsed as a schema object")
+    fun testArrayItemsPropertyIsParsedAsSchemaObjectWhenItIsASingleJsonSchema() {
+        val model = ClasspathUtils.readAsString("/json/v3/3.0.0/model/channel/message/messageWithArrayPayloadJsonSchema.json")
+        val schema = objectMapper.readValue(model, Message::class.java).payload as Schema
+        assertTrue(
+            schema.items is Schema
+        )
+    }
+
+    @Test
+    @DisplayName("Test array items property is parsed as list of schemas")
+    fun testArrayItemsPropertyIsParsedAsArrayListOfSchemasWhenItIsAnArrayOfSchemas() {
+        val model = ClasspathUtils.readAsString("/json/v3/3.0.0/model/channel/message/messageWithArrayPayloadArrayOfSchemas.json")
+        val schema = objectMapper.readValue(model, Message::class.java).payload as Schema
+        assertTrue(schema.items is ArrayList<*> && (schema.items as ArrayList<*>).all { it is Schema })
+    }
+}

asyncapi-core/src/test/resources/json/v3/3.0.0/model/channel/message/messageWithArrayPayloadArrayOfSchemas.json

@@ -0,0 +1,19 @@
+{
+  "bindings": {
+    "kafka": {
+      "key": {
+        "type": "string"
+      },
+      "bindingVersion": "0.4.0"
+    }
+  },
+  "payload": {
+    "type": "array",
+    "items": [
+      { "type": "number" },
+      { "type": "string" },
+      { "enum": ["Street", "Avenue", "Boulevard"] },
+      { "enum": ["NW", "NE", "SW", "SE"] }
+    ]
+  }
+}

asyncapi-core/src/test/resources/json/v3/3.0.0/model/channel/message/messageWithArrayPayloadJsonSchema.json

@@ -0,0 +1,32 @@
+{
+  "bindings": {
+    "kafka": {
+      "key": {
+        "type": "string"
+      },
+      "bindingVersion": "0.4.0"
+    }
+  },
+  "payload": {
+    "type": "array",
+    "items": {
+      "type": "object",
+      "required": [
+        "id",
+        "name",
+        "done"
+      ],
+      "properties": {
+        "id": {
+          "type": "integer"
+        },
+        "name": {
+          "type": "string"
+        },
+        "done": {
+          "type": "boolean"
+        }
+      }
+    }
+  }
+}