16003527935

Committed 01 Jul 2025 03:23PM UTC coverage: 80.116% (-0.4%) from 80.505%

Build # 16003527935

Build Type

push

github

Committed by

web-flow

Commit Message

fix: Meter for batch consumer (#147)

* fix: Meter for batch consumer

* debug: Not send links field if link empty

---------

Co-authored-by: gleip <kuchinsn@lad24.ru>

Coverage Stats

414 of 561 branches covered (73.8%)

Branch coverage included in aggregate %.

14 of 20 new or added lines in 2 files covered. (70.0%)

972 of 1169 relevant lines covered (83.15%)

5.45 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.25

/src/Client.ts

import * as opentelemetry from '@opentelemetry/api';
import Ajv from 'ajv';
import { JsMsg, JSONCodec, Msg, Subscription, StringCodec, MsgHdrs } from 'nats';
import { createHash } from 'node:crypto';
import * as http from 'node:http';
import { EventEmitter, Readable } from 'node:stream';
import { setTimeout } from 'node:timers/promises';
import { CacheSettings } from '.';
import {
  Baggage,
  ClientParam,
  Emitter,
  EmitterStreamEvent,
  Events,
  GetBatchListenerOptions,
  GetListenerOptions,
  HttpSettings,
  Listener,
  ListenerBatch,
  Message,
  MethodOptions,
  MethodSettings,
} from './interfaces';
import { Root } from './Root';
import { StreamManager } from './StreamManager';
import { StreamBatchMsgFetcher } from './StreamBatchMsgFetcher';
import { StreamSingleMsgFetcher } from './StreamSingleMsgFetcher';
import { Meter } from './Meter';

type RequestData = Record<string, unknown> | Readable;
export class Client<E extends Emitter = Emitter> extends Root {
  private serviceName: string;
  private baggage?: Baggage;
  private cache?: CacheSettings;
  private events?: Events<E>;
  private Ref?: object;

  private subscriptions = new Map<keyof E, Subscription | StreamSingleMsgFetcher>();
  private REQUEST_HTTP_SETTINGS_TIMEOUT = 1000; // ms

  constructor({ broker, events, loggerOutputFormatter, serviceName, baggage, cache, Ref }: ClientParam<E>) {
    super(broker, loggerOutputFormatter);
    this.logger.setLocation(serviceName);
    this.serviceName = serviceName;
    this.baggage = baggage;
    this.cache = cache;
    this.events = events;
    this.Ref = Ref;
  }

  private async startWatch(
    subscription: Subscription | StreamSingleMsgFetcher,
    listener: EventEmitter,
    eventName: string,
  ) {
    for await (const event of subscription) {
      if (!event) {
        continue;
      }
      let data: unknown;
      try {
        data = JSONCodec<unknown>().decode(event.data);
      } catch (error) {
        data = StringCodec().decode(event.data);
      }
      const message: Partial<EmitterStreamEvent<any>> = { data };

      let baggage: Baggage | undefined;
      if (event.headers) {
        baggage = this.getBaggageFromNATSHeader(event.headers);
      }

      message.meter = new Meter(eventName, baggage);

      if (this.isJsMessage(event)) {
        message.ack = event.ack.bind(event);
        message.nak = event.nak.bind(event);
      }

      listener.emit(eventName, message);
    }
  }

  private async startBatchWatch(fetcher: StreamBatchMsgFetcher, listener: EventEmitter, eventName: string) {
    while (true) {
      const batch: Partial<EmitterStreamEvent<any>>[] = [];
      const baggages: Baggage[] = [];

      const events = await fetcher.fetch();

      for await (const event of events) {
        let data: unknown;
        try {
          data = JSONCodec<unknown>().decode(event.data);
        } catch (error) {
          data = StringCodec().decode(event.data);
        }
        const message: Partial<EmitterStreamEvent<any>> = { data };
        message.ack = event.ack.bind(event);
        message.nak = event.nak.bind(event);

        batch.push(message);

        let baggage: Baggage | undefined;
        if (event.headers) {
          baggage = this.getBaggageFromNATSHeader(event.headers);
          if (baggage) {
            baggages.push(baggage);
          }
        }

        message.meter = new Meter(eventName, baggage);
      }

      if (batch.length > 0) {
        const meter = new Meter(eventName, undefined, baggages);
        listener.emit(eventName, batch, meter);
      }
    }
  }

  /**
   * Make listener for service events. Auto subscribe and unsubscribe to subject
   */
  public getListener<A extends keyof E>(serviceNameFrom: string, options?: GetListenerOptions): Listener<E>;
  public getListener<A extends keyof E>(serviceNameFrom: string, options?: GetBatchListenerOptions): ListenerBatch<E>;
  public getListener<A extends keyof E>(
    serviceNameFrom: string,
    options?: GetListenerOptions | GetBatchListenerOptions,
  ): ListenerBatch<E> | ListenerBatch<E> {
    if (!this.events) {
      throw new Error('The service does not generate events');
    }

    const listener = new EventEmitter();

    return new Proxy(listener, {
      get: (target, prop, receiver) => {
        const method = Reflect.get(target, prop, receiver);

        if (prop === 'on') {
          return async (eventName: A, handler: (param: unknown) => void) => {
            try {
              this.logger.info('Subscribe', eventName);

              const action = this.events?.list[eventName];

              if (!action) {
                throw new Error(`The service does not generate ${String(eventName)} event`);
              }

              const isStream = action.options?.stream;

              let subscription: Subscription | StreamSingleMsgFetcher | StreamBatchMsgFetcher;

              if (isStream) {
                subscription = await new StreamManager({
                  broker: this.broker,
                  options: this.events!.streamOptions,
                  serviceName: this.serviceName,
                }).createConsumer(serviceNameFrom, String(eventName), options);
              } else {
                const queue = options?.queue ? { queue: options.queue } : {};
                subscription = this.broker.subscribe(`${this.serviceName}.${eventName as string}`, queue);
              }

              this.subscriptions.set(eventName, subscription);
              if (StreamManager.isStreamFetcher(subscription) && StreamManager.isPullConsumerOptions(options)) {
                this.startBatchWatch(subscription, listener, String(eventName));
              } else {
                this.startWatch(subscription, listener, String(eventName));
              }

              return method.call(target, eventName, handler);
            } catch (error) {
              const errorMessage = 'Failed subscribe to subject';
              this.logger.error(errorMessage, error);
              throw new Error(`${errorMessage} ${this.serviceName}.${String(eventName)}`);
            }
          };
        }

        if (prop === 'off') {
          return (eventName: A, listener: (params: unknown) => void) => {
            this.logger.info('Unsubscribe', eventName);
            const subscription = this.subscriptions.get(eventName);
            subscription?.unsubscribe();
            this.subscriptions.delete(eventName);
            return method.call(target, eventName, listener);
          };
        }
        return method;
      },
    }) as Listener<E>;
  }

  private createCacheKey(subject: string, data: Record<string, unknown>) {
    const dataHash = createHash('sha1').update(JSON.stringify(data)).digest('hex');
    return `${this.CACHE_SERVICE_KEY}:${subject}:${dataHash}`;
  }

  private validate(data: any, schema: Record<string, unknown>) {
    const ajv = new Ajv();
    if (this.Ref) {
      ajv.addSchema(this.Ref);
    }

    const requestValidator = ajv.compile(schema);
    const valid = requestValidator(data);
    if (!valid) {
      throw new Error(JSON.stringify(requestValidator.errors));
    }
  }

  protected async request<R = any, P extends RequestData = RequestData>(
    subject: string,
    data: P,
    { options, request, response }: MethodSettings,
  ): Promise<R> {
    const tracer = opentelemetry.trace.getTracer('');
    const span = tracer.startSpan(subject, undefined, this.getContext(this.baggage));
    try {
      if (options?.runTimeValidation?.request && request) {
        this.validate(data, request);
      }

      const { spanId, traceId, traceFlags } = span.spanContext();
      const expired = this.getExpired(this.baggage?.expired, options?.timeout);
      const requestId = this.baggage?.requestId;
      const message: Message = { payload: data, baggage: { expired, traceId, spanId, traceFlags, requestId } };
      const timeout = expired - Date.now();

      if (timeout <= 0) {
        throw new Error('Timeout request service ' + subject);
      }

      let key = '';

      if (options?.cache && !this.isStream(data) && this.cache) {
        try {
          key = this.createCacheKey(subject, data);
          const result = await Promise.race<any>([this.cache.service.get(key), setTimeout(this.cache.timeout, null)]);
          if (result) {
            return JSON.parse(result);
          }
        } catch (error) {
          this.logger.warn('get cache: ', error);
        }
      }
      const result = options?.useStream
        ? await this.makeHttpRequest(subject, message, options, timeout)
        : await this.makeBrokerRequest(subject, message, timeout);

      if (result.error) {
        throw new Error(result.error.message ?? result.error);
      }

      if (options?.runTimeValidation?.response && response) {
        this.validate(result.payload, response);
      }
      if (options?.cache && !this.isStream(result.payload) && this.cache) {
        this.cache.service.set(key, JSON.stringify(result.payload), options.cache);
      }

      return result.payload;
    } catch (error) {
      span.setAttribute('error', true);
      span.setAttribute('error.kind', error);
      this.logger.error(error);
      throw error;
    } finally {
      span.end();
    }
  }

  private async getHTTPSettingsFromRemoteService() {
    const subject = `${this.serviceName}.${this.SERVICE_SUBJECT_FOR_GET_HTTP_SETTINGS}`;
    const result = await this.broker.request(subject, Buffer.from(JSON.stringify('')), {
      timeout: this.REQUEST_HTTP_SETTINGS_TIMEOUT,
    });
    const { ip, port } = JSONCodec<HttpSettings>().decode(result.data);
    if (!ip || !port) {
      throw new Error(`Remote service ${this.serviceName} did not return http settings`);
    }
    return { ip, port };
  }

  private isStream(data: unknown | Readable): data is Readable {
    return data instanceof Readable;
  }

  private async makeBrokerRequest(subject: string, message: Message, timeout: number) {
    try {
      const result = await this.broker.request(subject, Buffer.from(JSON.stringify(message)), { timeout });
      return JSONCodec<Message<any>>().decode(result.data);
    } catch (error) {
      const errorMessage = new Error(`${error?.message}. Subject: ${subject} `);
      return this.buildErrorMessage(errorMessage);
    }
  }

  private async makeHttpRequest(
    subject: string,
    message: Message | Message<Readable>,
    options: MethodOptions,
    timeout: number,
  ): Promise<Message<any>> {
    return new Promise(async resolve => {
      const { ip, port } = await this.getHTTPSettingsFromRemoteService();
      const serviceActionPath = subject.split('.');
      const headersFromBaggage = this.convertBaggaggeToExternalHeader(message.baggage);

      const headers = {
        'Content-Type': this.isStream(message.payload) ? 'application/octet-stream' : 'application/json',
        ...headersFromBaggage,
      };
      if (!this.isStream(message.payload)) {
        headers['Content-Length'] = Buffer.byteLength(JSON.stringify(message.payload));
      }
      const request = http.request(
        {
          host: ip,
          port,
          path: `/${serviceActionPath.join('/')}`,
          method: 'POST',
          headers,
          timeout,
        },
        async response => {
          if (options?.useStream?.response && response.statusCode !== 500) {
            resolve({ payload: response });
            return;
          }

          const data: Buffer[] = [];
          for await (const chunk of response) {
            data.push(chunk);
          }

          const responseDataString = Buffer.concat(data).toString();
          try {
            resolve(JSON.parse(responseDataString));
          } catch (error) {
            const errorMessage = new Error(`${error?.message}. Subject: ${subject} `);
            resolve(this.buildErrorMessage(errorMessage));
          }
        },
      );

      request.on('error', error => {
        const errorMessage = new Error(`${error?.message}. Subject: ${subject} `);
        resolve(this.buildErrorMessage(errorMessage));
      });

      if (this.isStream(message.payload)) {
        message.payload.pipe(request);
        return;
      }

      request.write(JSON.stringify(message.payload));
      request.end();
    });
  }

  private isJsMessage(message: JsMsg | Msg): message is JsMsg {
    return !!(message as JsMsg).ack && !!(message as JsMsg).nak;
  }

  private getBaggageFromNATSHeader(headers: MsgHdrs): Baggage | undefined {
    const traceId = headers.get('nsc-trace-id');
    const spanId = headers.get('nsc-span-id');
    const traceFlags = headers.has('nsc-trace-flags') ? +headers.get('nsc-trace-flags') : undefined;
    const requestId = headers.has('x-request-id') ? String(headers.get('x-request-id')) : undefined;

    if (traceId && spanId && traceFlags) {
      return {
        traceId,
        spanId,
        traceFlags,
        requestId,
      };
    }
  }
}

1	import * as opentelemetry from '@opentelemetry/api';	4✔
2	import Ajv from 'ajv';	4✔
3	import { JsMsg, JSONCodec, Msg, Subscription, StringCodec, MsgHdrs } from 'nats';	4✔
4	import { createHash } from 'node:crypto';	4✔
5	import * as http from 'node:http';	4✔
6	import { EventEmitter, Readable } from 'node:stream';	4✔
7	import { setTimeout } from 'node:timers/promises';	4✔
8	import { CacheSettings } from '.';
9	import {
10	Baggage,
11	ClientParam,
12	Emitter,
13	EmitterStreamEvent,
14	Events,
15	GetBatchListenerOptions,
16	GetListenerOptions,
17	HttpSettings,
18	Listener,
19	ListenerBatch,
20	Message,
21	MethodOptions,
22	MethodSettings,
23	} from './interfaces';
24	import { Root } from './Root';	4✔
25	import { StreamManager } from './StreamManager';	4✔
26	import { StreamBatchMsgFetcher } from './StreamBatchMsgFetcher';
27	import { StreamSingleMsgFetcher } from './StreamSingleMsgFetcher';
28	import { Meter } from './Meter';	4✔
29
30	type RequestData = Record<string, unknown> \| Readable;
31	export class Client<E extends Emitter = Emitter> extends Root {	4✔
32	private serviceName: string;
33	private baggage?: Baggage;
34	private cache?: CacheSettings;
35	private events?: Events<E>;
36	private Ref?: object;
37
38	private subscriptions = new Map<keyof E, Subscription \| StreamSingleMsgFetcher>();	9✔
39	private REQUEST_HTTP_SETTINGS_TIMEOUT = 1000; // ms	9✔
40
41	constructor({ broker, events, loggerOutputFormatter, serviceName, baggage, cache, Ref }: ClientParam<E>) {
42	super(broker, loggerOutputFormatter);	9✔
43	this.logger.setLocation(serviceName);	9✔
44	this.serviceName = serviceName;	9✔
45	this.baggage = baggage;	9✔
46	this.cache = cache;	9✔
47	this.events = events;	9✔
48	this.Ref = Ref;	9✔
49	}
50
51	private async startWatch(
52	subscription: Subscription \| StreamSingleMsgFetcher,
53	listener: EventEmitter,
54	eventName: string,
55	) {
56	for await (const event of subscription) {	3✔
57	if (!event) {	2!
58	continue;	×
59	}
60	let data: unknown;
61	try {	2✔
62	data = JSONCodec<unknown>().decode(event.data);	2✔
63	} catch (error) {
64	data = StringCodec().decode(event.data);	×
65	}
66	const message: Partial<EmitterStreamEvent<any>> = { data };	2✔
67
68	let baggage: Baggage \| undefined;
69	if (event.headers) {	2!
70	baggage = this.getBaggageFromNATSHeader(event.headers);	×
71	}
72
73	message.meter = new Meter(eventName, baggage);	2✔
74
75	if (this.isJsMessage(event)) {	2✔
76	message.ack = event.ack.bind(event);	1✔
77	message.nak = event.nak.bind(event);	1✔
78	}
79
80	listener.emit(eventName, message);	2✔
81	}
82	}
83
84	private async startBatchWatch(fetcher: StreamBatchMsgFetcher, listener: EventEmitter, eventName: string) {
85	while (true) {	1✔
86	const batch: Partial<EmitterStreamEvent<any>>[] = [];	2✔
87	const baggages: Baggage[] = [];	2✔
88
89	const events = await fetcher.fetch();	2✔
90
91	for await (const event of events) {	2✔
92	let data: unknown;
93	try {	3✔
94	data = JSONCodec<unknown>().decode(event.data);	3✔
95	} catch (error) {
96	data = StringCodec().decode(event.data);	×
97	}
98	const message: Partial<EmitterStreamEvent<any>> = { data };	3✔
99	message.ack = event.ack.bind(event);	3✔
100	message.nak = event.nak.bind(event);	3✔
101
102	batch.push(message);	3✔
103
104	let baggage: Baggage \| undefined;
105	if (event.headers) {	3!
NEW 106	baggage = this.getBaggageFromNATSHeader(event.headers);	×
NEW 107	if (baggage) {	×
NEW 108	baggages.push(baggage);	×
109	}
110	}
111
112	message.meter = new Meter(eventName, baggage);	3✔
113	}
114
115	if (batch.length > 0) {	1✔
116	const meter = new Meter(eventName, undefined, baggages);	1✔
117	listener.emit(eventName, batch, meter);	1✔
118	}
119	}
120	}
121
122	/**
123	* Make listener for service events. Auto subscribe and unsubscribe to subject
124	*/
125	public getListener<A extends keyof E>(serviceNameFrom: string, options?: GetListenerOptions): Listener<E>;
126	public getListener<A extends keyof E>(serviceNameFrom: string, options?: GetBatchListenerOptions): ListenerBatch<E>;
127	public getListener<A extends keyof E>(
128	serviceNameFrom: string,
129	options?: GetListenerOptions \| GetBatchListenerOptions,
130	): ListenerBatch<E> \| ListenerBatch<E> {
131	if (!this.events) {	5!
132	throw new Error('The service does not generate events');	×
133	}
134
135	const listener = new EventEmitter();	5✔
136
137	return new Proxy(listener, {	5✔
138	get: (target, prop, receiver) => {
139	const method = Reflect.get(target, prop, receiver);	12✔
140
141	if (prop === 'on') {	12✔
142	return async (eventName: A, handler: (param: unknown) => void) => {	4✔
143	try {	4✔
144	this.logger.info('Subscribe', eventName);	4✔
145
146	const action = this.events?.list[eventName];	4!
147
148	if (!action) {	4!
149	throw new Error(`The service does not generate ${String(eventName)} event`);	×
150	}
151
152	const isStream = action.options?.stream;	4✔
153
154	let subscription: Subscription \| StreamSingleMsgFetcher \| StreamBatchMsgFetcher;
155
156	if (isStream) {	4✔
157	subscription = await new StreamManager({	2✔
158	broker: this.broker,
159	options: this.events!.streamOptions,
160	serviceName: this.serviceName,
161	}).createConsumer(serviceNameFrom, String(eventName), options);
162	} else {
163	const queue = options?.queue ? { queue: options.queue } : {};	2!
164	subscription = this.broker.subscribe(`${this.serviceName}.${eventName as string}`, queue);	2✔
165	}
166
167	this.subscriptions.set(eventName, subscription);	4✔
168	if (StreamManager.isStreamFetcher(subscription) && StreamManager.isPullConsumerOptions(options)) {	4✔
169	this.startBatchWatch(subscription, listener, String(eventName));	1✔
170	} else {
171	this.startWatch(subscription, listener, String(eventName));	3✔
172	}
173
174	return method.call(target, eventName, handler);	4✔
175	} catch (error) {
176	const errorMessage = 'Failed subscribe to subject';	×
177	this.logger.error(errorMessage, error);	×
178	throw new Error(`${errorMessage} ${this.serviceName}.${String(eventName)}`);	×
179	}
180	};
181	}
182
183	if (prop === 'off') {	8✔
184	return (eventName: A, listener: (params: unknown) => void) => {	2✔
185	this.logger.info('Unsubscribe', eventName);	2✔
186	const subscription = this.subscriptions.get(eventName);	2✔
187	subscription?.unsubscribe();	2!
188	this.subscriptions.delete(eventName);	2✔
189	return method.call(target, eventName, listener);	2✔
190	};
191	}
192	return method;	6✔
193	},
194	}) as Listener<E>;
195	}
196
197	private createCacheKey(subject: string, data: Record<string, unknown>) {
198	const dataHash = createHash('sha1').update(JSON.stringify(data)).digest('hex');	2✔
199	return `${this.CACHE_SERVICE_KEY}:${subject}:${dataHash}`;	2✔
200	}
201
202	private validate(data: any, schema: Record<string, unknown>) {
203	const ajv = new Ajv();	11✔
204	if (this.Ref) {	11!
205	ajv.addSchema(this.Ref);	×
206	}
207
208	const requestValidator = ajv.compile(schema);	11✔
209	const valid = requestValidator(data);	11✔
210	if (!valid) {	11✔
211	throw new Error(JSON.stringify(requestValidator.errors));	2✔
212	}
213	}
214
215	protected async request<R = any, P extends RequestData = RequestData>(
216	subject: string,
217	data: P,
218	{ options, request, response }: MethodSettings,
219	): Promise<R> {
220	const tracer = opentelemetry.trace.getTracer('');	17✔
221	const span = tracer.startSpan(subject, undefined, this.getContext(this.baggage));	17✔
222	try {	17✔
223	if (options?.runTimeValidation?.request && request) {	17✔
224	this.validate(data, request);	8✔
225	}
226
227	const { spanId, traceId, traceFlags } = span.spanContext();	16✔
228	const expired = this.getExpired(this.baggage?.expired, options?.timeout);	16✔
229	const requestId = this.baggage?.requestId;	16✔
230	const message: Message = { payload: data, baggage: { expired, traceId, spanId, traceFlags, requestId } };	16✔
231	const timeout = expired - Date.now();	16✔
232
233	if (timeout <= 0) {	16✔
234	throw new Error('Timeout request service ' + subject);	1✔
235	}
236
237	let key = '';	15✔
238
239	if (options?.cache && !this.isStream(data) && this.cache) {	15✔
240	try {	2✔
241	key = this.createCacheKey(subject, data);	2✔
242	const result = await Promise.race<any>([this.cache.service.get(key), setTimeout(this.cache.timeout, null)]);	2✔
243	if (result) {	2✔
244	return JSON.parse(result);	1✔
245	}
246	} catch (error) {
247	this.logger.warn('get cache: ', error);	×
248	}
249	}
250	const result = options?.useStream	14✔
251	? await this.makeHttpRequest(subject, message, options, timeout)
252	: await this.makeBrokerRequest(subject, message, timeout);
253
254	if (result.error) {	14✔
255	throw new Error(result.error.message ?? result.error);	1!
256	}
257
258	if (options?.runTimeValidation?.response && response) {	13✔
259	this.validate(result.payload, response);	3✔
260	}
261	if (options?.cache && !this.isStream(result.payload) && this.cache) {	12✔
262	this.cache.service.set(key, JSON.stringify(result.payload), options.cache);	1✔
263	}
264
265	return result.payload;	12✔
266	} catch (error) {
267	span.setAttribute('error', true);	4✔
268	span.setAttribute('error.kind', error);	4✔
269	this.logger.error(error);	4✔
270	throw error;	4✔
271	} finally {
272	span.end();	17✔
273	}
274	}
275
276	private async getHTTPSettingsFromRemoteService() {
277	const subject = `${this.serviceName}.${this.SERVICE_SUBJECT_FOR_GET_HTTP_SETTINGS}`;	5✔
278	const result = await this.broker.request(subject, Buffer.from(JSON.stringify('')), {	5✔
279	timeout: this.REQUEST_HTTP_SETTINGS_TIMEOUT,
280	});
281	const { ip, port } = JSONCodec<HttpSettings>().decode(result.data);	5✔
282	if (!ip \|\| !port) {	5!
283	throw new Error(`Remote service ${this.serviceName} did not return http settings`);	×
284	}
285	return { ip, port };	5✔
286	}
287
288	private isStream(data: unknown \| Readable): data is Readable {
289	return data instanceof Readable;	24✔
290	}
291
292	private async makeBrokerRequest(subject: string, message: Message, timeout: number) {
293	try {	9✔
294	const result = await this.broker.request(subject, Buffer.from(JSON.stringify(message)), { timeout });	9✔
295	return JSONCodec<Message<any>>().decode(result.data);	9✔
296	} catch (error) {
297	const errorMessage = new Error(`${error?.message}. Subject: ${subject} `);	×
298	return this.buildErrorMessage(errorMessage);	×
299	}
300	}
301
302	private async makeHttpRequest(
303	subject: string,
304	message: Message \| Message<Readable>,
305	options: MethodOptions,
306	timeout: number,
307	): Promise<Message<any>> {
308	return new Promise(async resolve => {	5✔
309	const { ip, port } = await this.getHTTPSettingsFromRemoteService();	5✔
310	const serviceActionPath = subject.split('.');	5✔
311	const headersFromBaggage = this.convertBaggaggeToExternalHeader(message.baggage);	5✔
312
313	const headers = {	5✔
314	'Content-Type': this.isStream(message.payload) ? 'application/octet-stream' : 'application/json',	5✔
315	...headersFromBaggage,
316	};
317	if (!this.isStream(message.payload)) {	5✔
318	headers['Content-Length'] = Buffer.byteLength(JSON.stringify(message.payload));	2✔
319	}
320	const request = http.request(	5✔
321	{
322	host: ip,
323	port,
324	path: `/${serviceActionPath.join('/')}`,
325	method: 'POST',
326	headers,
327	timeout,
328	},
329	async response => {
330	if (options?.useStream?.response && response.statusCode !== 500) {	5!
331	resolve({ payload: response });	3✔
332	return;	3✔
333	}
334
335	const data: Buffer[] = [];	2✔
336	for await (const chunk of response) {	2✔
337	data.push(chunk);	2✔
338	}
339
340	const responseDataString = Buffer.concat(data).toString();	2✔
341	try {	2✔
342	resolve(JSON.parse(responseDataString));	2✔
343	} catch (error) {
344	const errorMessage = new Error(`${error?.message}. Subject: ${subject} `);	×
345	resolve(this.buildErrorMessage(errorMessage));	×
346	}
347	},
348	);
349
350	request.on('error', error => {	5✔
351	const errorMessage = new Error(`${error?.message}. Subject: ${subject} `);	×
352	resolve(this.buildErrorMessage(errorMessage));	×
353	});
354
355	if (this.isStream(message.payload)) {	5✔
356	message.payload.pipe(request);	3✔
357	return;	3✔
358	}
359
360	request.write(JSON.stringify(message.payload));	2✔
361	request.end();	2✔
362	});
363	}
364
365	private isJsMessage(message: JsMsg \| Msg): message is JsMsg {
366	return !!(message as JsMsg).ack && !!(message as JsMsg).nak;	2✔
367	}
368
369	private getBaggageFromNATSHeader(headers: MsgHdrs): Baggage \| undefined {
370	const traceId = headers.get('nsc-trace-id');	×
371	const spanId = headers.get('nsc-span-id');	×
372	const traceFlags = headers.has('nsc-trace-flags') ? +headers.get('nsc-trace-flags') : undefined;	×
373	const requestId = headers.has('x-request-id') ? String(headers.get('x-request-id')) : undefined;	×
374
375	if (traceId && spanId && traceFlags) {	×
376	return {	×
377	traceId,
378	spanId,
379	traceFlags,
380	requestId,
381	};
382	}
383	}
384	}

lad-tech / nsc-toolkit / 16003527935

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous