mirror of
https://github.com/hanzomsg/nats.js.git
synced 2026-10-30 20:31:36 +00:00
298 lines
78 KiB
HTML
298 lines
78 KiB
HTML
<!DOCTYPE html><html class="default" lang="en" data-base="./"><head><meta charset="utf-8"/><meta http-equiv="x-ua-compatible" content="IE=edge"/><title>@nats-io/jetstream</title><meta name="description" content="Documentation for @nats-io/jetstream"/><meta name="viewport" content="width=device-width, initial-scale=1"/><link rel="stylesheet" href="assets/style.css"/><link rel="stylesheet" href="assets/highlight.css"/><script defer src="assets/main.js"></script><script async src="assets/icons.js" id="tsd-icons-script"></script><script async src="assets/search.js" id="tsd-search-script"></script><script async src="assets/navigation.js" id="tsd-nav-script"></script></head><body><script>document.documentElement.dataset.theme = localStorage.getItem("tsd-theme") || "os";document.body.style.display="none";setTimeout(() => window.app?app.showPage():document.body.style.removeProperty("display"),500)</script><header class="tsd-page-toolbar"><div class="tsd-toolbar-contents container"><a href="index.html" class="title">@nats-io/jetstream</a><div id="tsd-toolbar-links"></div><button id="tsd-search-trigger" class="tsd-widget" aria-label="Search"><svg width="16" height="16" viewBox="0 0 16 16" fill="none" aria-hidden="true"><use href="assets/icons.svg#icon-search"></use></svg></button><dialog id="tsd-search" aria-label="Search"><input role="combobox" id="tsd-search-input" aria-controls="tsd-search-results" aria-autocomplete="list" aria-expanded="true" autocapitalize="off" autocomplete="off" placeholder="Search the docs" maxLength="100"/><ul role="listbox" id="tsd-search-results"></ul><div id="tsd-search-status" aria-live="polite" aria-atomic="true"><div>Preparing search index...</div></div></dialog><a href="#" class="tsd-widget menu" id="tsd-toolbar-menu-trigger" data-toggle="menu" aria-label="Menu"><svg width="16" height="16" viewBox="0 0 16 16" fill="none" aria-hidden="true"><use href="assets/icons.svg#icon-menu"></use></svg></a></div></header><div class="container container-main"><div class="col-content"><div class="tsd-page-title"><h1>@nats-io/jetstream</h1></div><div class="tsd-panel tsd-typography"><p><a href="./LICENSE"><img src="https://img.shields.io/badge/Licence-Apache%202.0-blue.svg" alt="License"></a>
|
|
<img src="https://github.com/nats-io/nats.js/actions/workflows/test.yml/badge.svg" alt="jetstream">
|
|
<a href="https://nats-io.github.io/nats.js/jetstream/index.html"><img src="https://img.shields.io/badge/JSDoc-reference-blue" alt="JSDoc"></a></p>
|
|
<p><a href="https://jsr.io/@nats-io/jetstream"><img src="https://jsr.io/badges/@nats-io/jetstream" alt="JSR"></a>
|
|
<a href="https://jsr.io/@nats-io/jetstream"><img src="https://jsr.io/badges/@nats-io/jetstream/score" alt="JSR"></a></p>
|
|
<p><a href="https://www.npmjs.com/package/@nats-io/jetstream"><img src="https://img.shields.io/npm/v/%40nats-io%2Fjetstream" alt="NPM Version"></a>
|
|
<img src="https://img.shields.io/npm/dt/%40nats-io%2Fjetstream" alt="NPM Downloads">
|
|
<img src="https://img.shields.io/npm/dm/%40nats-io%2Fjetstream" alt="NPM Downloads"></p>
|
|
<h1 id="jetstream" class="tsd-anchor-link">JetStream<a href="#jetstream" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h1><p>The jetstream module implements the JetStream protocol functionality for
|
|
JavaScript clients. JetStream is the NATS persistence engine providing
|
|
streaming, message, and worker queues with At-Least-Once semantics.</p>
|
|
<p>To use JetStream simply install this library, and create a <code>jetstream(nc)</code> or
|
|
<code>jetstreamManager(nc)</code> with a connection provided by your chosen transport
|
|
module. JetStreamManager allows you to interact with the NATS server to manage
|
|
JetStream resources. The JetStream client allows you to interact with JetStream
|
|
resources.</p>
|
|
<h2 id="installation" class="tsd-anchor-link">Installation<a href="#installation" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h2><p>For a quick overview of the libraries and how to install them, see
|
|
<a href="media/runtimes.md">runtimes.md</a>.</p>
|
|
<p>Note that this library is distributed in two different registries:</p>
|
|
<ul>
|
|
<li>npm a node-specific library supporting CJS (<code>require</code>) and ESM (<code>import</code>)</li>
|
|
<li>jsr a node and other ESM (<code>import</code>) compatible runtimes (deno, browser, node)</li>
|
|
</ul>
|
|
<p>If your application doesn't use <code>require</code>, you can simply depend on the JSR
|
|
version.</p>
|
|
<h3 id="npm" class="tsd-anchor-link">NPM<a href="#npm" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h3><p>The NPM registry hosts a node-only compatible version of the library
|
|
<a href="https://www.npmjs.com/package/@nats-io/jetstream">@nats-io/jetstream</a>
|
|
supporting both CJS and ESM:</p>
|
|
<pre><code class="bash"><span class="hl-0">npm</span><span class="hl-1"> </span><span class="hl-2">install</span><span class="hl-1"> </span><span class="hl-2">@nats-io/jetstream</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<h3 id="jsr" class="tsd-anchor-link">JSR<a href="#jsr" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h3><p>The JSR registry hosts the ESM-only
|
|
<a href="https://jsr.io/@nats-io/jetstream">@nats-io/jetstream</a> version of the library.</p>
|
|
<pre><code class="bash"><span class="hl-0">deno</span><span class="hl-1"> </span><span class="hl-2">add</span><span class="hl-1"> </span><span class="hl-2">jsr:@nats-io/jetstream</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<pre><code class="bash"><span class="hl-0">npx</span><span class="hl-1"> </span><span class="hl-2">jsr</span><span class="hl-1"> </span><span class="hl-2">add</span><span class="hl-1"> </span><span class="hl-2">@nats-io/jetstream</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<pre><code class="bash"><span class="hl-0">yarn</span><span class="hl-1"> </span><span class="hl-2">dlx</span><span class="hl-1"> </span><span class="hl-2">jsr</span><span class="hl-1"> </span><span class="hl-2">add</span><span class="hl-1"> </span><span class="hl-2">@nats-io/jetstream</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<pre><code class="bash"><span class="hl-0">bunx</span><span class="hl-1"> </span><span class="hl-2">jsr</span><span class="hl-1"> </span><span class="hl-2">add</span><span class="hl-1"> </span><span class="hl-2">@nats-io/jetstream</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<h2 id="referencing-the-library" class="tsd-anchor-link">Referencing the library<a href="#referencing-the-library" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h2><p>Once you import the library, you can reference in your code as:</p>
|
|
<pre><code class="javascript"><span class="hl-3">import</span><span class="hl-1"> { </span><span class="hl-4">jetstream</span><span class="hl-1">, </span><span class="hl-4">jetstreamManager</span><span class="hl-1"> } </span><span class="hl-3">from</span><span class="hl-1"> </span><span class="hl-2">"@nats-io/jetstream"</span><span class="hl-1">;</span><br/><br/><span class="hl-5">// or in node (only when using CJS)</span><br/><span class="hl-6">const</span><span class="hl-1"> { </span><span class="hl-7">jetstream</span><span class="hl-1">, </span><span class="hl-7">jetstreamManager</span><span class="hl-1"> } = </span><span class="hl-0">require</span><span class="hl-1">(</span><span class="hl-2">"@nats-io/jetstream"</span><span class="hl-1">);</span><br/><br/><span class="hl-5">// using a nats connection:</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">js</span><span class="hl-1"> = </span><span class="hl-0">jetstream</span><span class="hl-1">(</span><span class="hl-4">nc</span><span class="hl-1">);</span><br/><span class="hl-5">// and/or</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">jsm</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-0">jetstreamManager</span><span class="hl-1">(</span><span class="hl-4">nc</span><span class="hl-1">);</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<h1 id="jetstream-1" class="tsd-anchor-link">JetStream<a href="#jetstream-1" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h1><p>JetStream is the NATS persistence engine providing streaming, message, and
|
|
worker queues with At-Least-Once semantics. JetStream stores messages in
|
|
<em>streams</em>. A stream defines how messages are stored and limits such as how long
|
|
they persist or how many to keep. To store a message in JetStream, you simply
|
|
need to publish to a subject that is associated with a stream.</p>
|
|
<p>Messages are replayed from a stream by <em>consumers</em>. A consumer configuration
|
|
specifies which messages should be presented. For example a consumer may only be
|
|
interested in viewing messages from a specific sequence or starting from a
|
|
specific time, or having a specific subject. The configuration also specifies if
|
|
the server should require messages to be acknowledged and how long to wait for
|
|
acknowledgements. The consumer configuration also specifies options to control
|
|
the rate at which messages are presented to the client.</p>
|
|
<p>For more information about JetStream, please visit the
|
|
<a href="https://docs.nats.io/nats-concepts/jetstream">JetStream docs</a>.</p>
|
|
<h2 id="migration" class="tsd-anchor-link">Migration<a href="#migration" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h2><p>If you were using an embedded version of JetStream as provided by the npm
|
|
nats@^2.0.0 or nats.deno or <a href="http://nats.ws">nats.ws</a> libraries, you will have to import this
|
|
library and replace your usages of <code>NatsConnection#jetstream()</code> or
|
|
<code>NatsConnection#jetstreamManager()</code> with <code>jetstream(nc)</code> or
|
|
<code>await jetstreamManager(nc)</code> where you pass your actual connection to the above
|
|
functions.</p>
|
|
<p>Also note that if you are using <a href="media/README.md">KV</a> or
|
|
<a href="media/README-1.md">ObjectStore</a>, these APIs are now provided by a different
|
|
libraries <code>@nats-io/kv</code> and <code>@nats-io/obj</code> respectively. If you are only using
|
|
KV or ObjectStore, there's no need to reference this library directly unless you
|
|
need to do some specific JetStreamManager API, as both <code>@nats-io/kv</code> and
|
|
<code>@nats-io/obj</code> depend on this library already and use it under the hood.</p>
|
|
<h2 id="jetstreammanager-jsm" class="tsd-anchor-link">JetStreamManager (JSM)<a href="#jetstreammanager-jsm" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h2><p>The JetStreamManager provides CRUD functionality to manage streams and consumers
|
|
resources. To access a JetStream manager:</p>
|
|
<pre><code class="typescript"><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">jsm</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-0">jetstreamManager</span><span class="hl-1">(</span><span class="hl-4">nc</span><span class="hl-1">);</span><br/><br/><span class="hl-3">for</span><span class="hl-1"> </span><span class="hl-3">await</span><span class="hl-1"> (</span><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">si</span><span class="hl-1"> </span><span class="hl-6">of</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">streams</span><span class="hl-1">.</span><span class="hl-0">list</span><span class="hl-1">()) {</span><br/><span class="hl-1"> </span><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-4">si</span><span class="hl-1">);</span><br/><span class="hl-1">}</span><br/><br/><span class="hl-5">// add a stream - jetstream can capture nats core messages</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">stream</span><span class="hl-1"> = </span><span class="hl-2">"mystream"</span><span class="hl-1">;</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">subj</span><span class="hl-1"> = </span><span class="hl-2">`mystream.*`</span><span class="hl-1">;</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">streams</span><span class="hl-1">.</span><span class="hl-0">add</span><span class="hl-1">({ </span><span class="hl-4">name:</span><span class="hl-1"> </span><span class="hl-4">stream</span><span class="hl-1">, </span><span class="hl-4">subjects:</span><span class="hl-1"> [</span><span class="hl-4">subj</span><span class="hl-1">] });</span><br/><br/><span class="hl-3">for</span><span class="hl-1"> (</span><span class="hl-6">let</span><span class="hl-1"> </span><span class="hl-4">i</span><span class="hl-1"> = </span><span class="hl-8">0</span><span class="hl-1">; </span><span class="hl-4">i</span><span class="hl-1"> < </span><span class="hl-8">100</span><span class="hl-1">; </span><span class="hl-4">i</span><span class="hl-1">++) {</span><br/><span class="hl-1"> </span><span class="hl-4">nc</span><span class="hl-1">.</span><span class="hl-0">publish</span><span class="hl-1">(</span><span class="hl-2">`</span><span class="hl-6">${</span><span class="hl-4">subj</span><span class="hl-6">}</span><span class="hl-2">.a`</span><span class="hl-1">, </span><span class="hl-4">Empty</span><span class="hl-1">);</span><br/><span class="hl-1">}</span><br/><br/><span class="hl-5">// find a stream that stores a specific subject:</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">name</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">streams</span><span class="hl-1">.</span><span class="hl-0">find</span><span class="hl-1">(</span><span class="hl-2">"mystream.A"</span><span class="hl-1">);</span><br/><br/><span class="hl-5">// retrieve info about the stream by its name</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">si</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">streams</span><span class="hl-1">.</span><span class="hl-0">info</span><span class="hl-1">(</span><span class="hl-4">name</span><span class="hl-1">);</span><br/><br/><span class="hl-5">// update a stream configuration</span><br/><span class="hl-4">si</span><span class="hl-1">.</span><span class="hl-4">config</span><span class="hl-1">.</span><span class="hl-4">subjects</span><span class="hl-1">?.</span><span class="hl-0">push</span><span class="hl-1">(</span><span class="hl-2">"a.b"</span><span class="hl-1">);</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">streams</span><span class="hl-1">.</span><span class="hl-0">update</span><span class="hl-1">(</span><span class="hl-4">si</span><span class="hl-1">.</span><span class="hl-4">config</span><span class="hl-1">);</span><br/><br/><span class="hl-5">// get a particular stored message in the stream by sequence</span><br/><span class="hl-5">// this is not associated with a consumer</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">sm</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">streams</span><span class="hl-1">.</span><span class="hl-0">getMessage</span><span class="hl-1">(</span><span class="hl-4">stream</span><span class="hl-1">, { </span><span class="hl-4">seq:</span><span class="hl-1"> </span><span class="hl-8">1</span><span class="hl-1"> });</span><br/><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-4">sm</span><span class="hl-1">.</span><span class="hl-4">seq</span><span class="hl-1">);</span><br/><br/><span class="hl-5">// delete the 5th message in the stream, securely erasing it</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">streams</span><span class="hl-1">.</span><span class="hl-0">deleteMessage</span><span class="hl-1">(</span><span class="hl-4">stream</span><span class="hl-1">, </span><span class="hl-8">5</span><span class="hl-1">);</span><br/><br/><span class="hl-5">// purge all messages in the stream, the stream itself remains.</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">streams</span><span class="hl-1">.</span><span class="hl-0">purge</span><span class="hl-1">(</span><span class="hl-4">stream</span><span class="hl-1">);</span><br/><br/><span class="hl-5">// purge all messages with a specific subject (filter can be a wildcard)</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">streams</span><span class="hl-1">.</span><span class="hl-0">purge</span><span class="hl-1">(</span><span class="hl-4">stream</span><span class="hl-1">, { </span><span class="hl-4">filter:</span><span class="hl-1"> </span><span class="hl-2">"a.b"</span><span class="hl-1"> });</span><br/><br/><span class="hl-5">// purge messages with a specific subject keeping some messages</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">streams</span><span class="hl-1">.</span><span class="hl-0">purge</span><span class="hl-1">(</span><span class="hl-4">stream</span><span class="hl-1">, { </span><span class="hl-4">filter:</span><span class="hl-1"> </span><span class="hl-2">"a.c"</span><span class="hl-1">, </span><span class="hl-4">keep:</span><span class="hl-1"> </span><span class="hl-8">5</span><span class="hl-1"> });</span><br/><br/><span class="hl-5">// purge all messages with upto (not including seq)</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">streams</span><span class="hl-1">.</span><span class="hl-0">purge</span><span class="hl-1">(</span><span class="hl-4">stream</span><span class="hl-1">, { </span><span class="hl-4">seq:</span><span class="hl-1"> </span><span class="hl-8">90</span><span class="hl-1"> });</span><br/><br/><span class="hl-5">// purge all messages with upto sequence that have a matching subject</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">streams</span><span class="hl-1">.</span><span class="hl-0">purge</span><span class="hl-1">(</span><span class="hl-4">stream</span><span class="hl-1">, { </span><span class="hl-4">filter:</span><span class="hl-1"> </span><span class="hl-2">"a.d"</span><span class="hl-1">, </span><span class="hl-4">seq:</span><span class="hl-1"> </span><span class="hl-8">100</span><span class="hl-1"> });</span><br/><br/><span class="hl-5">// list all consumers for a stream:</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">consumers</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">consumers</span><span class="hl-1">.</span><span class="hl-0">list</span><span class="hl-1">(</span><span class="hl-4">stream</span><span class="hl-1">).</span><span class="hl-0">next</span><span class="hl-1">();</span><br/><span class="hl-4">consumers</span><span class="hl-1">.</span><span class="hl-0">forEach</span><span class="hl-1">((</span><span class="hl-4">ci</span><span class="hl-1">) </span><span class="hl-6">=></span><span class="hl-1"> {</span><br/><span class="hl-1"> </span><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-4">ci</span><span class="hl-1">);</span><br/><span class="hl-1">});</span><br/><br/><span class="hl-5">// add a new durable consumer</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">consumers</span><span class="hl-1">.</span><span class="hl-0">add</span><span class="hl-1">(</span><span class="hl-4">stream</span><span class="hl-1">, {</span><br/><span class="hl-1"> </span><span class="hl-4">durable_name:</span><span class="hl-1"> </span><span class="hl-2">"me"</span><span class="hl-1">,</span><br/><span class="hl-1"> </span><span class="hl-4">ack_policy:</span><span class="hl-1"> </span><span class="hl-4">AckPolicy</span><span class="hl-1">.</span><span class="hl-4">Explicit</span><span class="hl-1">,</span><br/><span class="hl-1">});</span><br/><br/><span class="hl-5">// retrieve a consumer's status and configuration</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">ci</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">consumers</span><span class="hl-1">.</span><span class="hl-0">info</span><span class="hl-1">(</span><span class="hl-4">stream</span><span class="hl-1">, </span><span class="hl-2">"me"</span><span class="hl-1">);</span><br/><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-4">ci</span><span class="hl-1">);</span><br/><br/><span class="hl-5">// delete a particular consumer</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">consumers</span><span class="hl-1">.</span><span class="hl-0">delete</span><span class="hl-1">(</span><span class="hl-4">stream</span><span class="hl-1">, </span><span class="hl-2">"me"</span><span class="hl-1">);</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<h2 id="jetstream-client" class="tsd-anchor-link">JetStream Client<a href="#jetstream-client" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h2><p>The JetStream client presents an API for adding messages to a stream or
|
|
processing messages stored in a stream.</p>
|
|
<pre><code class="typescript"><span class="hl-5">// create the stream</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">jsm</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-0">jetstreamManager</span><span class="hl-1">(</span><span class="hl-4">nc</span><span class="hl-1">);</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">jsm</span><span class="hl-1">.</span><span class="hl-4">streams</span><span class="hl-1">.</span><span class="hl-0">add</span><span class="hl-1">({ </span><span class="hl-4">name:</span><span class="hl-1"> </span><span class="hl-2">"a"</span><span class="hl-1">, </span><span class="hl-4">subjects:</span><span class="hl-1"> [</span><span class="hl-2">"a.*"</span><span class="hl-1">] });</span><br/><br/><span class="hl-5">// create a jetstream client:</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">js</span><span class="hl-1"> = </span><span class="hl-0">jetstream</span><span class="hl-1">(</span><span class="hl-4">nc</span><span class="hl-1">);</span><br/><br/><span class="hl-5">// publish a message received by a stream</span><br/><span class="hl-6">let</span><span class="hl-1"> </span><span class="hl-4">pa</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">js</span><span class="hl-1">.</span><span class="hl-0">publish</span><span class="hl-1">(</span><span class="hl-2">"a.b"</span><span class="hl-1">);</span><br/><span class="hl-5">// jetstream returns an acknowledgement with the</span><br/><span class="hl-5">// stream that captured the message, it's assigned sequence</span><br/><span class="hl-5">// and whether the message is a duplicate.</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">stream</span><span class="hl-1"> = </span><span class="hl-4">pa</span><span class="hl-1">.</span><span class="hl-4">stream</span><span class="hl-1">;</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">seq</span><span class="hl-1"> = </span><span class="hl-4">pa</span><span class="hl-1">.</span><span class="hl-4">seq</span><span class="hl-1">;</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">duplicate</span><span class="hl-1"> = </span><span class="hl-4">pa</span><span class="hl-1">.</span><span class="hl-4">duplicate</span><span class="hl-1">;</span><br/><br/><span class="hl-5">// More interesting is the ability to prevent duplicates</span><br/><span class="hl-5">// on messages that are stored in the server. If</span><br/><span class="hl-5">// you assign a message ID, the server will keep looking</span><br/><span class="hl-5">// for the same ID for a configured amount of time (within a</span><br/><span class="hl-5">// configurable time window), and reject messages that</span><br/><span class="hl-5">// have the same ID:</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">js</span><span class="hl-1">.</span><span class="hl-0">publish</span><span class="hl-1">(</span><span class="hl-2">"a.b"</span><span class="hl-1">, </span><span class="hl-4">Empty</span><span class="hl-1">, { </span><span class="hl-4">msgID:</span><span class="hl-1"> </span><span class="hl-2">"a"</span><span class="hl-1"> });</span><br/><br/><span class="hl-5">// you can also specify constraints that should be satisfied.</span><br/><span class="hl-5">// For example, you can request the message to have as its</span><br/><span class="hl-5">// last sequence before accepting the new message:</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">js</span><span class="hl-1">.</span><span class="hl-0">publish</span><span class="hl-1">(</span><span class="hl-2">"a.b"</span><span class="hl-1">, </span><span class="hl-4">Empty</span><span class="hl-1">, { </span><span class="hl-4">expect:</span><span class="hl-1"> { </span><span class="hl-4">lastMsgID:</span><span class="hl-1"> </span><span class="hl-2">"a"</span><span class="hl-1"> } });</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">js</span><span class="hl-1">.</span><span class="hl-0">publish</span><span class="hl-1">(</span><span class="hl-2">"a.b"</span><span class="hl-1">, </span><span class="hl-4">Empty</span><span class="hl-1">, { </span><span class="hl-4">expect:</span><span class="hl-1"> { </span><span class="hl-4">lastSequence:</span><span class="hl-1"> </span><span class="hl-8">3</span><span class="hl-1"> } });</span><br/><span class="hl-5">// save the last sequence for this publish</span><br/><span class="hl-4">pa</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">js</span><span class="hl-1">.</span><span class="hl-0">publish</span><span class="hl-1">(</span><span class="hl-2">"a.b"</span><span class="hl-1">, </span><span class="hl-4">Empty</span><span class="hl-1">, { </span><span class="hl-4">expect:</span><span class="hl-1"> { </span><span class="hl-4">streamName:</span><span class="hl-1"> </span><span class="hl-2">"a"</span><span class="hl-1"> } });</span><br/><span class="hl-5">// you can also mix the above combinations</span><br/><br/><span class="hl-5">// this stream here accepts wildcards, you can assert that the</span><br/><span class="hl-5">// last message sequence recorded on a particular subject matches:</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">buf</span><span class="hl-1">: </span><span class="hl-9">Promise</span><span class="hl-1"><</span><span class="hl-9">PubAck</span><span class="hl-1">>[] = [];</span><br/><span class="hl-3">for</span><span class="hl-1"> (</span><span class="hl-6">let</span><span class="hl-1"> </span><span class="hl-4">i</span><span class="hl-1"> = </span><span class="hl-8">0</span><span class="hl-1">; </span><span class="hl-4">i</span><span class="hl-1"> < </span><span class="hl-8">100</span><span class="hl-1">; </span><span class="hl-4">i</span><span class="hl-1">++) {</span><br/><span class="hl-1"> </span><span class="hl-4">buf</span><span class="hl-1">.</span><span class="hl-0">push</span><span class="hl-1">(</span><span class="hl-4">js</span><span class="hl-1">.</span><span class="hl-0">publish</span><span class="hl-1">(</span><span class="hl-2">"a.a"</span><span class="hl-1">, </span><span class="hl-4">Empty</span><span class="hl-1">));</span><br/><span class="hl-1">}</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-9">Promise</span><span class="hl-1">.</span><span class="hl-0">all</span><span class="hl-1">(</span><span class="hl-4">buf</span><span class="hl-1">);</span><br/><span class="hl-5">// if additional "a.b" has been recorded, this will fail</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">js</span><span class="hl-1">.</span><span class="hl-0">publish</span><span class="hl-1">(</span><span class="hl-2">"a.b"</span><span class="hl-1">, </span><span class="hl-4">Empty</span><span class="hl-1">, { </span><span class="hl-4">expect:</span><span class="hl-1"> { </span><span class="hl-4">lastSubjectSequence:</span><span class="hl-1"> </span><span class="hl-4">pa</span><span class="hl-1">.</span><span class="hl-4">seq</span><span class="hl-1"> } });</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<h3 id="processing-messages" class="tsd-anchor-link">Processing Messages<a href="#processing-messages" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h3><p>The JetStream API provides different mechanisms for retrieving messages. Each
|
|
mechanism offers a different "buffering strategy" that provides advantages that
|
|
map to how your application works and processes messages.</p>
|
|
<h4 id="basics" class="tsd-anchor-link">Basics<a href="#basics" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h4><p>To understand how these strategies differentiate, let's review some aspects of
|
|
processing a stream which will help you choose and design a strategy that works
|
|
for your application.</p>
|
|
<p>First and foremost, processing a stream is different from processing NATS core
|
|
messages:</p>
|
|
<p>In NATS core, you are presented with a message whenever a message is published
|
|
to a subject that you have subscribed to. When you process a stream you can
|
|
filter messages found on a stream to those matching subjects that interest you,
|
|
but the rate of delivery can be much higher, as the stream could store many more
|
|
messages that match your consumer than you would normally receive in a core NATS
|
|
subscription. When processing a stream, you can simulate the original rate at
|
|
which messages were ingested, but typically messages are processed "as fast as
|
|
possible". This means that a client could be overwhelmed by the number of
|
|
messages presented by the server.</p>
|
|
<p>In NATS core, if you want to ensure that your message was received as intended,
|
|
you publish a request. The receiving client can then respond an acknowledgement
|
|
or return some result. When processing a stream, the consumer configuration
|
|
dictates whether messages sent to the consumer should be acknowledged or not.
|
|
The server tracks acknowledged messages and knows which messages the consumer
|
|
has not seen or that may need to be resent due to a missing acknowledgement. By
|
|
default, clients have 30 seconds to respond or extend the acknowledgement. If a
|
|
message fails to be acknowledged in time, the server will resend the message
|
|
again. This functionality has a very important implications. Consumers should
|
|
not buffer more messages than they can process and acknowledge within the
|
|
acknowledgement window.</p>
|
|
<p>Lastly, the NATS server protects itself and when it detects that a client
|
|
connection is not draining data quickly enough, it disconnects it to prevent the
|
|
degradation from impacting other clients.</p>
|
|
<p>Given these competing conditions, the JetStream APIs allow a client to express
|
|
not only the buffering strategy for reading a stream at a pace the client can
|
|
sustain, but also how the reading happens.</p>
|
|
<p>JetStream allows the client to:</p>
|
|
<ul>
|
|
<li>Request the next message (one message at time)</li>
|
|
<li>Request the next N messages</li>
|
|
<li>Request and maintain a buffer of N messages</li>
|
|
</ul>
|
|
<p>The first two options allow the client to control and manage its buffering
|
|
manually, when the client is done processing the messages, it can at its
|
|
discretion to request additional messages or not.</p>
|
|
<p>The last option auto buffers messages for the client controlling the message and
|
|
data rates. The client specifies how many messages it wants to receive, and as
|
|
it consumes them, the library requests additional messages in an effort to
|
|
prevent the consumer from stalling, and thus maximize performance.</p>
|
|
<h4 id="retrieving-the-consumer" class="tsd-anchor-link">Retrieving the Consumer<a href="#retrieving-the-consumer" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h4><p>Before messages can be read from a stream, the consumer should have been created
|
|
using the JSM APIs as shown above. After the consumer exists, the client simply
|
|
retrieves it:</p>
|
|
<pre><code class="typescript"><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">stream</span><span class="hl-1"> = </span><span class="hl-2">"a"</span><span class="hl-1">;</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">consumer</span><span class="hl-1"> = </span><span class="hl-2">"a"</span><span class="hl-1">;</span><br/><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">js</span><span class="hl-1"> = </span><span class="hl-0">jetstream</span><span class="hl-1">(</span><span class="hl-4">nc</span><span class="hl-1">);</span><br/><br/><span class="hl-5">// retrieve an existing consumer</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">c</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">js</span><span class="hl-1">.</span><span class="hl-4">consumers</span><span class="hl-1">.</span><span class="hl-0">get</span><span class="hl-1">(</span><span class="hl-4">stream</span><span class="hl-1">, </span><span class="hl-4">consumer</span><span class="hl-1">);</span><br/><br/><span class="hl-5">// getting an ordered consumer requires no name</span><br/><span class="hl-5">// as the library will create it</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">oc</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">js</span><span class="hl-1">.</span><span class="hl-4">consumers</span><span class="hl-1">.</span><span class="hl-0">get</span><span class="hl-1">(</span><span class="hl-4">stream</span><span class="hl-1">);</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<p><a href="media/01_consumers.js">full example</a></p>
|
|
<p>With the consumer in hand, the client can start reading messages using whatever
|
|
API is appropriate for the application.</p>
|
|
<h4 id="jsmsg" class="tsd-anchor-link">JsMsg<a href="#jsmsg" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h4><p>All messages are <code>JsMsg</code>s, a <code>JsMsg</code> is a wrapped <code>Msg</code> - it has all the
|
|
standard fields in a NATS <code>Msg</code>, a <code>JsMsg</code> and provides functionality for
|
|
inspecting metadata encoded into the message's reply subject. This metadata
|
|
includes:</p>
|
|
<ul>
|
|
<li>sequence (<code>seq</code>)</li>
|
|
<li><code>redelivered</code></li>
|
|
<li>full info via info which shows the delivery sequence, and how many messages
|
|
are pending among other things.</li>
|
|
<li>Multiple ways to acknowledge a message:
|
|
<ul>
|
|
<li><code>ack()</code></li>
|
|
<li><code>nak(millis?)</code> - like ack, but tells the server you failed to process it,
|
|
and it should be resent. If a number is specified, the message will be
|
|
resent after the specified value. The additional argument only supported on
|
|
server versions 2.7.1 or greater</li>
|
|
<li><code>working()</code> - informs the server that you are still working on the message
|
|
and thus prevent receiving the message again as a redelivery.</li>
|
|
<li><code>term()</code> - specifies that you failed to process the message and instructs
|
|
the server to not send it again (to any consumer).</li>
|
|
</ul>
|
|
</li>
|
|
</ul>
|
|
<h4 id="requesting-a-single-message" class="tsd-anchor-link">Requesting a single message<a href="#requesting-a-single-message" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h4><p>The simplest mechanism to process messages is to request a single message. This
|
|
requires sending a request to the server. When no messages are available, the
|
|
request will return a <code>null</code> message. Since the client is explicitly requesting
|
|
the message, it is in full control of when to ask for another.</p>
|
|
<p>The request will reject if there's an exceptional condition, such as when the
|
|
underlying consumer or stream is deleted after retrieving the consumer instance,
|
|
or by a change to the clients subject permissions that prevent interactions with
|
|
JetStream, or JetStream is not available.</p>
|
|
<pre><code class="typescript"><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">m</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">c</span><span class="hl-1">.</span><span class="hl-0">next</span><span class="hl-1">();</span><br/><span class="hl-3">if</span><span class="hl-1"> (</span><span class="hl-4">m</span><span class="hl-1">) {</span><br/><span class="hl-1"> </span><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-4">subject</span><span class="hl-1">);</span><br/><span class="hl-1"> </span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-0">ack</span><span class="hl-1">();</span><br/><span class="hl-1">} </span><span class="hl-3">else</span><span class="hl-1"> {</span><br/><span class="hl-1"> </span><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-2">`didn't get a message`</span><span class="hl-1">);</span><br/><span class="hl-1">}</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<p><a href="media/02_next.js">full example</a></p>
|
|
<p>The operation takes an optional argument. Currently, the only option is an
|
|
<code>expires</code> option which specifies the maximum number of milliseconds to wait for
|
|
a message. This is defaulted to 30 seconds. Note this default is a good value
|
|
because it gives the opportunity to retrieve a message without excessively
|
|
polling the server (which could affect the server performance).</p>
|
|
<p><code>next()</code> should be your go-to API when implementing services that process
|
|
messages or work queue streams, as it allows you to horizontally scale your
|
|
processing simply by starting and managing multiple processes.</p>
|
|
<p>Keep in mind that you can also simulate <code>next()</code> in a loop and have the library
|
|
provide an iterator by using <code>consume()</code>. That API be explained later in the
|
|
document.</p>
|
|
<h4 id="fetching-batch-of-messages" class="tsd-anchor-link">Fetching batch of messages<a href="#fetching-batch-of-messages" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h4><p>You can request multiple messages at time. The request is a <em>long</em> poll. So it
|
|
remains open and keep dispatching you messages until the desired number of
|
|
messages is received, or the <code>expires</code> time triggers. This means that the number
|
|
of messages you request is only a hint and it is just the upper bound on the
|
|
number of messages you will receive. By default <code>fetch()</code> will retrieve 100
|
|
messages in a batch, but you can control it as shown in this example:</p>
|
|
<pre><code class="typescript"><span class="hl-3">for</span><span class="hl-1"> (</span><span class="hl-6">let</span><span class="hl-1"> </span><span class="hl-4">i</span><span class="hl-1"> = </span><span class="hl-8">0</span><span class="hl-1">; </span><span class="hl-4">i</span><span class="hl-1"> < </span><span class="hl-8">3</span><span class="hl-1">; </span><span class="hl-4">i</span><span class="hl-1">++) {</span><br/><span class="hl-1"> </span><span class="hl-6">let</span><span class="hl-1"> </span><span class="hl-4">messages</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">c</span><span class="hl-1">.</span><span class="hl-0">fetch</span><span class="hl-1">({ </span><span class="hl-4">max_messages:</span><span class="hl-1"> </span><span class="hl-8">4</span><span class="hl-1">, </span><span class="hl-4">expires:</span><span class="hl-1"> </span><span class="hl-8">2000</span><span class="hl-1"> });</span><br/><span class="hl-1"> </span><span class="hl-3">for</span><span class="hl-1"> </span><span class="hl-3">await</span><span class="hl-1"> (</span><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">m</span><span class="hl-1"> </span><span class="hl-6">of</span><span class="hl-1"> </span><span class="hl-4">messages</span><span class="hl-1">) {</span><br/><span class="hl-1"> </span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-0">ack</span><span class="hl-1">();</span><br/><span class="hl-1"> }</span><br/><span class="hl-1"> </span><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-2">`batch completed: </span><span class="hl-6">${</span><span class="hl-4">messages</span><span class="hl-10">.</span><span class="hl-0">getProcessed</span><span class="hl-10">()</span><span class="hl-6">}</span><span class="hl-2"> msgs processed`</span><span class="hl-1">);</span><br/><span class="hl-1">}</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<p><a href="media/03_batch.js">full example</a></p>
|
|
<p>Fetching batches is useful if you parallelize a number of requests to take
|
|
advantage of the asynchronous processing of data with a number of workers for
|
|
example. To get a new batch simply fetch again.</p>
|
|
<h4 id="consuming-messages" class="tsd-anchor-link">Consuming messages<a href="#consuming-messages" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h4><p>In the previous two sections messages were retrieved manually by your
|
|
application, and allowed you to remain in control of whether you wanted to
|
|
receive one or more messages with a single request.</p>
|
|
<p>A third option automates the process of re-requesting more messages. The library
|
|
will monitor messages it yields, and request additional messages to maintain
|
|
your processing momentum. The operation will continue until you <code>break</code> or call
|
|
<code>stop()</code> the iterator.</p>
|
|
<p>The <code>consume()</code> operation maintains an internal buffer of messages that auto
|
|
refreshes whenever 50% of the initial buffer is consumed. This allows the client
|
|
to process messages in a loop forever.</p>
|
|
<pre><code class="typescript"><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">messages</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">c</span><span class="hl-1">.</span><span class="hl-0">consume</span><span class="hl-1">();</span><br/><span class="hl-3">for</span><span class="hl-1"> </span><span class="hl-3">await</span><span class="hl-1"> (</span><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">m</span><span class="hl-1"> </span><span class="hl-6">of</span><span class="hl-1"> </span><span class="hl-4">messages</span><span class="hl-1">) {</span><br/><span class="hl-1"> </span><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-4">seq</span><span class="hl-1">);</span><br/><span class="hl-1"> </span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-0">ack</span><span class="hl-1">();</span><br/><span class="hl-1">}</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<p><a href="media/04_consume.js">full example</a></p>
|
|
<p>Note that it is possible to do an automatic version of <code>next()</code> by simply
|
|
setting the maximum number of messages to buffer to <code>1</code>:</p>
|
|
<pre><code class="typescript"><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">messages</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">c</span><span class="hl-1">.</span><span class="hl-0">consume</span><span class="hl-1">({ </span><span class="hl-4">max_messages:</span><span class="hl-1"> </span><span class="hl-8">1</span><span class="hl-1"> });</span><br/><span class="hl-3">for</span><span class="hl-1"> </span><span class="hl-3">await</span><span class="hl-1"> (</span><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">m</span><span class="hl-1"> </span><span class="hl-6">of</span><span class="hl-1"> </span><span class="hl-4">messages</span><span class="hl-1">) {</span><br/><span class="hl-1"> </span><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-4">seq</span><span class="hl-1">);</span><br/><span class="hl-1"> </span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-0">ack</span><span class="hl-1">();</span><br/><span class="hl-1">}</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<p>The API simply asks for one message, but as soon as that message is processed or
|
|
the request expires, another is requested.</p>
|
|
<h4 id="horizontally-scaling-consumers-previously-known-as-queue-consumer" class="tsd-anchor-link">Horizontally Scaling Consumers (Previously known as Queue Consumer)<a href="#horizontally-scaling-consumers-previously-known-as-queue-consumer" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h4><p>Scaling processing in a consumer is simply calling <code>next()/fetch()/consume()</code> on
|
|
a shared consumer. When horizontally scaling limiting the number of buffered
|
|
messages will likely yield better results as requests will be mapped 1-1 with
|
|
the processes preventing some processes from booking more messages while others
|
|
are idle.</p>
|
|
<p>A more balanced approach is to simply use <code>next()</code> or
|
|
<code>consume({max_messages: 1})</code>. This makes it so that if you start or stop
|
|
processes you automatically scale your processing, and if there's a failure you
|
|
won't delay the redelivery of messages that were in flight but never processed
|
|
by the client.</p>
|
|
<h4 id="callbacks" class="tsd-anchor-link">Callbacks<a href="#callbacks" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h4><p>The <code>consume()</code> API normally use iterators for processing. If you want to
|
|
specify a callback, you can:</p>
|
|
<pre><code class="typescript"><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">c</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">js</span><span class="hl-1">.</span><span class="hl-4">consumers</span><span class="hl-1">.</span><span class="hl-0">get</span><span class="hl-1">(</span><span class="hl-4">stream</span><span class="hl-1">, </span><span class="hl-4">consumer</span><span class="hl-1">);</span><br/><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-2">"waiting for messages"</span><span class="hl-1">);</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">c</span><span class="hl-1">.</span><span class="hl-0">consume</span><span class="hl-1">({</span><br/><span class="hl-1"> </span><span class="hl-0">callback</span><span class="hl-4">:</span><span class="hl-1"> (</span><span class="hl-4">m</span><span class="hl-1">) </span><span class="hl-6">=></span><span class="hl-1"> {</span><br/><span class="hl-1"> </span><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-4">seq</span><span class="hl-1">);</span><br/><span class="hl-1"> </span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-0">ack</span><span class="hl-1">();</span><br/><span class="hl-1"> },</span><br/><span class="hl-1">});</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<h4 id="iterators-callbacks-and-concurrency" class="tsd-anchor-link">Iterators, Callbacks, and Concurrency<a href="#iterators-callbacks-and-concurrency" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h4><p>The <code>consume()</code> and <code>fetch()</code> APIs yield a <code>ConsumerMessages</code>. One thing to keep
|
|
in mind is that the iterator for processing messages will not yield a new
|
|
message until the body of the loop completes.</p>
|
|
<p>Compare:</p>
|
|
<pre><code class="typescript"><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">msgs</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">c</span><span class="hl-1">.</span><span class="hl-0">consume</span><span class="hl-1">();</span><br/><span class="hl-3">for</span><span class="hl-1"> </span><span class="hl-3">await</span><span class="hl-1"> (</span><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">m</span><span class="hl-1"> </span><span class="hl-6">of</span><span class="hl-1"> </span><span class="hl-4">msgs</span><span class="hl-1">) {</span><br/><span class="hl-1"> </span><span class="hl-3">try</span><span class="hl-1"> {</span><br/><span class="hl-1"> </span><span class="hl-5">// this is simplest but could also create a head-of-line blocking</span><br/><span class="hl-1"> </span><span class="hl-5">// as no other message on the current buffer will be processed until</span><br/><span class="hl-1"> </span><span class="hl-5">// this iteration completes</span><br/><span class="hl-1"> </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-0">asyncFn</span><span class="hl-1">(</span><span class="hl-4">m</span><span class="hl-1">);</span><br/><span class="hl-1"> </span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-0">ack</span><span class="hl-1">();</span><br/><span class="hl-1"> } </span><span class="hl-3">catch</span><span class="hl-1"> (</span><span class="hl-4">err</span><span class="hl-1">) {</span><br/><span class="hl-1"> </span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-0">nack</span><span class="hl-1">();</span><br/><span class="hl-1"> }</span><br/><span class="hl-1">}</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<p>With</p>
|
|
<pre><code class="typescript"><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">msgs</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">c</span><span class="hl-1">.</span><span class="hl-0">consume</span><span class="hl-1">();</span><br/><span class="hl-3">for</span><span class="hl-1"> </span><span class="hl-3">await</span><span class="hl-1"> (</span><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">m</span><span class="hl-1"> </span><span class="hl-6">of</span><span class="hl-1"> </span><span class="hl-4">msgs</span><span class="hl-1">) {</span><br/><span class="hl-1"> </span><span class="hl-5">// this potentially has the problem of generating a very large number</span><br/><span class="hl-1"> </span><span class="hl-5">// of async operations which may exceed the limits of the runtime</span><br/><span class="hl-1"> </span><span class="hl-0">asyncFn</span><span class="hl-1">(</span><span class="hl-4">m</span><span class="hl-1">)</span><br/><span class="hl-1"> .</span><span class="hl-0">then</span><span class="hl-1">(() </span><span class="hl-6">=></span><span class="hl-1"> {</span><br/><span class="hl-1"> </span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-0">ack</span><span class="hl-1">();</span><br/><span class="hl-1"> })</span><br/><span class="hl-1"> .</span><span class="hl-0">catch</span><span class="hl-1">((</span><span class="hl-4">err</span><span class="hl-1">) </span><span class="hl-6">=></span><span class="hl-1"> {</span><br/><span class="hl-1"> </span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-0">nack</span><span class="hl-1">();</span><br/><span class="hl-1"> });</span><br/><span class="hl-1">}</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<p>In the first scenario, the processing is sequential. The second scenario is
|
|
concurrent.</p>
|
|
<p>Both of these behaviors are standard JavaScript, but you can use this to your
|
|
advantage. You can improve latency by not awaiting, but that will require a more
|
|
complex handling as you'll need to restrict and limit how many concurrent
|
|
operations you create and thus avoid hitting limits in your runtime.</p>
|
|
<p>One possible strategy is to use <code>fetch()</code>, and process asynchronously without
|
|
awaiting as you process message you'll need to implement accounting to track
|
|
when you should re-fetch, but a far simpler solution is to use <code>next()</code>, process
|
|
asynchronously and scale by horizontally managing processes instead.</p>
|
|
<p>Here's a solution that introduces a rate limiter:</p>
|
|
<pre><code class="typescript"><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">messages</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">c</span><span class="hl-1">.</span><span class="hl-0">consume</span><span class="hl-1">({ </span><span class="hl-4">max_messages:</span><span class="hl-1"> </span><span class="hl-8">10</span><span class="hl-1"> });</span><br/><br/><span class="hl-5">// this rate limiter is just example code, do not use in production</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">rl</span><span class="hl-1"> = </span><span class="hl-6">new</span><span class="hl-1"> </span><span class="hl-0">SimpleMutex</span><span class="hl-1">(</span><span class="hl-8">5</span><span class="hl-1">);</span><br/><br/><span class="hl-6">async</span><span class="hl-1"> </span><span class="hl-6">function</span><span class="hl-1"> </span><span class="hl-0">schedule</span><span class="hl-1">(</span><span class="hl-4">m</span><span class="hl-1">: </span><span class="hl-9">JsMsg</span><span class="hl-1">): </span><span class="hl-9">Promise</span><span class="hl-1"><</span><span class="hl-9">void</span><span class="hl-1">> {</span><br/><span class="hl-1"> </span><span class="hl-5">// pretend to do work</span><br/><span class="hl-1"> </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-0">delay</span><span class="hl-1">(</span><span class="hl-8">1000</span><span class="hl-1">);</span><br/><span class="hl-1"> </span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-0">ack</span><span class="hl-1">();</span><br/><span class="hl-1"> </span><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-2">`</span><span class="hl-6">${</span><span class="hl-4">m</span><span class="hl-10">.</span><span class="hl-4">seq</span><span class="hl-6">}</span><span class="hl-2">`</span><span class="hl-1">);</span><br/><span class="hl-1">}</span><br/><br/><span class="hl-3">for</span><span class="hl-1"> </span><span class="hl-3">await</span><span class="hl-1"> (</span><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">m</span><span class="hl-1"> </span><span class="hl-6">of</span><span class="hl-1"> </span><span class="hl-4">messages</span><span class="hl-1">) {</span><br/><span class="hl-1"> </span><span class="hl-5">// block reading messages until we have capacity</span><br/><span class="hl-1"> </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">rl</span><span class="hl-1">.</span><span class="hl-0">lock</span><span class="hl-1">();</span><br/><span class="hl-1"> </span><span class="hl-0">schedule</span><span class="hl-1">(</span><span class="hl-4">m</span><span class="hl-1">)</span><br/><span class="hl-1"> .</span><span class="hl-0">catch</span><span class="hl-1">((</span><span class="hl-4">err</span><span class="hl-1">) </span><span class="hl-6">=></span><span class="hl-1"> {</span><br/><span class="hl-1"> </span><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-2">`failed processing: </span><span class="hl-6">${</span><span class="hl-4">err</span><span class="hl-10">.</span><span class="hl-4">message</span><span class="hl-6">}</span><span class="hl-2">`</span><span class="hl-1">);</span><br/><span class="hl-1"> </span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-0">nak</span><span class="hl-1">();</span><br/><span class="hl-1"> })</span><br/><span class="hl-1"> .</span><span class="hl-0">finally</span><span class="hl-1">(() </span><span class="hl-6">=></span><span class="hl-1"> {</span><br/><span class="hl-1"> </span><span class="hl-5">// unblock, allowing a lock to resolve</span><br/><span class="hl-1"> </span><span class="hl-4">rl</span><span class="hl-1">.</span><span class="hl-0">unlock</span><span class="hl-1">();</span><br/><span class="hl-1"> });</span><br/><span class="hl-1">}</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<p><a href="media/07_consume_jobs.js">full example</a></p>
|
|
<h4 id="processing-a-stream" class="tsd-anchor-link">Processing a Stream<a href="#processing-a-stream" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h4><p>Here's a contrived example on how a process examines a stream. Once all the
|
|
messages in the stream are processed the consumer is deleted.</p>
|
|
<p>Our processing is simply creating a frequency table of the second token in the
|
|
subject, printing the results after it is done.</p>
|
|
<pre><code class="typescript"><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">messages</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">c</span><span class="hl-1">.</span><span class="hl-0">consume</span><span class="hl-1">();</span><br/><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">data</span><span class="hl-1"> = </span><span class="hl-6">new</span><span class="hl-1"> </span><span class="hl-0">Map</span><span class="hl-1"><</span><span class="hl-9">string</span><span class="hl-1">, </span><span class="hl-9">number</span><span class="hl-1">>();</span><br/><span class="hl-3">for</span><span class="hl-1"> </span><span class="hl-3">await</span><span class="hl-1"> (</span><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">m</span><span class="hl-1"> </span><span class="hl-6">of</span><span class="hl-1"> </span><span class="hl-4">messages</span><span class="hl-1">) {</span><br/><span class="hl-1"> </span><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">chunks</span><span class="hl-1"> = </span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-4">subject</span><span class="hl-1">.</span><span class="hl-0">split</span><span class="hl-1">(</span><span class="hl-2">"."</span><span class="hl-1">);</span><br/><span class="hl-1"> </span><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">v</span><span class="hl-1"> = </span><span class="hl-4">data</span><span class="hl-1">.</span><span class="hl-0">get</span><span class="hl-1">(</span><span class="hl-4">chunks</span><span class="hl-1">[</span><span class="hl-8">1</span><span class="hl-1">]) || </span><span class="hl-8">0</span><span class="hl-1">;</span><br/><span class="hl-1"> </span><span class="hl-4">data</span><span class="hl-1">.</span><span class="hl-0">set</span><span class="hl-1">(</span><span class="hl-4">chunks</span><span class="hl-1">[</span><span class="hl-8">1</span><span class="hl-1">], </span><span class="hl-4">v</span><span class="hl-1"> + </span><span class="hl-8">1</span><span class="hl-1">);</span><br/><span class="hl-1"> </span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-0">ack</span><span class="hl-1">();</span><br/><br/><span class="hl-1"> </span><span class="hl-5">// if no pending, then we have processed the stream</span><br/><span class="hl-1"> </span><span class="hl-5">// and we can break</span><br/><span class="hl-1"> </span><span class="hl-3">if</span><span class="hl-1"> (</span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-4">info</span><span class="hl-1">.</span><span class="hl-4">pending</span><span class="hl-1"> === </span><span class="hl-8">0</span><span class="hl-1">) {</span><br/><span class="hl-1"> </span><span class="hl-3">break</span><span class="hl-1">;</span><br/><span class="hl-1"> }</span><br/><span class="hl-1">}</span><br/><br/><span class="hl-5">// we can safely delete the consumer</span><br/><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">c</span><span class="hl-1">.</span><span class="hl-0">delete</span><span class="hl-1">();</span><br/><br/><span class="hl-5">// and print results</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">keys</span><span class="hl-1"> = [];</span><br/><span class="hl-3">for</span><span class="hl-1"> (</span><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">k</span><span class="hl-1"> </span><span class="hl-6">of</span><span class="hl-1"> </span><span class="hl-4">data</span><span class="hl-1">.</span><span class="hl-0">keys</span><span class="hl-1">()) {</span><br/><span class="hl-1"> </span><span class="hl-4">keys</span><span class="hl-1">.</span><span class="hl-0">push</span><span class="hl-1">(</span><span class="hl-4">k</span><span class="hl-1">);</span><br/><span class="hl-1">}</span><br/><span class="hl-4">keys</span><span class="hl-1">.</span><span class="hl-0">sort</span><span class="hl-1">();</span><br/><span class="hl-4">keys</span><span class="hl-1">.</span><span class="hl-0">forEach</span><span class="hl-1">((</span><span class="hl-4">k</span><span class="hl-1">) </span><span class="hl-6">=></span><span class="hl-1"> {</span><br/><span class="hl-1"> </span><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-2">`</span><span class="hl-6">${</span><span class="hl-4">k</span><span class="hl-6">}</span><span class="hl-2">: </span><span class="hl-6">${</span><span class="hl-4">data</span><span class="hl-10">.</span><span class="hl-0">get</span><span class="hl-10">(</span><span class="hl-4">k</span><span class="hl-10">)</span><span class="hl-6">}</span><span class="hl-2">`</span><span class="hl-1">);</span><br/><span class="hl-1">});</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<p><a href="media/08_consume_process.js">full example</a></p>
|
|
<h3 id="heartbeats" class="tsd-anchor-link">Heartbeats<a href="#heartbeats" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h3><p>Since JetStream is available through NATS it is possible for your network
|
|
connection to not be directly connected to the JetStream server providing you
|
|
with messages. In those cases, it is possible for a JetStream server to go away,
|
|
and for the client to not notice it. This would mean your client would sit idle
|
|
thinking there are no messages, when in reality the JetStream service may have
|
|
restarted elsewhere.</p>
|
|
<p>For most issues, the client will auto-recover, but if it doesn't, and it starts
|
|
reporting <code>HeartbeatsMissed</code> statuses, you will want to <code>stop()</code> the
|
|
<code>ConsumerMessages</code>, and recreate it. Note that in the example below this is done
|
|
in a loop for example purposes:</p>
|
|
<pre><code class="typescript"><span class="hl-3">while</span><span class="hl-1"> (</span><span class="hl-6">true</span><span class="hl-1">) {</span><br/><span class="hl-1"> </span><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">messages</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">c</span><span class="hl-1">.</span><span class="hl-0">consume</span><span class="hl-1">({ </span><span class="hl-4">max_messages:</span><span class="hl-1"> </span><span class="hl-8">1</span><span class="hl-1"> });</span><br/><br/><span class="hl-1"> </span><span class="hl-5">// watch the to see if the consume operation misses heartbeats</span><br/><span class="hl-1"> (</span><span class="hl-6">async</span><span class="hl-1"> () </span><span class="hl-6">=></span><span class="hl-1"> {</span><br/><span class="hl-1"> </span><span class="hl-3">for</span><span class="hl-1"> </span><span class="hl-3">await</span><span class="hl-1"> (</span><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">s</span><span class="hl-1"> </span><span class="hl-6">of</span><span class="hl-1"> </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">messages</span><span class="hl-1">.</span><span class="hl-0">status</span><span class="hl-1">()) {</span><br/><span class="hl-1"> </span><span class="hl-3">switch</span><span class="hl-1"> (</span><span class="hl-4">s</span><span class="hl-1">.</span><span class="hl-4">type</span><span class="hl-1">) {</span><br/><span class="hl-1"> </span><span class="hl-3">case</span><span class="hl-1"> </span><span class="hl-2">"heartbeats_missed"</span><span class="hl-1">:</span><br/><span class="hl-1"> </span><span class="hl-5">// you can decide how many heartbeats you are willing to miss</span><br/><span class="hl-1"> </span><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">n</span><span class="hl-1"> = </span><span class="hl-4">s</span><span class="hl-1">.</span><span class="hl-4">data</span><span class="hl-1"> </span><span class="hl-3">as</span><span class="hl-1"> </span><span class="hl-9">number</span><span class="hl-1">;</span><br/><span class="hl-1"> </span><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-2">`</span><span class="hl-6">${</span><span class="hl-4">s</span><span class="hl-10">.</span><span class="hl-4">count</span><span class="hl-6">}</span><span class="hl-2"> heartbeats missed`</span><span class="hl-1">);</span><br/><span class="hl-1"> </span><span class="hl-3">if</span><span class="hl-1"> (</span><span class="hl-4">s</span><span class="hl-1">.</span><span class="hl-4">count</span><span class="hl-1"> === </span><span class="hl-8">2</span><span class="hl-1">) {</span><br/><span class="hl-1"> </span><span class="hl-5">// by calling `stop()` the message processing loop ends</span><br/><span class="hl-1"> </span><span class="hl-5">// in this case this is wrapped by a loop, so it attempts</span><br/><span class="hl-1"> </span><span class="hl-5">// to re-setup the consume</span><br/><span class="hl-1"> </span><span class="hl-4">messages</span><span class="hl-1">.</span><span class="hl-0">stop</span><span class="hl-1">();</span><br/><span class="hl-1"> }</span><br/><span class="hl-1"> }</span><br/><span class="hl-1"> }</span><br/><span class="hl-1"> })();</span><br/><span class="hl-1"> </span><span class="hl-3">for</span><span class="hl-1"> </span><span class="hl-3">await</span><span class="hl-1"> (</span><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">m</span><span class="hl-1"> </span><span class="hl-6">of</span><span class="hl-1"> </span><span class="hl-4">messages</span><span class="hl-1">) {</span><br/><span class="hl-1"> </span><span class="hl-4">console</span><span class="hl-1">.</span><span class="hl-0">log</span><span class="hl-1">(</span><span class="hl-2">`</span><span class="hl-6">${</span><span class="hl-4">m</span><span class="hl-10">.</span><span class="hl-4">seq</span><span class="hl-6">}</span><span class="hl-2"> </span><span class="hl-6">${</span><span class="hl-4">m</span><span class="hl-10">?.</span><span class="hl-4">subject</span><span class="hl-6">}</span><span class="hl-2">`</span><span class="hl-1">);</span><br/><span class="hl-1"> </span><span class="hl-4">m</span><span class="hl-1">.</span><span class="hl-0">ack</span><span class="hl-1">();</span><br/><span class="hl-1"> }</span><br/><span class="hl-1">}</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<p><a href="media/06_heartbeats.js">full example</a></p>
|
|
<p>Note that while the heartbeat interval is configurable, you shouldn't change it.</p>
|
|
<h4 id="jetstream-ordered-consumers" class="tsd-anchor-link">JetStream Ordered Consumers<a href="#jetstream-ordered-consumers" aria-label="Permalink" class="tsd-anchor-icon"><svg viewBox="0 0 24 24" aria-hidden="true"><use href="assets/icons.svg#icon-anchor"></use></svg></a></h4><p>An Ordered Consumer is a specialized consumer that ensures that messages are
|
|
presented in the correct order. If a message is out of order, the consumer is
|
|
recreated at the expected sequence.</p>
|
|
<p>The underlying consumer is created, managed and destroyed under the covers, so
|
|
you only have to specify the stream and possible startup options, or filtering:</p>
|
|
<pre><code class="typescript"><span class="hl-5">// note the name of the consumer is not specified</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">a</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">js</span><span class="hl-1">.</span><span class="hl-4">consumers</span><span class="hl-1">.</span><span class="hl-0">get</span><span class="hl-1">(</span><span class="hl-4">name</span><span class="hl-1">);</span><br/><span class="hl-6">const</span><span class="hl-1"> </span><span class="hl-7">b</span><span class="hl-1"> = </span><span class="hl-3">await</span><span class="hl-1"> </span><span class="hl-4">js</span><span class="hl-1">.</span><span class="hl-4">consumers</span><span class="hl-1">.</span><span class="hl-0">get</span><span class="hl-1">(</span><span class="hl-4">name</span><span class="hl-1">, { </span><span class="hl-4">filterSubjects:</span><span class="hl-1"> [</span><span class="hl-2">`</span><span class="hl-6">${</span><span class="hl-4">name</span><span class="hl-6">}</span><span class="hl-2">.a`</span><span class="hl-1">] });</span>
|
|
</code><button type="button">Copy</button></pre>
|
|
|
|
<p>Note that uses of the consumer API for reading messages are checked for
|
|
concurrency preventing the ordered consumer from having operations initiated
|
|
with <code>fetch()</code> and <code>consume()</code> or <code>next()</code> while another is active.</p>
|
|
</div></div><div class="col-sidebar"><div class="page-menu"><div class="tsd-navigation settings"><details class="tsd-accordion"><summary class="tsd-accordion-summary"><svg width="20" height="20" viewBox="0 0 24 24" fill="none" aria-hidden="true"><use href="assets/icons.svg#icon-chevronDown"></use></svg><h3>Settings</h3></summary><div class="tsd-accordion-details"><div class="tsd-filter-visibility"><span class="settings-label">Member Visibility</span><ul id="tsd-filter-options"><li class="tsd-filter-item"><label class="tsd-filter-input"><input type="checkbox" id="tsd-filter-protected" name="protected"/><svg width="32" height="32" viewBox="0 0 32 32" aria-hidden="true"><rect class="tsd-checkbox-background" width="30" height="30" x="1" y="1" rx="6" fill="none"></rect><path class="tsd-checkbox-checkmark" d="M8.35422 16.8214L13.2143 21.75L24.6458 10.25" stroke="none" stroke-width="3.5" stroke-linejoin="round" fill="none"></path></svg><span>Protected</span></label></li><li class="tsd-filter-item"><label class="tsd-filter-input"><input type="checkbox" id="tsd-filter-inherited" name="inherited" checked/><svg width="32" height="32" viewBox="0 0 32 32" aria-hidden="true"><rect class="tsd-checkbox-background" width="30" height="30" x="1" y="1" rx="6" fill="none"></rect><path class="tsd-checkbox-checkmark" d="M8.35422 16.8214L13.2143 21.75L24.6458 10.25" stroke="none" stroke-width="3.5" stroke-linejoin="round" fill="none"></path></svg><span>Inherited</span></label></li><li class="tsd-filter-item"><label class="tsd-filter-input"><input type="checkbox" id="tsd-filter-external" name="external"/><svg width="32" height="32" viewBox="0 0 32 32" aria-hidden="true"><rect class="tsd-checkbox-background" width="30" height="30" x="1" y="1" rx="6" fill="none"></rect><path class="tsd-checkbox-checkmark" d="M8.35422 16.8214L13.2143 21.75L24.6458 10.25" stroke="none" stroke-width="3.5" stroke-linejoin="round" fill="none"></path></svg><span>External</span></label></li></ul></div><div class="tsd-theme-toggle"><label class="settings-label" for="tsd-theme">Theme</label><select id="tsd-theme"><option value="os">OS</option><option value="light">Light</option><option value="dark">Dark</option></select></div></div></details></div><details open class="tsd-accordion tsd-page-navigation"><summary class="tsd-accordion-summary"><svg width="20" height="20" viewBox="0 0 24 24" fill="none" aria-hidden="true"><use href="assets/icons.svg#icon-chevronDown"></use></svg><h3>On This Page</h3></summary><div class="tsd-accordion-details"><a href="#jetstream"><span>Jet<wbr/>Stream</span></a><ul><li><a href="#installation"><span>Installation</span></a></li><li><ul><li><a href="#npm"><span>NPM</span></a></li><li><a href="#jsr"><span>JSR</span></a></li></ul></li><li><a href="#referencing-the-library"><span>Referencing the library</span></a></li></ul><a href="#jetstream-1"><span>Jet<wbr/>Stream</span></a><ul><li><a href="#migration"><span>Migration</span></a></li><li><a href="#jetstreammanager-jsm"><span>Jet<wbr/>Stream<wbr/>Manager (JSM)</span></a></li><li><a href="#jetstream-client"><span>Jet<wbr/>Stream <wbr/>Client</span></a></li><li><ul><li><a href="#processing-messages"><span>Processing <wbr/>Messages</span></a></li><li><ul><li><a href="#basics"><span>Basics</span></a></li><li><a href="#retrieving-the-consumer"><span>Retrieving the <wbr/>Consumer</span></a></li><li><a href="#jsmsg"><span>Js<wbr/>Msg</span></a></li><li><a href="#requesting-a-single-message"><span>Requesting a single message</span></a></li><li><a href="#fetching-batch-of-messages"><span>Fetching batch of messages</span></a></li><li><a href="#consuming-messages"><span>Consuming messages</span></a></li><li><a href="#horizontally-scaling-consumers-previously-known-as-queue-consumer"><span>Horizontally <wbr/>Scaling <wbr/>Consumers (<wbr/>Previously known as <wbr/>Queue <wbr/>Consumer)</span></a></li><li><a href="#callbacks"><span>Callbacks</span></a></li><li><a href="#iterators-callbacks-and-concurrency"><span>Iterators, <wbr/>Callbacks, and <wbr/>Concurrency</span></a></li><li><a href="#processing-a-stream"><span>Processing a <wbr/>Stream</span></a></li></ul></li><li><a href="#heartbeats"><span>Heartbeats</span></a></li><li><ul><li><a href="#jetstream-ordered-consumers"><span>Jet<wbr/>Stream <wbr/>Ordered <wbr/>Consumers</span></a></li></ul></li></ul></li></ul></div></details></div><div class="site-menu"><nav class="tsd-navigation"><a href="modules.html">@nats-io/jetstream</a><ul class="tsd-small-nested-navigation" id="tsd-nav-container"><li>Loading...</li></ul></nav></div></div></div><footer><p class="tsd-generator">Generated using <a href="https://typedoc.org/" target="_blank">TypeDoc</a></p></footer><div class="overlay"></div></body></html>
|