Files

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">&quot;@nats-io/jetstream&quot;</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">&quot;@nats-io/jetstream&quot;</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">&quot;mystream&quot;</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"> &lt; </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">&quot;mystream.A&quot;</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">&quot;a.b&quot;</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">&quot;a.b&quot;</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">&quot;a.c&quot;</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">&quot;a.d&quot;</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">=&gt;</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">&quot;me&quot;</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&#39;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">&quot;me&quot;</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">&quot;me&quot;</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">&quot;a&quot;</span><span class="hl-1">, </span><span class="hl-4">subjects:</span><span class="hl-1"> [</span><span class="hl-2">&quot;a.*&quot;</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">&quot;a.b&quot;</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&#39;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">&quot;a.b&quot;</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">&quot;a&quot;</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">&quot;a.b&quot;</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">&quot;a&quot;</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">&quot;a.b&quot;</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">&quot;a.b&quot;</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">&quot;a&quot;</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">&lt;</span><span class="hl-9">PubAck</span><span class="hl-1">&gt;[] = [];</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"> &lt; </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">&quot;a.a&quot;</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 &quot;a.b&quot; 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">&quot;a.b&quot;</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 &quot;buffering strategy&quot; 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 &quot;as fast as
possible&quot;. 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">&quot;a&quot;</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">&quot;a&quot;</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&#39;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"> &lt; </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">&quot;waiting for messages&quot;</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">=&gt;</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">=&gt;</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">=&gt;</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">&lt;</span><span class="hl-9">void</span><span class="hl-1">&gt; {</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">=&gt;</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">=&gt;</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">&lt;</span><span class="hl-9">string</span><span class="hl-1">, </span><span class="hl-9">number</span><span class="hl-1">&gt;();</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">&quot;.&quot;</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">=&gt;</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">=&gt;</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">&quot;heartbeats_missed&quot;</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>