feat: add event source client

docs/README.md

@@ -46,15 +46,7 @@ Each input has its own limitations, corner cases, and features; thus, each has s
 ### Protocols
 Each protocol has its own limitations, corner cases, and features; thus, each has separate documentation.
 - [NATS](./protocols/nats.md)
-
-
-
-
-
-
-
-
-
-
-
-
+- [AMQP](./protocols/amqp.md)
+- [Kafka](./protocols/kafka.md)
+- [MQTT](./protocols/mqtt.md)
+- [EventSource](./protocols/eventsource.md)

docs/generators/channels.md

@@ -26,7 +26,7 @@ This is supported through the following inputs: [`asyncapi`](../inputs/asyncapi.
 
 It supports the following languages; [`typescript`](#typescript)
 
-It supports the following protocols; [`nats`](../protocols/nats.md), [`kafka`](../protocols/kafka.md), [`mqtt`](../protocols/mqtt.md), [`amqp`](../protocols/amqp.md)
+It supports the following protocols; [`nats`](../protocols/nats.md), [`kafka`](../protocols/kafka.md), [`mqtt`](../protocols/mqtt.md), [`amqp`](../protocols/amqp.md), [`event_source_client`](../protocols/eventsource.md#client)
 
 ## Options
 These are the available options for the `channels` generator; 
@@ -44,12 +44,13 @@ Depending on which protocol, these are the dependencies:
 - `Kafka`: https://github.com/tulios/kafkajs v2
 - `MQTT`: https://github.com/mqttjs/MQTT.js v5
 - `AMQP`: https://github.com/amqp-node/amqplib v0
+- `EventSource client`: https://github.com/Azure/fetch-event-source v2
 
 For TypeScript what is generated is a single file that include functions to help easier interact with AsyncAPI channels. For example;
 
 ```ts
 import { Protocols } from 'src/__gen__/index';
-const { nats, kafka, mqtt, amqp ... } = Protocols;
+const { nats, kafka, mqtt, amqp, event_source_client, ... } = Protocols;
 const { jetStreamPublishTo..., jetStreamPullSubscribeTo..., jetStreamPushSubscriptionFrom..., publishTo..., subscribeTo... } = nats;
 ```
 

docs/protocols/eventsource.md

@@ -0,0 +1,88 @@
+---
+sidebar_position: 99
+---
+
+# EventSource
+`Event Source` is currently available through the generators ([channels](#channels)):
+
+| **Languages** | [client](#client) | server |
+|---|---|---|
+| TypeScript | ✔️ |  |
+
+All of this is available through [AsyncAPI](../inputs/asyncapi.md).
+
+## Client
+
+The client generated code is to listen for events from the server and act accordingly. It currently supports 
+
+## 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:
+  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
+// 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 { event_source_client } = Protocols;
+const { listenForUserSignedup } = event_source_client;
+const listenCallback = async (
+  messageEvent: UserSignedUp | null, 
+  parameters: UserSignedUpParameters | null,
+  error?: string
+) => {
+  // Do stuff once you receive the event
+};
+listenForUserSignedup(listenCallback, {baseUrl: 'http://localhost:3000'})
+```	
+</td>
+  </tr>
+</tbody>
+</table>

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

@@ -8,14 +8,15 @@ const sendingFunctionTypes = [
   ChannelFunctionTypes.MQTT_PUBLISH,
   ChannelFunctionTypes.KAFKA_PUBLISH,
   ChannelFunctionTypes.AMQP_EXCHANGE_PUBLISH,
-  ChannelFunctionTypes.AMQP_QUEUE_PUBLISH
+  ChannelFunctionTypes.AMQP_QUEUE_PUBLISH,
 ];
 const receivingFunctionTypes = [
   ChannelFunctionTypes.NATS_JETSTREAM_PULL_SUBSCRIBE,
   ChannelFunctionTypes.NATS_JETSTREAM_PUSH_SUBSCRIBE,
   ChannelFunctionTypes.NATS_REPLY,
   ChannelFunctionTypes.NATS_SUBSCRIBE,
-  ChannelFunctionTypes.KAFKA_SUBSCRIBE
+  ChannelFunctionTypes.KAFKA_SUBSCRIBE,
+  ChannelFunctionTypes.EVENT_SOURCE_FETCH,
 ];
 
 // eslint-disable-next-line sonarjs/cognitive-complexity

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

@@ -27,6 +27,7 @@ import {renderCoreReply} from './protocols/nats/coreReply';
 import * as MqttRenderer from './protocols/mqtt';
 import * as AmqpRenderer from './protocols/amqp';
 import * as KafkaRenderer from './protocols/kafka';
+import * as EventSourceRenderer from './protocols/eventsource';
 import {shouldRenderFunctionType} from './asyncapi';
 import {
   renderedFunctionType,
@@ -564,7 +565,7 @@ export async function generateTypeScriptChannels(
           const renders = [];
           const operations = channel.operations().all();
           const exchangeName =
-            channel.bindings().get('amqp')?.value().exchange.name ?? undefined;
+            channel.bindings().get('amqp')?.value()?.exchange?.name ?? undefined;
           if (operations.length > 0 && !ignoreOperation) {
             for (const operation of operations) {
               const payloadId = findOperationId(operation, channel);
@@ -667,6 +668,92 @@ export async function generateTypeScriptChannels(
           dependencies.push(...(new Set(renderedDependencies) as any));
           break;
         }
+
+        case 'event_source_client': {
+          const topic = simpleContext.topic;
+          let eventSourceContext: RenderRegularParameters = {
+            ...simpleContext,
+            topic,
+            messageType: ''
+          };
+          const renders = [];
+          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);
+              eventSourceContext = {
+                ...eventSourceContext,
+                messageType,
+                messageModule,
+                subName: findNameFromOperation(operation, channel)
+              };
+              const action = operation.action();
+              if (
+                shouldRenderFunctionType(
+                  functionTypeMapping,
+                  ChannelFunctionTypes.EVENT_SOURCE_FETCH,
+                  action,
+                  generator.asyncapiReverseOperations
+                )
+              ) {
+                renders.push(
+                  EventSourceRenderer.renderListenForEvent(eventSourceContext)
+                );
+              }
+            }
+          } else {
+            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);
+            eventSourceContext = {...eventSourceContext, messageType, messageModule};
+
+            if (
+              shouldRenderFunctionType(
+                functionTypeMapping,
+                ChannelFunctionTypes.EVENT_SOURCE_FETCH,
+                'receive',
+                generator.asyncapiReverseOperations
+              )
+            ) {
+              renders.push(
+                EventSourceRenderer.renderListenForEvent(eventSourceContext)
+              );
+            }
+          }
+          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/eventsource/index.ts

@@ -0,0 +1 @@
+export {renderListenForEvent} from './listenForEvent';

src/codegen/generators/typescript/channels/protocols/eventsource/listenForEvent.ts

@@ -0,0 +1,91 @@
+/* eslint-disable sonarjs/no-nested-template-literals */
+import {ChannelFunctionTypes} from '../..';
+import {SingleFunctionRenderType} from '../../../../../types';
+import {pascalCase} from '../../../utils';
+import {RenderRegularParameters} from '../../types';
+
+export function renderListenForEvent({
+  topic,
+  messageType,
+  messageModule,
+  channelParameters,
+  subName = pascalCase(topic),
+  functionName = `listenFor${subName}`
+}: RenderRegularParameters): SingleFunctionRenderType {
+  const addressToUse = channelParameters
+    ? `parameters.getChannelWithParameters('${topic}')`
+    : `'${topic}'`;
+  const messageUnmarshalling = `${messageModule ?? messageType}.unmarshal(ev.data)`;
+  messageType = messageModule ? `${messageModule}.${messageType}` : messageType;
+
+  const functionParameters = [
+    {
+      parameter: `callback: (messageEvent: ${messageType} | null, error?: string) => void`,
+      jsDoc: ' * @param callback to call when receiving events'
+    },
+    ...(channelParameters
+      ? [
+          {
+            parameter: `parameters: ${channelParameters.type}`,
+            jsDoc: ' * @param parameters for listening'
+          }
+        ]
+      : []),
+    {
+      parameter: 'options: {authorization?: string, onClose?: () => void, baseUrl: string}',
+      jsDoc: ' * @param options additionally used to handle the event source'
+    }
+
+  ];
+
+  const code = `/**
+ * Event source fetch for \`${topic}\`
+ * 
+ ${functionParameters.map((param) => param.jsDoc).join('\n')}
+ */
+${functionName}: (
+  ${functionParameters.map((param) => param.parameter).join(', \n  ')}
+) => {
+	let eventsUrl: string = ${addressToUse};
+	const url = \`\${options.baseUrl}/\${eventsUrl}\`
+  const headers: Record<string, string> = {
+    Accept: 'text/event-stream',
+  }
+  if(options.authorization) {
+    headers['authorization'] = \`Bearer \${options?.authorization}\`;
+  }
+	fetchEventSource(\`\${url}\`, {
+		method: 'GET',
+		headers,
+		onmessage: (ev: EventSourceMessage) => {
+      const callbackData = ${messageUnmarshalling};
+			callback(callbackData, undefined);
+		},
+		onerror: () => {
+			options.onClose?.();
+		},
+		onclose: () => {
+			options.onClose?.();
+		},
+		async onopen(response: { ok: any; headers: { get: (arg0: string) => any }; status: number }) {
+			if (response.ok && response.headers.get('content-type') === EventStreamContentType) {
+				return // everything's good
+			} else if (response.status >= 400 && response.status < 500 && response.status !== 429) {
+				// client-side errors are usually non-retriable:
+				callback(null, 'Client side error, could not open event connection')
+			} else {
+				callback(null, 'Unknown error, could not open event connection');
+			}
+		},
+	})
+}
+`;
+
+  return {
+    messageType,
+    code,
+    functionName,
+    dependencies: [`import { fetchEventSource, EventStreamContentType, EventSourceMessage } from '@microsoft/fetch-event-source'; `],
+    functionType: ChannelFunctionTypes.KAFKA_PUBLISH
+  };
+}

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

@@ -17,7 +17,8 @@ export enum ChannelFunctionTypes {
   KAFKA_PUBLISH = 'kafka_publish',
   KAFKA_SUBSCRIBE = 'kafka_subscribe',
   AMQP_QUEUE_PUBLISH = 'amqp_queue_publish',
-  AMQP_EXCHANGE_PUBLISH = 'amqp_exchange_publish'
+  AMQP_EXCHANGE_PUBLISH = 'amqp_exchange_publish',
+  EVENT_SOURCE_FETCH = 'event_source_fetch'
 }
 
 export const zodTypescriptChannelsGenerator = z.object({
@@ -29,8 +30,8 @@ export const zodTypescriptChannelsGenerator = z.object({
   preset: z.literal('channels').default('channels'),
   outputPath: z.string().default('src/__gen__/channels'),
   protocols: z
-    .array(z.enum(['nats', 'kafka', 'mqtt', 'amqp']))
-    .default(['nats', 'kafka', 'mqtt', 'amqp']),
+    .array(z.enum(['nats', 'kafka', 'mqtt', 'amqp', 'event_source_client']))
+    .default(['nats', 'kafka', 'mqtt', 'amqp', 'event_source_client']),
   parameterGeneratorId: z
     .string()
     .optional()
@@ -130,4 +131,4 @@ export interface RenderRequestReplyParameters {
   functionName?: string;
 }
 
-export type SupportedProtocols = 'nats' | 'kafka' | 'mqtt' | 'amqp';
+export type SupportedProtocols = 'nats' | 'kafka' | 'mqtt' | 'amqp' | 'event_source_client';

test/blackbox/projects/typescript/package.json

@@ -5,7 +5,8 @@
     "nats": "2.28.2",
     "kafkajs": "^2.2.4",
     "mqtt": "^5.10.3",
-    "amqplib": "^0.10.5"
+    "amqplib": "^0.10.5",
+    "@microsoft/fetch-event-source": "^2.0.1"
   },
   "devDependencies": {
     "@types/amqplib": "^0.10.6"