feat: enable NATS functions to have header models if present

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

@@ -5,6 +5,7 @@ import {
 } from '@asyncapi/parser';
 import {TypeScriptParameterRenderType} from '../parameters';
 import {TypeScriptPayloadRenderType} from '../payloads';
+import {TypeScriptHeadersRenderType} from '../headers';
 import {
   ChannelFunctionTypes,
   TypeScriptChannelRenderedFunctionType,
@@ -91,6 +92,7 @@ export async function generateTypeScriptChannelsForAsyncAPI(
   context: TypeScriptChannelsContext,
   parameters: TypeScriptParameterRenderType,
   payloads: TypeScriptPayloadRenderType,
+  headers: TypeScriptHeadersRenderType,
   protocolsToUse: SupportedProtocols[],
   protocolCodeFunctions: Record<string, string[]>,
   externalProtocolFunctionInformation: Record<
@@ -117,12 +119,17 @@ export async function generateTypeScriptChannelsForAsyncAPI(
       }
     }
 
+    // Headers are now always available, but the specific channel may not have headers defined
+    const headerModel: OutputModel | undefined =
+      headers.channelModels[channel.id()];
+
     for (const protocol of protocolsToUse) {
       const protocolContext: TypeScriptChannelsGeneratorContext = {
         ...context,
         subName,
         topic: channel.address()!,
         parameter: parameter?.model as ConstrainedObjectModel,
+        headers: headerModel?.model as ConstrainedObjectModel,
         payloads
       };
 

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

@@ -14,6 +14,11 @@ import {
   TypeScriptPayloadRenderType,
   defaultTypeScriptPayloadGenerator
 } from '../payloads';
+import {
+  TypeScriptHeadersRenderType,
+  TypescriptHeadersGenerator,
+  defaultTypeScriptHeadersOptions
+} from '../headers';
 import {
   TypeScriptChannelRenderedFunctionType,
   TypeScriptChannelRenderType,
@@ -24,7 +29,11 @@ import {
   zodTypescriptChannelsGenerator,
   ChannelFunctionTypes
 } from './types';
-import {addParametersToDependencies, addPayloadsToDependencies} from './utils';
+import {
+  addParametersToDependencies,
+  addPayloadsToDependencies,
+  addHeadersToDependencies
+} from './utils';
 import {generateTypeScriptChannelsForAsyncAPI} from './asyncapi';
 export {
   TypeScriptChannelRenderedFunctionType,
@@ -50,7 +59,7 @@ export async function generateTypeScriptChannels(
     TypeScriptChannelRenderedFunctionType[]
   > = {};
 
-  const {parameters, payloads} = validateContext(context);
+  const {parameters, payloads, headers} = validateContext(context);
   const generator = context.generator;
 
   const protocolsToUse = generator.protocols;
@@ -59,12 +68,13 @@ export async function generateTypeScriptChannels(
     externalProtocolFunctionInformation[protocol] = [];
   }
   const dependencies: string[] = [];
-  addDependencies(payloads, parameters, context, dependencies);
+  addDependencies(payloads, parameters, headers, context, dependencies);
   if (context.inputType === 'asyncapi') {
     await generateTypeScriptChannelsForAsyncAPI(
       context,
       parameters,
       payloads,
+      headers,
       protocolsToUse,
       protocolCodeFunctions,
       externalProtocolFunctionInformation,
@@ -121,6 +131,7 @@ ${Object.entries(protocolCodeFunctions)
 function addDependencies(
   payloads: TypeScriptPayloadRenderType,
   parameters: TypeScriptParameterRenderType,
+  headers: TypeScriptHeadersRenderType | undefined,
   context: TypeScriptChannelsContext,
   dependencies: string[]
 ) {
@@ -148,11 +159,20 @@ function addDependencies(
     context.generator,
     dependencies
   );
+  if (headers) {
+    addHeadersToDependencies(
+      headers.channelModels,
+      headers.generator,
+      context.generator,
+      dependencies
+    );
+  }
 }
 
 function validateContext(context: TypeScriptChannelsContext): {
   payloads: TypeScriptPayloadRenderType;
   parameters: TypeScriptParameterRenderType;
+  headers: TypeScriptHeadersRenderType;
 } {
   const {generator} = context;
   if (!context.dependencyOutputs) {
@@ -166,6 +186,9 @@ function validateContext(context: TypeScriptChannelsContext): {
   const parameters = context.dependencyOutputs[
     generator.parameterGeneratorId
   ] as TypeScriptParameterRenderType;
+  const headers = context.dependencyOutputs[
+    generator.headerGeneratorId
+  ] as TypeScriptHeadersRenderType;
   if (!payloads) {
     throw new Error(
       'Internal error, could not determine previous rendered payloads generator that is required for channel TypeScript generator'
@@ -176,7 +199,12 @@ function validateContext(context: TypeScriptChannelsContext): {
       'Internal error, could not determine previous rendered parameters generator that is required for channel TypeScript generator'
     );
   }
-  return {payloads, parameters};
+  if (!headers) {
+    throw new Error(
+      'Internal error, could not determine previous rendered headers generator that is required for channel TypeScript generator'
+    );
+  }
+  return {payloads, parameters, headers};
 }
 
 /**
@@ -189,6 +217,7 @@ export function includeTypeScriptChannelDependencies(
   const newGenerators: any[] = [];
   const parameterGeneratorId = generator.parameterGeneratorId;
   const payloadGeneratorId = generator.payloadGeneratorId;
+  const headerGeneratorId = generator.headerGeneratorId;
   const hasParameterGenerator =
     config.generators.find(
       (generatorSearch) => generatorSearch.id === parameterGeneratorId
@@ -197,6 +226,10 @@ export function includeTypeScriptChannelDependencies(
     config.generators.find(
       (generatorSearch) => generatorSearch.id === payloadGeneratorId
     ) !== undefined;
+  const hasHeaderGenerator =
+    config.generators.find(
+      (generatorSearch) => generatorSearch.id === headerGeneratorId
+    ) !== undefined;
   if (!hasParameterGenerator) {
     const defaultChannelParameterGenerator: TypescriptParametersGenerator = {
       ...defaultTypeScriptParametersOptions,
@@ -211,5 +244,12 @@ export function includeTypeScriptChannelDependencies(
     };
     newGenerators.push(defaultChannelPayloadGenerator);
   }
+  if (!hasHeaderGenerator) {
+    const defaultChannelHeaderGenerator: TypescriptHeadersGenerator = {
+      ...defaultTypeScriptHeadersOptions,
+      outputPath: path.resolve(generator.outputPath ?? '', './headers')
+    };
+    newGenerators.push(defaultChannelHeaderGenerator);
+  }
   return newGenerators;
 }

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

@@ -36,6 +36,7 @@ export async function generateAmqpChannels(
 
   const amqpContext: RenderRegularParameters = {
     channelParameters: parameter,
+    channelHeaders: undefined, // AMQP doesn't support headers in this implementation
     topic,
     messageType: '',
     subName: context.subName,

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

@@ -35,6 +35,7 @@ export async function generateEventSourceChannels(
 
   const eventSourceContext: RenderRegularParameters = {
     channelParameters: parameter,
+    channelHeaders: undefined, // EventSource doesn't support headers in this implementation
     topic,
     messageType: '',
     subName: context.subName,

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

@@ -37,6 +37,7 @@ export async function generateKafkaChannels(
 
   const kafkaContext: RenderRegularParameters = {
     channelParameters: parameter,
+    channelHeaders: undefined, // Kafka doesn't support headers in this implementation
     topic: kafkaTopic,
     messageType: '',
     subName: context.subName,

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

@@ -33,6 +33,7 @@ export async function generateMqttChannels(
   const ignoreOperation = !generator.asyncapiGenerateForOperations;
   const mqttContext: RenderRegularParameters = {
     channelParameters: parameter,
+    channelHeaders: undefined, // MQTT doesn't support headers in this implementation
     topic,
     subName: context.subName,
     messageType: '',

src/codegen/generators/typescript/channels/protocols/nats/corePublish.ts

@@ -3,12 +3,14 @@ import {ChannelFunctionTypes} from '../..';
 import {SingleFunctionRenderType} from '../../../../../types';
 import {pascalCase} from '../../../utils';
 import {RenderRegularParameters} from '../../types';
+import {generateHeaderSetupCode, generateHeaderParameter} from './utils';
 
 export function renderCorePublish({
   topic,
   messageType,
   messageModule,
   channelParameters,
+  channelHeaders,
   subName = pascalCase(topic),
   functionName = `publishTo${subName}`
 }: RenderRegularParameters): SingleFunctionRenderType {
@@ -20,13 +22,18 @@ export function renderCorePublish({
     messageMarshalling = `${messageModule}.marshal(message)`;
   }
   messageType = messageModule ? `${messageModule}.${messageType}` : messageType;
+  const headerSetup = generateHeaderSetupCode(channelHeaders);
+
   const publishOperation =
     messageType === 'null'
-      ? `await nc.publish(${addressToUse}, Nats.Empty, options);`
+      ? `${headerSetup}
+      await nc.publish(${addressToUse}, Nats.Empty, options);`
       : `let dataToSend: any = ${messageMarshalling};
+      ${headerSetup}
 dataToSend = codec.encode(dataToSend);
 nc.publish(${addressToUse}, dataToSend, options);`;
 
+  const headerParam = generateHeaderParameter(channelHeaders);
   const functionParameters = [
     {
       parameter: `message`,
@@ -42,6 +49,7 @@ nc.publish(${addressToUse}, dataToSend, options);`;
           }
         ]
       : []),
+    ...(headerParam ? [headerParam] : []),
     {
       parameter: 'nc',
       parameterType: 'nc: Nats.NatsConnection',

src/codegen/generators/typescript/channels/protocols/nats/coreSubscribe.ts

@@ -5,12 +5,18 @@ import {SingleFunctionRenderType} from '../../../../../types';
 import {findRegexFromChannel, pascalCase} from '../../../utils';
 import {RenderRegularParameters} from '../../types';
 import {getValidationFunctions} from '../../utils';
+import {
+  generateHeaderExtractionCode,
+  generateHeaderCallbackParameter,
+  generateMessageReceivingCode
+} from './utils';
 
 export function renderCoreSubscribe({
   topic,
   messageType,
   messageModule,
   channelParameters,
+  channelHeaders,
   subName = pascalCase(topic),
   payloadGenerator,
   functionName = `subscribeTo${subName}`
@@ -29,13 +35,19 @@ export function renderCoreSubscribe({
       includeValidation,
       messageModule,
       messageType,
-      onValidationFail: channelParameters
-        ? `onDataCallback(new Error(\`Invalid message payload received; $\{JSON.stringify({cause: errors})}\`), undefined, parameters, msg); continue;`
-        : `onDataCallback(new Error(\`Invalid message payload received; $\{JSON.stringify({cause: errors})}\`), undefined, msg); continue;`
+      onValidationFail:
+        channelParameters && channelHeaders
+          ? `onDataCallback(new Error(\`Invalid message payload received; $\{JSON.stringify({cause: errors})}\`), undefined, parameters, extractedHeaders, msg); continue;`
+          : channelParameters
+            ? `onDataCallback(new Error(\`Invalid message payload received; $\{JSON.stringify({cause: errors})}\`), undefined, parameters, msg); continue;`
+            : channelHeaders
+              ? `onDataCallback(new Error(\`Invalid message payload received; $\{JSON.stringify({cause: errors})}\`), undefined, extractedHeaders, msg); continue;`
+              : `onDataCallback(new Error(\`Invalid message payload received; $\{JSON.stringify({cause: errors})}\`), undefined, msg); continue;`
     });
 
   messageType = messageModule ? `${messageModule}.${messageType}` : messageType;
 
+  const headerCallbackParam = generateHeaderCallbackParameter(channelHeaders);
   const callbackFunctionParameters = [
     {
       parameter: 'err?: Error',
@@ -53,6 +65,7 @@ export function renderCoreSubscribe({
           }
         ]
       : []),
+    ...(headerCallbackParam ? [headerCallbackParam] : []),
     {
       parameter: `natsMsg?: Nats.Msg`,
       jsDoc: ' * @param natsMsg'
@@ -97,22 +110,15 @@ export function renderCoreSubscribe({
         ' * @param skipMessageValidation turn off runtime validation of incoming messages'
     }
   ];
-  let whenReceivingMessage = '';
-  if (channelParameters) {
-    if (messageType === 'null') {
-      whenReceivingMessage = `onDataCallback(undefined, null, parameters, msg);`;
-    } else {
-      whenReceivingMessage = `let receivedData: any = codec.decode(msg.data);
-${potentialValidationFunction}
-onDataCallback(undefined, ${messageUnmarshalling}, parameters, msg);`;
-    }
-  } else if (messageType === 'null') {
-    whenReceivingMessage = `onDataCallback(undefined, null, msg);`;
-  } else {
-    whenReceivingMessage = `let receivedData: any = codec.decode(msg.data);
-${potentialValidationFunction}
-onDataCallback(undefined, ${messageUnmarshalling}, msg);`;
-  }
+  const headerExtraction = generateHeaderExtractionCode(channelHeaders);
+  const whenReceivingMessage = generateMessageReceivingCode({
+    channelParameters,
+    channelHeaders,
+    messageType,
+    messageUnmarshalling,
+    headerExtraction,
+    potentialValidationFunction
+  });
   const jsDocParameters = functionParameters
     .map((param) => param.jsDoc)
     .join('\n');

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

@@ -48,13 +48,14 @@ export async function generateNatsChannels(
   >,
   dependencies: string[]
 ) {
-  const {parameter, topic, payloads} = context;
+  const {parameter, headers, topic, payloads} = context;
   const ignoreOperation = !context.generator.asyncapiGenerateForOperations;
   let natsTopic = topic.startsWith('/') ? topic.slice(1) : topic;
   natsTopic = natsTopic.replace(/\//g, '.');
 
   const natsContext: RenderRegularParameters = {
     channelParameters: parameter,
+    channelHeaders: headers,
     topic: natsTopic,
     messageType: '',
     subName: context.subName,

src/codegen/generators/typescript/channels/protocols/nats/jetstreamPublish.ts

@@ -3,11 +3,13 @@ import {SingleFunctionRenderType} from '../../../../../types';
 import {pascalCase} from '../../../utils';
 import {ChannelFunctionTypes} from '../../index';
 import {RenderRegularParameters} from '../../types';
+import {generateHeaderSetupCode, generateHeaderParameter} from './utils';
 export function renderJetstreamPublish({
   topic,
   messageType,
   messageModule,
   channelParameters,
+  channelHeaders,
   subName = pascalCase(topic),
   functionName = `jetStreamPublishTo${subName}`
 }: RenderRegularParameters): SingleFunctionRenderType {
@@ -20,13 +22,18 @@ export function renderJetstreamPublish({
   }
   messageType = messageModule ? `${messageModule}.${messageType}` : messageType;
 
+  const headerSetup = generateHeaderSetupCode(channelHeaders);
+
   const publishOperation =
     messageType === 'null'
-      ? `await js.publish(${addressToUse}, Nats.Empty, options);`
+      ? `${headerSetup}
+      await js.publish(${addressToUse}, Nats.Empty, options);`
       : `let dataToSend: any = ${messageMarshalling};
+      ${headerSetup}
 dataToSend = codec.encode(dataToSend);
 await js.publish(${addressToUse}, dataToSend, options);`;
 
+  const headerParam = generateHeaderParameter(channelHeaders);
   const functionParameters = [
     {
       parameter: `message`,
@@ -42,6 +49,7 @@ await js.publish(${addressToUse}, dataToSend, options);`;
           }
         ]
       : []),
+    ...(headerParam ? [headerParam] : []),
     {
       parameter: 'js',
       parameterType: 'js: Nats.JetStreamClient',