Skip to content

Commit cadb6bc

Browse files
committed
[FEAT] [CORE] added the ability to publish/respond by providing a Msg argument. This is useful downstream to some applications.
1 parent 82e8fc5 commit cadb6bc

File tree

3 files changed

+41
-1
lines changed

3 files changed

+41
-1
lines changed

nats-base-client/core.ts

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -556,6 +556,13 @@ export interface NatsConnection {
556556
*/
557557
publishMessage(msg: Msg): void;
558558

559+
/**
560+
* Replies using the reply subject of the specified message, specifying the
561+
* data, headers in the message if any.
562+
* @param msg
563+
*/
564+
respondMessage(msg: Msg): boolean;
565+
559566
/**
560567
* Subscribe expresses interest in the specified subject. The subject may
561568
* have wildcards. Messages are delivered to the {@link SubOpts#callback |SubscriptionOptions callback}

nats-base-client/nats.ts

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -133,6 +133,17 @@ export class NatsConnectionImpl implements NatsConnection {
133133
});
134134
}
135135

136+
respondMessage(msg: Msg) {
137+
if (msg.reply) {
138+
this.publish(msg.reply, msg.data, {
139+
reply: msg.reply,
140+
headers: msg.headers,
141+
});
142+
return true;
143+
}
144+
return false;
145+
}
146+
136147
subscribe(
137148
subject: string,
138149
opts: SubscriptionOptions = {},

tests/basics_test.ts

Lines changed: 23 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1381,7 +1381,7 @@ Deno.test("basics - sync subscription", async () => {
13811381
await cleanup(ns, nc);
13821382
});
13831383

1384-
Deno.test("basics - respond message", async () => {
1384+
Deno.test("basics - publish message", async () => {
13851385
const { ns, nc } = await setup();
13861386
const sub = nc.subscribe("q");
13871387

@@ -1403,6 +1403,28 @@ Deno.test("basics - respond message", async () => {
14031403
await cleanup(ns, nc);
14041404
});
14051405

1406+
Deno.test("basics - respond message", async () => {
1407+
const { ns, nc } = await setup();
1408+
const sub = nc.subscribe("q");
1409+
1410+
const nis = new MM(nc);
1411+
nis.data = new TextEncoder().encode("not in service");
1412+
1413+
(async () => {
1414+
for await (const m of sub) {
1415+
if (m.reply) {
1416+
nis.reply = m.reply;
1417+
nc.respondMessage(nis);
1418+
}
1419+
}
1420+
})().then();
1421+
1422+
const r = await nc.request("q");
1423+
assertEquals(r.string(), "not in service");
1424+
1425+
await cleanup(ns, nc);
1426+
});
1427+
14061428
class MM implements Msg {
14071429
data!: Uint8Array;
14081430
sid: number;

0 commit comments

Comments
 (0)