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 pull subscribe wrappers (#480)
- Loading branch information
1 parent
5689ba0
commit d1e9ed5
Showing
12 changed files
with
359 additions
and
9 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,70 @@ | ||
| 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 JetstreamPullSubscription(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 pull 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)} | ||
| */ | ||
| export function jetStreamPullSubscribe( | ||
| 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.JetStreamPullSubscription > { | ||
| return new Promise(async (resolve, reject) => { | ||
| try { | ||
| const subscription = await js.pullSubscribe(${realizeChannelName(channelParameters, channelName)}, options); | ||
| (async () => { | ||
| for await (const msg of subscription) { | ||
| ${unwrap(channelName, channelParameters)} | ||
| ${whenReceivingMessage} | ||
| } | ||
| })(); | ||
| 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 JetstreamPullSubscribe(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 jetStreamPullSubscribeTo${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.JetStreamPullSubscription> { | ||
| return new Promise(async (resolve, reject) => { | ||
| if (!this.isClosed() && this.nc !== undefined && this.codec !== undefined && this.js !== undefined) { | ||
| try { | ||
| const sub = ${camelCase(channelName)}Channel.jetStreamPullSubscribe( | ||
| 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
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.