feat: add jetstream fetch wrappers (#479)

asyncapi · Nov 8, 2022 · cbda2ff · cbda2ff
1 parent d1df8d5
commit cbda2ff
Show file tree

Hide file tree

Showing 12 changed files with 311 additions and 4 deletions.
diff --git a/components/channel/jetStreamFetch.js b/components/channel/jetStreamFetch.js
@@ -0,0 +1,62 @@
+import { camelCase, getMessageType, realizeParametersForChannelWrapper, renderJSDocParameters, messageHasNullPayload, realizeChannelName} from '../../utils/index';
+// eslint-disable-next-line no-unused-vars
+import { Message, ChannelParameter } from '@asyncapi/parser';
+import { unwrap } from './ChannelParameterUnwrap';
+
+/**
+ * Component which returns a subscribe to function for the client
+ * 
+ * @param {string} defaultContentType 
+ * @param {string} channelName to publish to
+ * @param {Message} message which is being received
+ * @param {Object.<string, ChannelParameter>} channelParameters parameters to the channel
+ */
+export function JetstreamFetch(channelName, message, channelParameters) {
+  const messageType = getMessageType(message);
+  let parameters = [];
+  parameters = Object.entries(channelParameters).map(([parameterName]) => {
+    return `${camelCase(parameterName)}Param`;
+  });
+  const hasNullPayload = messageHasNullPayload(message.payload());
+
+  //Determine the callback process when receiving messages.
+  //If the message payload is null no hooks are called to process the received data.
+  let whenReceivingMessage = `onDataCallback(undefined, null ${parameters.length > 0 && `, ${parameters.join(',')}`});`;
+  if (!hasNullPayload) {
+    whenReceivingMessage =  `
+    let receivedData: any = codec.decode(msg.data);
+    onDataCallback(undefined, ${messageType}.unmarshal(receivedData) ${parameters.length > 0 && `, ${parameters.join(',')}`}, msg);
+    `;
+  }
+  return  `
+    /**
+     * Internal functionality to setup jetstrema fetch on the \`${channelName}\` channel 
+     * 
+     * @param onDataCallback to call when messages are received
+     * @param js client to fetch with
+     * @param codec used to convert messages
+     ${renderJSDocParameters(channelParameters)}
+      */
+      export function jetsStreamFetch(
+        onDataCallback: (
+          err ? : NatsTypescriptTemplateError,
+          msg ? : ${messageType}
+          ${realizeParametersForChannelWrapper(channelParameters, false)},
+          jetstreamMsg?: Nats.JsMsg) => void,
+        js: Nats.JetStreamClient,
+        codec: Nats.Codec < any > 
+        ${realizeParametersForChannelWrapper(channelParameters)},
+        durable: string, options?: Partial<Nats.PullOptions>
+      ) {
+        const stream = ${realizeChannelName(channelParameters, channelName)};
+        (async () => {
+          let msgs = await js.fetch(stream, durable, options);
+          for await (const msg of msgs) {
+            ${unwrap(channelName, channelParameters)}
+
+            ${whenReceivingMessage}
+          }
+        })();
+      }
+  `;
+}
diff --git a/components/index/jetStreamFetch.js b/components/index/jetStreamFetch.js
@@ -0,0 +1,39 @@
+import { camelCase, getMessageType, realizeParametersForChannelWrapper, renderJSDocParameters, realizeParametersForChannelWithoutType, pascalCase} from '../../utils/index';
+
+export function JetstreamFetch(channelName, message, messageDescription, channelParameters) {
+  return `
+  /**
+    * JetStream fetch function.
+    * 
+    * Pull message from \`${channelName}\`
+    * 
+    * ${messageDescription}
+    * 
+    * @param onDataCallback to call when messages are received
+    ${renderJSDocParameters(channelParameters)}
+    * @param options to pull message with, bindings from the AsyncAPI document overwrite these if specified
+    */
+    public jetStreamFetch${pascalCase(channelName)}(
+      onDataCallback: (
+        err ? : NatsTypescriptTemplateError,
+        msg?: ${getMessageType(message)}
+        ${realizeParametersForChannelWrapper(channelParameters, false)},
+        jetstreamMsg?: Nats.JsMsg) => void
+      ${realizeParametersForChannelWrapper(channelParameters)},
+      durable: string, options?: Partial<Nats.PullOptions>
+    ): void {
+      if (!this.isClosed() && this.nc !== undefined && this.codec !== undefined && this.js !== undefined) {
+          ${camelCase(channelName)}Channel.jetsStreamFetch(
+          onDataCallback,
+          this.js,
+          this.codec
+          ${Object.keys(channelParameters).length ? ` ,${realizeParametersForChannelWithoutType(channelParameters)}` : ''},
+          durable,
+          options
+        );
+      } else {
+        throw NatsTypescriptTemplateError.errorForCode(ErrorCode.NOT_CONNECTED);
+      }
+    }
+    `;
+}
diff --git a/examples/simple-publish/asyncapi-nats-client/API.md b/examples/simple-publish/asyncapi-nats-client/API.md
@@ -197,6 +197,7 @@ The test/mirror client which is the reverse to the normal NatsAsyncApiClient.
     * [.jetStreamPullStreetlightStreetlightIdCommandTurnon(onDataCallback, streetlight_id, options)](#NatsAsyncApiTestClient+jetStreamPullStreetlightStreetlightIdCommandTurnon)
     * [.jetStreamPushSubscribeToStreetlightStreetlightIdCommandTurnon(onDataCallback, streetlight_id, flush, options)](#NatsAsyncApiTestClient+jetStreamPushSubscribeToStreetlightStreetlightIdCommandTurnon)
     * [.jetStreamPullSubscribeToStreetlightStreetlightIdCommandTurnon(onDataCallback, streetlight_id, flush, options)](#NatsAsyncApiTestClient+jetStreamPullSubscribeToStreetlightStreetlightIdCommandTurnon)
+    * [.jetStreamFetchStreetlightStreetlightIdCommandTurnon(onDataCallback, streetlight_id, options)](#NatsAsyncApiTestClient+jetStreamFetchStreetlightStreetlightIdCommandTurnon)
 
 <a name="NatsAsyncApiTestClient+connect"></a>
 
@@ -330,3 +331,20 @@ Channel for the turn on command which should turn on the streetlight
 | flush | ensure client is force flushed after subscribing |
 | options | to subscribe with, bindings from the AsyncAPI document overwrite these if specified |
 
+<a name="NatsAsyncApiTestClient+jetStreamFetchStreetlightStreetlightIdCommandTurnon"></a>
+
+### natsAsyncApiTestClient.jetStreamFetchStreetlightStreetlightIdCommandTurnon(onDataCallback, streetlight_id, options)
+JetStream fetch function.
+
+Pull message from `streetlight/{streetlight_id}/command/turnon`
+
+Channel for the turn on command which should turn on the streetlight
+
+**Kind**: instance method of [<code>NatsAsyncApiTestClient</code>](#NatsAsyncApiTestClient)  
+
+| Param | Description |
+| --- | --- |
+| onDataCallback | to call when messages are received |
+| streetlight_id | parameter to use in topic |
+| options | to pull message with, bindings from the AsyncAPI document overwrite these if specified |
+
diff --git a/examples/simple-publish/asyncapi-nats-client/src/testclient/index.ts b/examples/simple-publish/asyncapi-nats-client/src/testclient/index.ts
@@ -245,4 +245,34 @@ export class NatsAsyncApiTestClient {
       }
     });
   }
+  /**
+   * JetStream fetch function.
+   * 
+   * Pull message from `streetlight/{streetlight_id}/command/turnon`
+   * 
+   * Channel for the turn on command which should turn on the streetlight
+   * 
+   * @param onDataCallback to call when messages are received
+   * @param streetlight_id parameter to use in topic
+   * @param options to pull message with, bindings from the AsyncAPI document overwrite these if specified
+   */
+  public jetStreamFetchStreetlightStreetlightIdCommandTurnon(
+    onDataCallback: (
+      err ? : NatsTypescriptTemplateError,
+      msg ? : TurnOn, streetlight_id ? : string,
+      jetstreamMsg ? : Nats.JsMsg) => void, streetlight_id: string,
+    durable: string, options ? : Partial < Nats.PullOptions >
+  ): void {
+    if (!this.isClosed() && this.nc !== undefined && this.codec !== undefined && this.js !== undefined) {
+      streetlightStreetlightIdCommandTurnonChannel.jetsStreamFetch(
+        onDataCallback,
+        this.js,
+        this.codec, streetlight_id,
+        durable,
+        options
+      );
+    } else {
+      throw NatsTypescriptTemplateError.errorForCode(ErrorCode.NOT_CONNECTED);
+    }
+  }
 }
diff --git a/...asyncapi-nats-client/src/testclient/testchannels/StreetlightStreetlightIdCommandTurnon.ts b/...asyncapi-nats-client/src/testclient/testchannels/StreetlightStreetlightIdCommandTurnon.ts
@@ -172,4 +172,40 @@ export function jetStreamPullSubscribe(
       reject(NatsTypescriptTemplateError.errorForCode(ErrorCode.INTERNAL_NATS_TS_ERROR, e));
     }
   })
+}
+/**
+ * Internal functionality to setup jetstrema fetch on the `streetlight/{streetlight_id}/command/turnon` channel 
+ * 
+ * @param onDataCallback to call when messages are received
+ * @param js client to fetch with
+ * @param codec used to convert messages
+ * @param streetlight_id parameter to use in topic
+ */
+export function jetsStreamFetch(
+  onDataCallback: (
+    err ? : NatsTypescriptTemplateError,
+    msg ? : TurnOn, streetlight_id ? : string,
+    jetstreamMsg ? : Nats.JsMsg) => void,
+  js: Nats.JetStreamClient,
+  codec: Nats.Codec < any > , streetlight_id: string,
+  durable: string, options ? : Partial < Nats.PullOptions >
+) {
+  const stream = `streetlight.${streetlight_id}.command.turnon`;
+  (async () => {
+    let msgs = await js.fetch(stream, durable, options);
+    for await (const msg of msgs) {
+      const unmodifiedChannel = `streetlight.{streetlight_id}.command.turnon`;
+      let channel = msg.subject;
+      const streetlightIdSplit = unmodifiedChannel.split("{streetlight_id}");
+      const splits = [
+        streetlightIdSplit[0],
+        streetlightIdSplit[1]
+      ];
+      channel = channel.substring(splits[0].length);
+      const streetlightIdEnd = channel.indexOf(splits[1]);
+      const streetlightIdParam = "" + channel.substring(0, streetlightIdEnd);
+      let receivedData: any = codec.decode(msg.data);
+      onDataCallback(undefined, TurnOn.unmarshal(receivedData), streetlightIdParam, msg);
+    }
+  })();
 }
diff --git a/examples/simple-subscribe/asyncapi-nats-client/API.md b/examples/simple-subscribe/asyncapi-nats-client/API.md
@@ -30,6 +30,7 @@ Module which wraps functionality for the `streetlight/{streetlight_id}/command/t
     * [~jetStreamPull(onDataCallback, js, codec, streetlight_id)](#module_streetlightStreetlightIdCommandTurnon..jetStreamPull)
     * [~jetStreamPushSubscribe(onDataCallback, nc, codec, streetlight_id, options)](#module_streetlightStreetlightIdCommandTurnon..jetStreamPushSubscribe)
     * [~jetStreamPullSubscribe(onDataCallback, nc, codec, streetlight_id)](#module_streetlightStreetlightIdCommandTurnon..jetStreamPullSubscribe)
+    * [~jetsStreamFetch(onDataCallback, js, codec, streetlight_id)](#module_streetlightStreetlightIdCommandTurnon..jetsStreamFetch)
 
 <a name="module_streetlightStreetlightIdCommandTurnon..subscribe"></a>
 
@@ -89,6 +90,20 @@ Internal functionality to setup jetstream pull subscription on the `streetlight/
 | codec | used to convert messages |
 | streetlight_id | parameter to use in topic |
 
+<a name="module_streetlightStreetlightIdCommandTurnon..jetsStreamFetch"></a>
+
+### streetlightStreetlightIdCommandTurnon~jetsStreamFetch(onDataCallback, js, codec, streetlight_id)
+Internal functionality to setup jetstrema fetch on the `streetlight/{streetlight_id}/command/turnon` channel
+
+**Kind**: inner method of [<code>streetlightStreetlightIdCommandTurnon</code>](#module_streetlightStreetlightIdCommandTurnon)  
+
+| Param | Description |
+| --- | --- |
+| onDataCallback | to call when messages are received |
+| js | client to fetch with |
+| codec | used to convert messages |
+| streetlight_id | parameter to use in topic |
+
 <a name="NatsAsyncApiClient"></a>
 
 ## NatsAsyncApiClient
@@ -110,6 +125,7 @@ The generated client based on your AsyncAPI document.
     * [.jetStreamPullStreetlightStreetlightIdCommandTurnon(onDataCallback, streetlight_id, options)](#NatsAsyncApiClient+jetStreamPullStreetlightStreetlightIdCommandTurnon)
     * [.jetStreamPushSubscribeToStreetlightStreetlightIdCommandTurnon(onDataCallback, streetlight_id, flush, options)](#NatsAsyncApiClient+jetStreamPushSubscribeToStreetlightStreetlightIdCommandTurnon)
     * [.jetStreamPullSubscribeToStreetlightStreetlightIdCommandTurnon(onDataCallback, streetlight_id, flush, options)](#NatsAsyncApiClient+jetStreamPullSubscribeToStreetlightStreetlightIdCommandTurnon)
+    * [.jetStreamFetchStreetlightStreetlightIdCommandTurnon(onDataCallback, streetlight_id, options)](#NatsAsyncApiClient+jetStreamFetchStreetlightStreetlightIdCommandTurnon)
 
 <a name="NatsAsyncApiClient+connect"></a>
 
@@ -243,6 +259,23 @@ Channel for the turn on command which should turn on the streetlight
 | flush | ensure client is force flushed after subscribing |
 | options | to subscribe with, bindings from the AsyncAPI document overwrite these if specified |
 
+<a name="NatsAsyncApiClient+jetStreamFetchStreetlightStreetlightIdCommandTurnon"></a>
+
+### natsAsyncApiClient.jetStreamFetchStreetlightStreetlightIdCommandTurnon(onDataCallback, streetlight_id, options)
+JetStream fetch function.
+
+Pull message from `streetlight/{streetlight_id}/command/turnon`
+
+Channel for the turn on command which should turn on the streetlight
+
+**Kind**: instance method of [<code>NatsAsyncApiClient</code>](#NatsAsyncApiClient)  
+
+| Param | Description |
+| --- | --- |
+| onDataCallback | to call when messages are received |
+| streetlight_id | parameter to use in topic |
+| options | to pull message with, bindings from the AsyncAPI document overwrite these if specified |
+
 <a name="NatsAsyncApiTestClient"></a>
 
 ## NatsAsyncApiTestClient

diff --git a/...mple-subscribe/asyncapi-nats-client/src/channels/StreetlightStreetlightIdCommandTurnon.ts b/...mple-subscribe/asyncapi-nats-client/src/channels/StreetlightStreetlightIdCommandTurnon.ts
@@ -172,4 +172,40 @@ export function jetStreamPullSubscribe(
       reject(NatsTypescriptTemplateError.errorForCode(ErrorCode.INTERNAL_NATS_TS_ERROR, e));
     }
   })
+}
+/**
+ * Internal functionality to setup jetstrema fetch on the `streetlight/{streetlight_id}/command/turnon` channel 
+ * 
+ * @param onDataCallback to call when messages are received
+ * @param js client to fetch with
+ * @param codec used to convert messages
+ * @param streetlight_id parameter to use in topic
+ */
+export function jetsStreamFetch(
+  onDataCallback: (
+    err ? : NatsTypescriptTemplateError,
+    msg ? : TurnOn, streetlight_id ? : string,
+    jetstreamMsg ? : Nats.JsMsg) => void,
+  js: Nats.JetStreamClient,
+  codec: Nats.Codec < any > , streetlight_id: string,
+  durable: string, options ? : Partial < Nats.PullOptions >
+) {
+  const stream = `streetlight.${streetlight_id}.command.turnon`;
+  (async () => {
+    let msgs = await js.fetch(stream, durable, options);
+    for await (const msg of msgs) {
+      const unmodifiedChannel = `streetlight.{streetlight_id}.command.turnon`;
+      let channel = msg.subject;
+      const streetlightIdSplit = unmodifiedChannel.split("{streetlight_id}");
+      const splits = [
+        streetlightIdSplit[0],
+        streetlightIdSplit[1]
+      ];
+      channel = channel.substring(splits[0].length);
+      const streetlightIdEnd = channel.indexOf(splits[1]);
+      const streetlightIdParam = "" + channel.substring(0, streetlightIdEnd);
+      let receivedData: any = codec.decode(msg.data);
+      onDataCallback(undefined, TurnOn.unmarshal(receivedData), streetlightIdParam, msg);
+    }
+  })();
 }
diff --git a/examples/simple-subscribe/asyncapi-nats-client/src/index.ts b/examples/simple-subscribe/asyncapi-nats-client/src/index.ts
@@ -253,4 +253,34 @@ export class NatsAsyncApiClient {
       }
     });
   }
+  /**
+   * JetStream fetch function.
+   * 
+   * Pull message from `streetlight/{streetlight_id}/command/turnon`
+   * 
+   * Channel for the turn on command which should turn on the streetlight
+   * 
+   * @param onDataCallback to call when messages are received
+   * @param streetlight_id parameter to use in topic
+   * @param options to pull message with, bindings from the AsyncAPI document overwrite these if specified
+   */
+  public jetStreamFetchStreetlightStreetlightIdCommandTurnon(
+    onDataCallback: (
+      err ? : NatsTypescriptTemplateError,
+      msg ? : TurnOn, streetlight_id ? : string,
+      jetstreamMsg ? : Nats.JsMsg) => void, streetlight_id: string,
+    durable: string, options ? : Partial < Nats.PullOptions >
+  ): void {
+    if (!this.isClosed() && this.nc !== undefined && this.codec !== undefined && this.js !== undefined) {
+      streetlightStreetlightIdCommandTurnonChannel.jetsStreamFetch(
+        onDataCallback,
+        this.js,
+        this.codec, streetlight_id,
+        durable,
+        options
+      );
+    } else {
+      throw NatsTypescriptTemplateError.errorForCode(ErrorCode.NOT_CONNECTED);
+    }
+  }
 }
diff --git a/template/src/channels/$$channel$$.ts.js b/template/src/channels/$$channel$$.ts.js
@@ -7,6 +7,7 @@ import { General } from '../../../components/channel/general';
 import { pascalCase, isRequestReply, isReplier, isRequester, isPubsub, camelCase} from '../../../utils/index';
 // eslint-disable-next-line no-unused-vars
 import { AsyncAPIDocument, Channel } from '@asyncapi/parser';
+import { JetstreamFetch } from '../../../components/channel/jetStreamFetch';
 import { JetstreamPushSubscription } from '../../../components/channel/jetstreamPushSubscription';
 import { JetstreamPull } from '../../../components/channel/jetstreamPull';
 import { JetstreamPullSubscription } from '../../../components/channel/jetStreamPullSubscription';
@@ -90,7 +91,12 @@ function getChannelCode(channel, channelName, params) {
         channelName, 
         publishMessage,
         channel.parameters());
-      channelcode = `${normalSubscribeCode}\n${jetstreamPullCode}\n${jetstreamPushSubscriptionCode}\n${jetstreamPullSubscriptionCode}`;
+      const jetstreamFetchCode = JetstreamFetch(
+        channelName, 
+        publishMessage, 
+        channel.parameters(),
+        publishOperation);
+      channelcode = `${normalSubscribeCode}\n${jetstreamPullCode}\n${jetstreamPushSubscriptionCode}\n${jetstreamPullSubscriptionCode}\n${jetstreamFetchCode}`;
     }
   }
   return channelcode;

diff --git a/template/src/index.ts.js b/template/src/index.ts.js
@@ -10,6 +10,7 @@ import { AsyncAPIDocument } from '@asyncapi/parser';
 import { JetstreamPushSubscription } from '../../components/index/jetstreamPushSubscription';
 import { JetstreamPull } from '../../components/index/jetstreamPull';
 import { JetstreamPullSubscribe } from '../../components/index/jetStreamPullSubscription';
+import { JetstreamFetch } from '../../components/index/jetStreamFetch';
 import { JetstreamPublish } from '../../components/index/jetstreamPublish';
 
 /**
@@ -83,6 +84,11 @@ function getChannelWrappers(asyncapi, params) {
           publishMessage, 
           channelDescription, 
           channelParameters);
+        const jetstreamFetchCode = JetstreamFetch(
+          channelName, 
+          publishMessage, 
+          channelDescription, 
+          channelParameters);
         const jetstreamPullSubscribe = JetstreamPullSubscribe(
           channelName, 
           publishMessage, 
@@ -98,7 +104,7 @@ function getChannelWrappers(asyncapi, params) {
           publishMessage, 
           channelDescription, 
           channelParameters);
-        return `${normalSubscribeCode}\n${jetstreamPullCode}\n${jetstreamPushSubscriptionCode}\n${jetstreamPullSubscribe}`;
+        return `${normalSubscribeCode}\n${jetstreamPullCode}\n${jetstreamPushSubscriptionCode}\n${jetstreamPullSubscribe}\n${jetstreamFetchCode}`;
       }
     }
   });