feat: add kafka channel support (#178)

Author: jonaslagoni

docs/generators/channels.md

@@ -12,7 +12,7 @@ export default {
       preset: 'channels',
       outputPath: './src/__gen__/', 
       language: 'typescript',
-      protocols: ['nats']
+      protocols: ['nats', 'kafka']
     }
   ]
 };
@@ -26,7 +26,7 @@ This is supported through the following inputs: [`asyncapi`](#inputs)
 
 It supports the following languages; [`typescript`](#typescript)
 
-It supports the following protocols; [`nats`](../protocols/nats.md)
+It supports the following protocols; [`nats`](../protocols/nats.md), [`kafka`](../protocols/kafka.md) 
 
 ## Options
 These are the available options for the `channels` generator; 
@@ -41,6 +41,7 @@ These are the available options for the `channels` generator;
 
 Depending on which protocol, these are the dependencies:
 - `NATS`: https://github.com/nats-io/nats.js v2
+- `Kafka`: https://github.com/tulios/kafkajs v2
 
 For TypeScript what is generated is a single file that include functions to help easier interact with AsyncAPI channels. For example;
 

docs/protocols/kafka.md

@@ -0,0 +1,113 @@
+---
+sidebar_position: 99
+---
+
+# Kafka
+Kafka is currently supported through the following generators ([channels](../generators/channels.md)):
+
+| **Languages** | Publish | Subscribe
+|---|---|---|
+| TypeScript | ✔️ | ✔️ |
+
+All of this is available through [AsyncAPI](../inputs/asyncapi.md). If you use 
+
+## Channels
+Read more about the [channels](../generators/channels.md) generator here before continuing.
+
+This generator provides support functions for each resource ensuring you the right payload and parameter are used. 
+<table>
+<thead>
+  <tr>
+    <th>Input (AsyncAPI)</th>
+    <th>Using the code</th>
+  </tr>
+</thead>
+<tbody>
+  <tr>
+    <td>
+
+```yaml
+asyncapi: 3.0.0
+info:
+  title: Account Service
+  version: 1.0.0
+  description: This service is in charge of processing user signups
+channels:
+  userSignups:
+    address: user/signedup
+    messages:
+      userSignedup:
+        $ref: '#/components/messages/UserSignedUp'
+operations:
+  publishUserSignups:
+    action: send
+    channel:
+      $ref: '#/channels/userSignups'
+  consumeUserSignups:
+    action: receive
+    channel:
+      $ref: '#/channels/userSignups'
+components:
+  messages:
+    UserSignedUp:
+      payload:
+        type: object
+        properties:
+          displayName:
+            type: string
+            description: Name of the user
+          email:
+            type: string
+            format: email
+            description: Email of the user
+
+```
+</td>
+    <td>
+
+```ts
+import { Kafka } from 'kafkajs';
+// Location depends on the payload generator configurations
+import { UserSignedup } from './__gen__/payloads/UserSignedup';
+// Location depends on the channel generator configurations
+import { Protocols } from './__gen__/channels';
+const { kafka } = Protocols;
+const { consumeFromConsumeUserSignups, produceToPublishUserSignups } = kafka;
+
+/**
+ * Setup the regular client
+ */
+const kafkaClient = new Kafka({
+  clientId: 'test',
+  brokers: ['localhost:9093'],
+});
+
+const myPayload = new UserSignedup({displayName: 'test', email: 'test@test.dk'});
+const myParameters = new UserSignedUpParameters({userId: 'test'});
+
+// Consume the messages with the generated channel function
+const consumerCallback = async (
+    err,
+    msg: UserSignedUp | undefined, 
+    parameters: UserSignedUpParameters | undefined, 
+    kafkaMsg: EachMessagePayload | undefined
+  ) => {
+  // Do stuff once you consumer from the topic
+};
+const consumer = await consumeFromConsumeUserSignups(
+  consumerCallback,
+  myParameters, 
+  kafkaClient, 
+  {
+    fromBeginning: true, 
+    groupId: 'testId1'
+  }
+);
+
+// Produce the messages with the generated channel function
+const producer = await produceToPublishUserSignups(myPayload, myParameters, kafkaClient);
+```	
+</td>
+  </tr>
+</tbody>
+</table>

package.json

@@ -115,10 +115,12 @@
     "runtime:typescript": "npm run runtime:typescript:setup && npm run runtime:typescript:test",
     "runtime:typescript:setup": "npm run runtime:prepare && npm run runtime:services:start && cd test/runtime/typescript && npm ci && npm run generate",
     "runtime:typescript:test": "cd test/runtime/typescript && npm run test",
-    "runtime:services:start": "npm run runtime:nats:start",
-    "runtime:services:stop": "npm run runtime:nats:stop",
+    "runtime:services:start": "npm run runtime:nats:start && npm run runtime:kafka:start",
+    "runtime:services:stop": "npm run runtime:nats:stop && npm run runtime:kafka:stop",
     "runtime:nats:start": "cd test/runtime && docker compose -f ./docker-compose-nats.yml up -d",
     "runtime:nats:stop": "cd test/runtime && docker compose -f ./docker-compose-nats.yml down",
+    "runtime:kafka:start": "cd test/runtime && docker compose -f ./docker-compose-kafka.yml up -d",
+    "runtime:kafka:stop": "cd test/runtime && docker compose -f ./docker-compose-kafka.yml down",
     "test:blackbox": "concurrently --group -n typescript \"npm run test:blackbox:typescript\"",
     "test:blackbox:typescript": "cross-env CI=true jest ./test/blackbox/typescript.spec.ts"
   },

src/codegen/generators/helpers/payloads.ts

@@ -159,24 +159,21 @@ export function findReplyId(
   reply: OperationReplyInterface,
   channel: ChannelInterface
 ) {
-  return (
-    (reply.json() as any)?.id ??
-    `${findOperationId(operation, reply.channel() ?? channel)}_reply`
-  );
+  return `${findOperationId(operation, reply.channel() ?? channel)}_reply`;
 }
 export function findOperationId(
   operation: OperationInterface,
   channel: ChannelInterface
 ) {
-  let operationId = operation.id();
-  operationId = operation.hasOperationId()
-    ? operation.operationId()
-    : operationId;
   const userSpecificName = findExtensionObject(operation)
     ? findExtensionObject(operation)['channelName']
     : undefined;
   if (userSpecificName) {
     return userSpecificName;
   }
+  let operationId = operation.id();
+  operationId = operation.hasOperationId()
+    ? operation.operationId()
+    : operationId;
   return operationId ?? channel.id();
 }

src/codegen/generators/typescript/channels/asyncapi.ts

@@ -4,13 +4,15 @@ type Action = 'send' | 'receive' | 'subscribe' | 'publish';
 const sendingFunctionTypes = [
   ChannelFunctionTypes.NATS_JETSTREAM_PUBLISH,
   ChannelFunctionTypes.NATS_PUBLISH,
-  ChannelFunctionTypes.NATS_REQUEST
+  ChannelFunctionTypes.NATS_REQUEST,
+  ChannelFunctionTypes.KAFKA_PUBLISH
 ];
 const receivingFunctionTypes = [
   ChannelFunctionTypes.NATS_JETSTREAM_PULL_SUBSCRIBE,
   ChannelFunctionTypes.NATS_JETSTREAM_PUSH_SUBSCRIBE,
   ChannelFunctionTypes.NATS_REPLY,
-  ChannelFunctionTypes.NATS_SUBSCRIBE
+  ChannelFunctionTypes.NATS_SUBSCRIBE,
+  ChannelFunctionTypes.KAFKA_SUBSCRIBE
 ];
 
 // eslint-disable-next-line sonarjs/cognitive-complexity

src/codegen/generators/typescript/channels/index.ts

@@ -41,6 +41,7 @@ import {
   addPayloadsToDependencies,
   getMessageTypeAndModule
 } from './utils';
+import * as KafkaRenderer from './protocols/kafka';
 export {
   renderedFunctionType,
   TypeScriptChannelRenderType,
@@ -366,7 +367,111 @@ export async function generateTypeScriptChannels(
           dependencies.push(...(new Set(renderedDependencies) as any));
           break;
         }
+        case 'kafka': {
+          // AsyncAPI v2 explicitly say to use RFC 6570 URI template, Kafka does not support '/' subject separation, so lets replace it
+          let topic = simpleContext.topic;
+          if (topic.startsWith('/')) {
+            topic = topic.slice(1);
+          }
+          topic = topic.replace(/\//g, generator.kafkaTopicSeparator);
+          let kafkaContext: RenderRegularParameters = {
+            ...simpleContext,
+            topic,
+            messageType: ''
+          };
+          const renders = [];
+          const payload = payloads.channelModels[channel.id()];
+          if (payload === undefined) {
+            throw new Error(
+              `Could not find payload for ${channel.id()} for channel typescript generator`
+            );
+          }
+          const {messageModule, messageType} =
+            getMessageTypeAndModule(payload);
+            kafkaContext = {...kafkaContext, messageType, messageModule};
+          const operations = channel.operations().all();
+          if (operations.length > 0 && !ignoreOperation) {
+            for (const operation of operations) {
+              const payloadId = findOperationId(operation, channel);
+              const payload = payloads.operationModels[payloadId];
+              if (payload === undefined) {
+                throw new Error(
+                  `Could not find payload for ${payloadId} for channel typescript generator ${JSON.stringify(payloads.operationModels, null, 4)}`
+                );
+              }
+              const {messageModule, messageType} =
+                getMessageTypeAndModule(payload);
+              kafkaContext = {
+                ...kafkaContext,
+                messageType,
+                messageModule,
+                subName: findNameFromOperation(operation, channel)
+              };
+              const action = operation.action();
+              if (
+                shouldRenderFunctionType(
+                  functionTypeMapping,
+                  ChannelFunctionTypes.KAFKA_PUBLISH,
+                  action,
+                  generator.asyncapiReverseOperations
+                )
+              ) {
+                renders.push(KafkaRenderer.renderPublish(kafkaContext));
+              }
+              if (
+                shouldRenderFunctionType(
+                  functionTypeMapping,
+                  ChannelFunctionTypes.KAFKA_SUBSCRIBE,
+                  action,
+                  generator.asyncapiReverseOperations
+                )
+              ) {
+                renders.push(KafkaRenderer.renderSubscribe(kafkaContext));
+              }
+            }
+          } else {
+            if (
+              shouldRenderFunctionType(
+                functionTypeMapping,
+                ChannelFunctionTypes.KAFKA_PUBLISH,
+                'send',
+                generator.asyncapiReverseOperations
+              )
+            ) {
+              renders.push(KafkaRenderer.renderPublish(kafkaContext));
+            }
+            if (
+              shouldRenderFunctionType(
+                functionTypeMapping,
+                ChannelFunctionTypes.KAFKA_SUBSCRIBE,
+                'receive',
+                generator.asyncapiReverseOperations
+              )
+            ) {
+              renders.push(KafkaRenderer.renderSubscribe(kafkaContext));
+            }
+          }
+          protocolCodeFunctions[protocol].push(
+            ...renders.map((value) => value.code)
+          );
 
+          externalProtocolFunctionInformation[protocol].push(
+            ...renders.map((value) => {
+              return {
+                functionType: value.functionType,
+                functionName: value.functionName,
+                messageType: value.messageType,
+                replyType: value.replyType,
+                parameterType: parameter?.model?.type
+              };
+            })
+          );
+          const renderedDependencies = renders
+            .map((value) => value.dependencies)
+            .flat(Infinity);
+          dependencies.push(...(new Set(renderedDependencies) as any));
+          break;
+        }
         default: {
           break;
         }

src/codegen/generators/typescript/channels/protocols/kafka/index.ts

@@ -0,0 +1,2 @@
+export {renderPublish} from './publish';
+export {renderSubscribe} from './subscribe';