10152999901

Committed 29 Jul 2024 10:46PM UTC coverage: 86.378% (-0.2%) from 86.538%

Build # 10152999901

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #721 from nats-io/fix-kv-sources

fix(kv): fixed an issue where kv sources were not properly processed

Run Details

1885 of 2251 branches covered (83.74%)

Branch coverage included in aggregate %.

8 of 11 new or added lines in 2 files covered. (72.73%)

27 existing lines in 3 files now uncovered.

11228 of 12930 relevant lines covered (86.84%)

870695.78 hits per line

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

90.77

/jetstream/jsbaseclient_api.ts

/*
 * Copyright 2021-2023 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 { Empty } from "../nats-base-client/encoders.ts";
import { Codec, JSONCodec } from "../nats-base-client/codec.ts";
import { backoff, delay, extend } from "../nats-base-client/util.ts";
import { NatsConnectionImpl } from "../nats-base-client/nats.ts";
import { checkJsErrorCode } from "./jsutil.ts";
import {
  ErrorCode,
  JetStreamOptions,
  Msg,
  NatsConnection,
  NatsError,
  RequestOptions,
} from "../nats-base-client/core.ts";
import { ApiResponse } from "./jsapi_types.ts";

const defaultPrefix = "$JS.API";
const defaultTimeout = 5000;

export function defaultJsOptions(opts?: JetStreamOptions): JetStreamOptions {
  opts = opts || {} as JetStreamOptions;
  if (opts.domain) {
    opts.apiPrefix = `$JS.${opts.domain}.API`;
    delete opts.domain;
  }
  return extend({ apiPrefix: defaultPrefix, timeout: defaultTimeout }, opts);
}

export interface StreamNames {
  streams: string[];
}

export interface StreamNameBySubject {
  subject: string;
}

export class BaseApiClient {
  nc: NatsConnectionImpl;
  opts: JetStreamOptions;
  prefix: string;
  timeout: number;
  jc: Codec<unknown>;

  constructor(nc: NatsConnection, opts?: JetStreamOptions) {
    this.nc = nc as NatsConnectionImpl;
    this.opts = defaultJsOptions(opts);
    this._parseOpts();
    this.prefix = this.opts.apiPrefix!;
    this.timeout = this.opts.timeout!;
    this.jc = JSONCodec();
  }

  getOptions(): JetStreamOptions {
    return Object.assign({}, this.opts);
  }

  _parseOpts() {
    let prefix = this.opts.apiPrefix;
    if (!prefix || prefix.length === 0) {
      throw new Error("invalid empty prefix");
    }
    const c = prefix[prefix.length - 1];
    if (c === ".") {
      prefix = prefix.substr(0, prefix.length - 1);
    }
    this.opts.apiPrefix = prefix;
  }

  async _request(
    subj: string,
    data: unknown = null,
    opts?: Partial<RequestOptions> & { retries?: number },
  ): Promise<unknown> {
    opts = opts || {} as RequestOptions;
    opts.timeout = this.timeout;

    let a: Uint8Array = Empty;
    if (data) {
      a = this.jc.encode(data);
    }

    let { retries } = opts as {
      retries: number;
    };

    retries = retries || 1;
    retries = retries === -1 ? Number.MAX_SAFE_INTEGER : retries;
    const bo = backoff();

    for (let i = 0; i < retries; i++) {
      try {
        const m = await this.nc.request(
          subj,
          a,
          opts as RequestOptions,
        );
        return this.parseJsResponse(m);
      } catch (err) {
        const ne = err as NatsError;
        if (
          (ne.code === "503" || ne.code === ErrorCode.Timeout) &&
          i + 1 < retries
        ) {
          await delay(bo.backoff(i));
        } else {
          throw err;
        }
      }
    }
  }

  async findStream(subject: string): Promise<string> {
    const q = { subject } as StreamNameBySubject;
    const r = await this._request(`${this.prefix}.STREAM.NAMES`, q);
    const names = r as StreamNames;
    if (!names.streams || names.streams.length !== 1) {
      throw new Error("no stream matches subject");
    }
    return names.streams[0];
  }

  getConnection(): NatsConnection {
    return this.nc;
  }

  parseJsResponse(m: Msg): unknown {
    const v = this.jc.decode(m.data);
    const r = v as ApiResponse;
    if (r.error) {
      const err = checkJsErrorCode(r.error.code, r.error.description);
      if (err !== null) {
        err.api_error = r.error;
        throw err;
      }
    }
    return v;
  }
}

1	/*	1✔
2	* Copyright 2021-2023 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	*/	53✔
15
16	import { Empty } from "../nats-base-client/encoders.ts";	53✔
17	import { Codec, JSONCodec } from "../nats-base-client/codec.ts";	53✔
18	import { backoff, delay, extend } from "../nats-base-client/util.ts";	53✔
19	import { NatsConnectionImpl } from "../nats-base-client/nats.ts";
20	import { checkJsErrorCode } from "./jsutil.ts";	53✔
21	import {	53✔
22	ErrorCode,	53✔
23	JetStreamOptions,
24	Msg,
25	NatsConnection,
26	NatsError,
27	RequestOptions,
28	} from "../nats-base-client/core.ts";	53✔
29	import { ApiResponse } from "./jsapi_types.ts";
30
31	const defaultPrefix = "$JS.API";	53✔
32	const defaultTimeout = 5000;	53✔
33
34	export function defaultJsOptions(opts?: JetStreamOptions): JetStreamOptions {	53✔
35	opts = opts \|\| {} as JetStreamOptions;	4,058✔
36	if (opts.domain) {	4,058✔
37	opts.apiPrefix = `$JS.${opts.domain}.API`;	4,065✔
38	delete opts.domain;	4,065✔
39	}	4,065✔
40	return extend({ apiPrefix: defaultPrefix, timeout: defaultTimeout }, opts);	16,232✔
41	}	4,058✔
42
43	export interface StreamNames {
44	streams: string[];
45	}
46
47	export interface StreamNameBySubject {
48	subject: string;
49	}
50
51	export class BaseApiClient {	53✔
52	nc: NatsConnectionImpl;	4,049✔
53	opts: JetStreamOptions;	4,049✔
54	prefix: string;	4,049✔
55	timeout: number;	4,049✔
56	jc: Codec<unknown>;	4,049✔
57
58	constructor(nc: NatsConnection, opts?: JetStreamOptions) {	4,049✔
59	this.nc = nc as NatsConnectionImpl;	8,045✔
60	this.opts = defaultJsOptions(opts);	8,045✔
61	this._parseOpts();	8,045✔
62	this.prefix = this.opts.apiPrefix!;	8,045✔
63	this.timeout = this.opts.timeout!;	8,045✔
64	this.jc = JSONCodec();	8,045✔
65	}	8,045✔
66
67	getOptions(): JetStreamOptions {	4,049✔
68	return Object.assign({}, this.opts);	4,103✔
69	}	4,103✔
70
71	_parseOpts() {	4,049✔
72	let prefix = this.opts.apiPrefix;	8,045✔
73	if (!prefix \|\| prefix.length === 0) {	8,045✔
74	throw new Error("invalid empty prefix");	8,048✔
75	}	8,048✔
76	const c = prefix[prefix.length - 1];	12,038✔
77	if (c === ".") {	8,045✔
78	prefix = prefix.substr(0, prefix.length - 1);	8,054✔
79	}	8,054✔
80	this.opts.apiPrefix = prefix;	12,038✔
81	}	8,045✔
82
83	async _request(	4,049✔
84	subj: string,	4,049✔
85	data: unknown = null,	4,049✔
86	opts?: Partial<RequestOptions> & { retries?: number },	4,049✔
87	): Promise<unknown> {	4,049✔
88	opts = opts \|\| {} as RequestOptions;	7,108✔
89	opts.timeout = this.timeout;	7,108✔
90
91	let a: Uint8Array = Empty;	7,108✔
92	if (data) {	7,108✔
93	a = this.jc.encode(data);	8,865✔
94	}	8,865✔
95
96	let { retries } = opts as {	7,108✔
97	retries: number;
98	};
99
100	retries = retries \|\| 1;	7,108✔
101	retries = retries === -1 ? Number.MAX_SAFE_INTEGER : retries;	7,108✔
102	const bo = backoff();	7,108✔
103
104	for (let i = 0; i < retries; i++) {	7,108✔
105	try {	7,108✔
106	const m = await this.nc.request(	7,108✔
107	subj,	7,108✔
108	a,	7,108✔
109	opts as RequestOptions,	7,108✔
110	);
111	return this.parseJsResponse(m);	10,164✔
112	} catch (err) {	7,108✔
113	const ne = err as NatsError;	7,398✔
UNCOV 114	if (	×
UNCOV 115	(ne.code === "503" \|\| ne.code === ErrorCode.Timeout) &&	✔
UNCOV 116	i + 1 < retries	×
UNCOV 117	) {	×
UNCOV 118	await delay(bo.backoff(i));	×
UNCOV 119	} else {	×
120	throw err;	7,398✔
121	}	7,398✔
122	}	7,398✔
123	}	7,108!
124	}	×
125
126	async findStream(subject: string): Promise<string> {	4,049✔
127	const q = { subject } as StreamNameBySubject;	12,450✔
128	const r = await this._request(`${this.prefix}.STREAM.NAMES`, q);	4,150✔
129	const names = r as StreamNames;	4,150✔
130	if (!names.streams \|\| names.streams.length !== 1) {	4,150✔
131	throw new Error("no stream matches subject");	4,153✔
132	}	4,153✔
133	return names.streams[0];	4,248✔
134	}	4,150✔
135
136	getConnection(): NatsConnection {	×
137	return this.nc;	×
138	}	×
139
140	parseJsResponse(m: Msg): unknown {	4,049✔
141	const v = this.jc.decode(m.data);	129,947✔
142	const r = v as ApiResponse;	129,947✔
143	if (r.error) {	129,947✔
144	const err = checkJsErrorCode(r.error.code, r.error.description);	130,258✔
145	if (err !== null) {	130,258✔
146	err.api_error = r.error;	130,258✔
147	throw err;	130,258✔
148	}	130,258✔
149	}	130,258✔
150	return v;	255,534✔
151	}	129,947✔
152	}	4,049✔

nats-io / nats.deno / 10152999901

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