fix: messages are not being generated for AsyncAPI replies (#1947)

src/processors/AsyncAPIInputProcessor.ts

@@ -7,7 +7,8 @@ import {
   SchemaInterface as AsyncAPISchemaInterface,
   SchemaV2 as AsyncAPISchema,
   fromFile,
-  createAsyncAPIDocument
+  createAsyncAPIDocument,
+  MessagesInterface
 } from '@asyncapi/parser';
 
 import { AbstractInputProcessor } from './AbstractInputProcessor';
@@ -122,37 +123,50 @@ export class AsyncAPIInputProcessor extends AbstractInputProcessor {
     if (channels.length) {
       for (const channel of doc.channels()) {
         for (const operation of channel.operations()) {
-          const operationMessages = operation.messages();
+          const handleMessages = (messages: MessagesInterface) => {
+            // treat multiple messages as oneOf
+            if (messages.length > 1) {
+              const oneOf: any[] = [];
 
-          // treat multiple messages as oneOf
-          if (operationMessages.length > 1) {
-            const oneOf: any[] = [];
+              for (const message of messages) {
+                const payload = message.payload();
 
-            for (const message of operationMessages) {
-              const payload = message.payload();
+                if (!payload) {
+                  continue;
+                }
 
-              if (!payload) {
-                continue;
+                oneOf.push(payload.json());
               }
 
-              oneOf.push(payload.json());
-            }
-
-            const payload = new AsyncAPISchema(
-              {
-                $id: channel.id(),
-                oneOf
-              },
-              channel.meta()
-            );
+              const payload = new AsyncAPISchema(
+                {
+                  $id: channel.id(),
+                  oneOf
+                },
+                channel.meta()
+              );
 
-            addToInputModel(payload);
-          } else if (operationMessages.length === 1) {
-            const payload = operationMessages[0].payload();
-            if (payload) {
               addToInputModel(payload);
+            } else if (messages.length === 1) {
+              const payload = messages[0].payload();
+              if (payload) {
+                addToInputModel(payload);
+              }
+            }
+          };
+          const replyOperation = operation.reply();
+          if (replyOperation !== undefined) {
+            const replyMessages = replyOperation.messages();
+            if (replyMessages.length > 0) {
+              handleMessages(replyMessages);
+            } else {
+              const replyChannelMessages = replyOperation.channel()?.messages();
+              if (replyChannelMessages) {
+                handleMessages(replyChannelMessages);
+              }
             }
           }
+          handleMessages(operation.messages());
         }
       }
     } else {

test/processors/AsyncAPIInputProcessor.spec.ts

@@ -20,10 +20,15 @@ const operationOneOf2DocString = fs.readFileSync(
   path.resolve(__dirname, './AsyncAPIInputProcessor/operation_oneof2.json'),
   'utf8'
 );
+const operationWithReply = fs.readFileSync(
+  path.resolve(__dirname, './AsyncAPIInputProcessor/operation_with_reply.json'),
+  'utf8'
+);
 const ymlFileURI = `file://${path.resolve(
   __dirname,
   './AsyncAPIInputProcessor/testasyncapi.yml'
 )}`;
+
 jest.mock('../../src/utils/LoggingInterface');
 
 describe('AsyncAPIInputProcessor', () => {
@@ -158,6 +163,13 @@ describe('AsyncAPIInputProcessor', () => {
       expect(commonInputModel instanceof InputMetaModel).toBeTruthy();
       expect(commonInputModel.models).toMatchSnapshot();
     });
+    test('should be able to process operation with reply', async () => {
+      const { document } = await parser.parse(operationWithReply);
+      const processor = new AsyncAPIInputProcessor();
+      const commonInputModel = await processor.process(document);
+      expect(commonInputModel instanceof InputMetaModel).toBeTruthy();
+      expect(commonInputModel.models).toMatchSnapshot();
+    });
 
     test('should be able to process operation with oneOf #1', async () => {
       const { document } = await parser.parse(operationOneOf1DocString);

test/processors/AsyncAPIInputProcessor/operation_with_reply.json

@@ -0,0 +1,94 @@
+{
+  "asyncapi": "3.0.0",
+  "info": {
+    "title": "Test",
+    "version": "1.0.0",
+    "description": "Test"
+  },
+  "defaultContentType": "application/json",
+  "servers": {
+    "local": {
+      "host": "localhost:1883",
+      "protocol": "mqtt"
+    }
+  },
+  "channels": {
+    "setParam": {
+      "address": "setParam",
+      "messages": {
+        "setParame": {
+          "$ref": "#/components/messages/setParam"
+        }
+      }
+    },
+    "setParamResult": {
+      "address": "setParamResult",
+      "messages": {
+        "setParamResult": {
+          "$ref": "#/components/messages/setParamResult"
+        }
+      }
+    }
+  },
+  "operations": {
+    "setParamOperation": {
+      "action": "send",
+      "channel": {
+        "$ref": "#/channels/setParam"
+      },
+      "reply": {
+        "channel": {
+          "$ref": "#/channels/setParamResult"
+        }
+      }
+    }
+  },
+  "components": {
+    "schemas": {
+      "correlationId": {
+        "type": "object",
+        "additionalProperties": false,
+        "title": "correlationId",
+        "required": [
+          "id"
+        ],
+        "properties": {
+          "id": {
+            "type": "string"
+          }
+        }
+      },
+      "setParamPayload": {
+        "title": "setParam",
+        "required": [
+          "parameterList"
+        ],
+        "allOf": [
+          {
+            "$ref": "#/components/schemas/correlationId"
+          }
+        ]
+      },
+      "setParamResultPayload": {
+        "title": "setParamResult",
+        "allOf": [
+          {
+            "$ref": "#/components/schemas/correlationId"
+          }
+        ]
+      }
+    },
+    "messages": {
+      "setParam": {
+        "payload": {
+          "$ref": "#/components/schemas/setParamPayload"
+        }
+      },
+      "setParamResult": {
+        "payload": {
+          "$ref": "#/components/schemas/setParamResultPayload"
+        }
+      }
+    }
+  }
+}

test/processors/__snapshots__/AsyncAPIInputProcessor.spec.ts.snap

@@ -4036,6 +4036,108 @@ InputMetaModel {
 }
 `;
 
+exports[`AsyncAPIInputProcessor process() should be able to process operation with reply 1`] = `
+Object {
+  "setParam": ObjectModel {
+    "name": "setParam",
+    "options": Object {
+      "isNullable": false,
+    },
+    "originalInput": AsyncapiV2Schema {
+      "allOf": Array [
+        AsyncapiV2Schema {
+          "additionalProperties": false,
+          "properties": Object {
+            "id": AsyncapiV2Schema {
+              "type": "string",
+              "x-modelgen-inferred-name": "anonymous_schema_1",
+              "x-parser-schema-id": "<anonymous-schema-1>",
+            },
+          },
+          "required": Array [
+            "id",
+          ],
+          "title": "correlationId",
+          "type": "object",
+          "x-modelgen-inferred-name": "correlationId",
+          "x-parser-schema-id": "correlationId",
+        },
+      ],
+      "required": Array [
+        "parameterList",
+      ],
+      "title": "setParam",
+      "x-modelgen-inferred-name": "setParamPayload",
+      "x-parser-schema-id": "setParamPayload",
+    },
+    "properties": Object {
+      "id": ObjectPropertyModel {
+        "property": StringModel {
+          "name": "anonymous_schema_1",
+          "options": Object {
+            "isNullable": false,
+          },
+          "originalInput": AsyncapiV2Schema {
+            "type": "string",
+            "x-modelgen-inferred-name": "anonymous_schema_1",
+            "x-parser-schema-id": "<anonymous-schema-1>",
+          },
+        },
+        "propertyName": "id",
+        "required": true,
+      },
+    },
+  },
+  "setParamResult": ObjectModel {
+    "name": "setParamResult",
+    "options": Object {
+      "isNullable": false,
+    },
+    "originalInput": AsyncapiV2Schema {
+      "allOf": Array [
+        AsyncapiV2Schema {
+          "additionalProperties": false,
+          "properties": Object {
+            "id": AsyncapiV2Schema {
+              "type": "string",
+              "x-modelgen-inferred-name": "anonymous_schema_1",
+              "x-parser-schema-id": "<anonymous-schema-1>",
+            },
+          },
+          "required": Array [
+            "id",
+          ],
+          "title": "correlationId",
+          "type": "object",
+          "x-modelgen-inferred-name": "correlationId",
+          "x-parser-schema-id": "correlationId",
+        },
+      ],
+      "title": "setParamResult",
+      "x-modelgen-inferred-name": "setParamResultPayload",
+      "x-parser-schema-id": "setParamResultPayload",
+    },
+    "properties": Object {
+      "id": ObjectPropertyModel {
+        "property": StringModel {
+          "name": "anonymous_schema_1",
+          "options": Object {
+            "isNullable": false,
+          },
+          "originalInput": AsyncapiV2Schema {
+            "type": "string",
+            "x-modelgen-inferred-name": "anonymous_schema_1",
+            "x-parser-schema-id": "<anonymous-schema-1>",
+          },
+        },
+        "propertyName": "id",
+        "required": true,
+      },
+    },
+  },
+}
+`;
+
 exports[`AsyncAPIInputProcessor process() should be able to process parsed objects 1`] = `
 InputMetaModel {
   "models": Object {