13091454488

Committed 01 Feb 2025 06:48PM UTC coverage: 82.681% (-0.05%) from 82.727%

Build # 13091454488

Build Type

push

github

Committed by

web-flow

Commit Message

fix(jetstream): direct batch get missing `next_by_subj` filter (#201)

Signed-off-by: Alberto Ricart <alberto@synadia.com>

Run Details

2272 of 3100 branches covered (73.29%)

Branch coverage included in aggregate %.

9625 of 11289 relevant lines covered (85.26%)

787241.0 hits per line

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

83.5

/core/src/msg.ts

/*
 * Copyright 2020-2024 The NATS Authors
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
import { MsgHdrsImpl } from "./headers.ts";
import type { MsgArg } from "./parser.ts";
import { Empty, TD } from "./encoders.ts";
import type {
  Msg,
  MsgHdrs,
  Payload,
  Publisher,
  RequestInfo,
  ReviverFn,
} from "./core.ts";

export class MsgImpl implements Msg {
  _headers?: MsgHdrs;
  _msg: MsgArg;
  _rdata: Uint8Array;
  _reply!: string;
  _subject!: string;
  publisher: Publisher;

  constructor(msg: MsgArg, data: Uint8Array, publisher: Publisher) {
    this._msg = msg;
    this._rdata = data;
    this.publisher = publisher;
  }

  get subject(): string {
    if (this._subject) {
      return this._subject;
    }
    this._subject = TD.decode(this._msg.subject);
    return this._subject;
  }

  get reply(): string {
    if (this._reply) {
      return this._reply;
    }
    this._reply = TD.decode(this._msg.reply);
    return this._reply;
  }

  get sid(): number {
    return this._msg.sid;
  }

  get headers(): MsgHdrs | undefined {
    if (this._msg.hdr > -1 && !this._headers) {
      const buf = this._rdata.subarray(0, this._msg.hdr);
      this._headers = MsgHdrsImpl.decode(buf);
    }
    return this._headers;
  }

  get data(): Uint8Array {
    if (!this._rdata) {
      return new Uint8Array(0);
    }
    return this._msg.hdr > -1
      ? this._rdata.subarray(this._msg.hdr)
      : this._rdata;
  }

  // eslint-ignore-next-line @typescript-eslint/no-explicit-any
  respond(
    data: Payload = Empty,
    opts?: { headers?: MsgHdrs; reply?: string },
  ): boolean {
    if (this.reply) {
      this.publisher.publish(this.reply, data, opts);
      return true;
    }
    return false;
  }

  size(): number {
    const subj = this._msg.subject.length;
    const reply = this._msg.reply?.length || 0;
    const payloadAndHeaders = this._msg.size === -1 ? 0 : this._msg.size;
    return subj + reply + payloadAndHeaders;
  }

  json<T = unknown>(reviver?: ReviverFn): T {
    return JSON.parse(this.string(), reviver);
  }

  string(): string {
    return TD.decode(this.data);
  }

  requestInfo(): RequestInfo | null {
    const v = this.headers?.get("Nats-Request-Info");
    if (v) {
      return JSON.parse(
        v,
        function (this: unknown, key: string, value: unknown): unknown {
          if ((key === "start" || key === "stop") && value !== "") {
            return new Date(Date.parse(value as string));
          }
          return value;
        },
      ) as RequestInfo;
    }
    return null;
  }
}

1	/*
2	* Copyright 2020-2024 The NATS Authors
3	* Licensed under the Apache License, Version 2.0 (the "License");
4	* you may not use this file except in compliance with the License.
5	* You may obtain a copy of the License at
6	*
7	* http://www.apache.org/licenses/LICENSE-2.0
8	*
9	* Unless required by applicable law or agreed to in writing, software
10	* distributed under the License is distributed on an "AS IS" BASIS,
11	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	* See the License for the specific language governing permissions and
13	* limitations under the License.
14	*/
15	import { MsgHdrsImpl } from "./headers.ts";	55✔
16	import type { MsgArg } from "./parser.ts";
17	import { Empty, TD } from "./encoders.ts";	55✔
18	import type {
19	Msg,
20	MsgHdrs,
21	Payload,
22	Publisher,
23	RequestInfo,
24	ReviverFn,
25	} from "./core.ts";
26
27	export class MsgImpl implements Msg {	55✔
28	_headers?: MsgHdrs;	54✔
29	_msg: MsgArg;	395,822✔
30	_rdata: Uint8Array;	395,822✔
31	_reply!: string;	395,822✔
32	_subject!: string;	395,822✔
33	publisher: Publisher;	54✔
34
35	constructor(msg: MsgArg, data: Uint8Array, publisher: Publisher) {	54✔
36	this._msg = msg;	395,822✔
37	this._rdata = data;	395,822✔
38	this.publisher = publisher;	395,822✔
39	}	395,822✔
40
41	get subject(): string {	54✔
42	if (this._subject) {	116,082✔
43	return this._subject;	117,736✔
44	}	117,736✔
45	this._subject = TD.decode(this._msg.subject);	245,315✔
46	return this._subject;	245,315✔
47	}	129,257✔
48
49	get reply(): string {	54✔
50	if (this._reply) {	12,100✔
51	return this._reply;	30,262✔
52	}	30,262✔
53	this._reply = TD.decode(this._msg.reply);	29,232✔
54	return this._reply;	29,232✔
55	}	12,510✔
56
57	get sid(): number {	51✔
58	return this._msg.sid;	54✔
59	}	54✔
60
61	get headers(): MsgHdrs \| undefined {	53✔
62	if (this._msg.hdr > -1 && !this._headers) {	1,764✔
63	const buf = this._rdata.subarray(0, this._msg.hdr);	2,287✔
64	this._headers = MsgHdrsImpl.decode(buf);	2,287✔
65	}	2,287✔
66	return this._headers;	1,764✔
67	}	1,764✔
68
69	get data(): Uint8Array {	54✔
70	if (!this._rdata) {	×
71	return new Uint8Array(0);	×
72	}	×
73	return this._msg.hdr > -1	210,218,224✔
74	? this._rdata.subarray(this._msg.hdr)	210,218,224✔
75	: this._rdata;	210,218,224✔
76	}	210,218,224✔
77
78	// eslint-ignore-next-line @typescript-eslint/no-explicit-any
79	respond(	52✔
80	data: Payload = Empty,	52✔
81	opts?: { headers?: MsgHdrs; reply?: string },	52✔
82	): boolean {	52✔
83	if (this.reply) {	3,835✔
84	this.publisher.publish(this.reply, data, opts);	4,005✔
85	return true;	4,005✔
86	}	4,005!
87	return false;	213✔
88	}	208✔
89
90	size(): number {	51✔
91	const subj = this._msg.subject.length;	1,052✔
92	const reply = this._msg.reply?.length \|\| 0;	×
93	const payloadAndHeaders = this._msg.size === -1 ? 0 : this._msg.size;	×
94	return subj + reply + payloadAndHeaders;	1,052✔
95	}	1,052✔
96
97	json<T = unknown>(reviver?: ReviverFn): T {	53✔
98	return JSON.parse(this.string(), reviver);	153✔
99	}	153✔
100
101	string(): string {	54✔
102	return TD.decode(this.data);	222✔
103	}	222✔
104
105	requestInfo(): RequestInfo \| null {	33✔
106	const v = this.headers?.get("Nats-Request-Info");	34✔
107	if (v) {	34✔
108	return JSON.parse(	34✔
109	v,	34✔
110	function (this: unknown, key: string, value: unknown): unknown {	34✔
111	if ((key === "start" \|\| key === "stop") && value !== "") {	×
112	return new Date(Date.parse(value as string));	×
113	}	×
114	return value;	37✔
115	},	34✔
116	) as RequestInfo;
117	}	34!
118	return null;	×
119	}	×
120	}	55✔

nats-io / nats.js / 13091454488

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