This repository has been archived by the owner on Oct 2, 2024. It is now read-only.
-
-
Notifications
You must be signed in to change notification settings - Fork 13
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: add jetstream push subscribe wrappers (#481)
- Loading branch information
1 parent
8cdb68c
commit b7c493f
Showing
13 changed files
with
409 additions
and
49 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| import { realizeChannelName, camelCase, getMessageType, messageHasNullPayload, realizeParametersForChannelWrapper, renderJSDocParameters} from '../../utils/index'; | ||
| import { unwrap } from './ChannelParameterUnwrap'; | ||
| // eslint-disable-next-line no-unused-vars | ||
| import { Message, ChannelParameter } from '@asyncapi/parser'; | ||
|
|
||
| /** | ||
| * Component which returns a function which subscribes to the given channel | ||
| * | ||
| * @param {string} defaultContentType | ||
| * @param {string} channelName to subscribe to | ||
| * @param {Message} message which is being received | ||
| * @param {Object.<string, ChannelParameter>} channelParameters parameters to the channel | ||
| */ | ||
| export function JetstreamPushSubscription(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(',')}`}); | ||
| `; | ||
| } | ||
|
|
||
| return ` | ||
| /** | ||
| * Internal functionality to setup jetstream push subscription on the \`${channelName}\` channel | ||
| * | ||
| * @param onDataCallback to call when messages are received | ||
| * @param nc to subscribe with | ||
| * @param codec used to convert messages | ||
| ${renderJSDocParameters(channelParameters)} | ||
| * @param options to subscribe with, bindings from the AsyncAPI document overwrite these if specified | ||
| */ | ||
| export function jetStreamPushSubscribe( | ||
| onDataCallback: ( | ||
| err ? : NatsTypescriptTemplateError, | ||
| msg?: ${messageType} | ||
| ${realizeParametersForChannelWrapper(channelParameters, false)}, | ||
| jetstreamMsg?: Nats.JsMsg) => void, | ||
| js: Nats.JetStreamClient, | ||
| codec: Nats.Codec < any > | ||
| ${realizeParametersForChannelWrapper(channelParameters)}, | ||
| options: Nats.ConsumerOptsBuilder | Partial<Nats.ConsumerOpts> | ||
| ): Promise < Nats.JetStreamSubscription > { | ||
| return new Promise(async (resolve, reject) => { | ||
| try { | ||
| let subscription = js.subscribe(${realizeChannelName(channelParameters, channelName)}, options); | ||
| (async () => { | ||
| for await (const msg of await subscription) { | ||
| ${unwrap(channelName, channelParameters)} | ||
| ${whenReceivingMessage} | ||
| } | ||
| console.log("subscription closed"); | ||
| })(); | ||
| resolve(subscription); | ||
| } catch (e: any) { | ||
| reject(NatsTypescriptTemplateError.errorForCode(ErrorCode.INTERNAL_NATS_TS_ERROR, e)); | ||
| } | ||
| }) | ||
| } | ||
| `; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| import { pascalCase, camelCase, getMessageType, realizeParametersForChannelWrapper, realizeParametersForChannelWithoutType, renderJSDocParameters} from '../../utils/index'; | ||
| // eslint-disable-next-line no-unused-vars | ||
| import { Message, ChannelParameter } from '@asyncapi/parser'; | ||
|
|
||
| /** | ||
| * 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 {string} messageDescription | ||
| * @param {Object.<string, ChannelParameter>} channelParameters parameters to the channel | ||
| */ | ||
| export function JetstreamPushSubscription(channelName, message, messageDescription, channelParameters) { | ||
| return ` | ||
| /** | ||
| * Push subscription to the \`${channelName}\` | ||
| * | ||
| * ${messageDescription} | ||
| * | ||
| * @param onDataCallback to call when messages are received | ||
| ${renderJSDocParameters(channelParameters)} | ||
| * @param flush ensure client is force flushed after subscribing | ||
| * @param options to subscribe with, bindings from the AsyncAPI document overwrite these if specified | ||
| */ | ||
| public jetStreamPushSubscribeTo${pascalCase(channelName)}( | ||
| onDataCallback: ( | ||
| err?: NatsTypescriptTemplateError, | ||
| msg?: ${getMessageType(message)} | ||
| ${realizeParametersForChannelWrapper(channelParameters, false)}, | ||
| jetstreamMsg?: Nats.JsMsg) => void | ||
| ${realizeParametersForChannelWrapper(channelParameters)}, | ||
| options: Nats.ConsumerOptsBuilder | Partial<Nats.ConsumerOpts> | ||
| ): Promise<Nats.JetStreamSubscription> { | ||
| return new Promise(async (resolve, reject) => { | ||
| if (!this.isClosed() && this.nc !== undefined && this.codec !== undefined && this.js !== undefined) { | ||
| try { | ||
| const sub = await ${camelCase(channelName)}Channel.jetStreamPushSubscribe( | ||
| onDataCallback, | ||
| this.js, | ||
| this.codec, | ||
| ${Object.keys(channelParameters).length ? `${realizeParametersForChannelWithoutType(channelParameters)},` : ''} | ||
| options | ||
| ); | ||
| resolve(sub); | ||
| } catch (e: any) { | ||
| reject(e); | ||
| } | ||
| } else { | ||
| reject(NatsTypescriptTemplateError.errorForCode(ErrorCode.NOT_CONNECTED)); | ||
| } | ||
| }); | ||
| } | ||
| `; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.