UNPKG

9.64 kBMarkdownView Raw
1# NATS Streaming - Node.js Client
2
3Node NATS Streaming is an extremely performant, lightweight reliable streaming platform powered by [NATS](http://nats.io) for [Node.js](http://nodejs.org/).
4
5[![license](https://img.shields.io/github/license/nats-io/node-nats-streaming.svg)](https://www.apache.org/licenses/LICENSE-2.0)
6[![Travis branch](https://img.shields.io/travis/nats-io/node-nats-streaming/master.svg)]()
7[![Coverage Status](https://coveralls.io/repos/github/nats-io/node-nats-streaming/badge.svg?branch=master)](https://coveralls.io/github/nats-io/node-nats-streaming?branch=master)[![npm](https://img.shields.io/npm/v/node-nats-streaming.svg)](https://www.npmjs.com/package/node-nats-streaming)
8[![npm](https://img.shields.io/npm/dt/node-nats-streaming.svg)](https://www.npmjs.com/package/node-nats-streaming)
9
10
11
12
13NATS Streaming provides the following high-level feature set:
14- Log based persistence
15- At-Least-Once Delivery model, giving reliable message delivery
16- Rate matched on a per subscription basis
17- Replay/Restart
18- Last Value Semantics
19
20## Installation
21
22```bash
23npm install node-nats-streaming
24```
25
26## Basic Usage
27```javascript
28#!/usr/bin/env node
29
30"use-strict";
31
32var stan = require('node-nats-streaming').connect('test-cluster', 'test');
33
34stan.on('connect', function () {
35
36 // Simple Publisher (all publishes are async in the node version of the client)
37 stan.publish('foo', 'Hello node-nats-streaming!', function(err, guid){
38 if(err) {
39 console.log('publish failed: ' + err);
40 } else {
41 console.log('published message with guid: ' + guid);
42 }
43 });
44
45 // Subscriber can specify how many existing messages to get.
46 var opts = stan.subscriptionOptions().setStartWithLastReceived();
47 var subscription = stan.subscribe('foo', opts);
48 subscription.on('message', function (msg) {
49 console.log('Received a message [' + msg.getSequence() + '] ' + msg.getData());
50 });
51
52 // After one second, unsubscribe, when that is done, close the connection
53 setTimeout(function() {
54 subscription.unsubscribe();
55 subscription.on('unsubscribed', function() {
56 stan.close();
57 });
58 }, 1000);
59});
60
61stan.on('close', function() {
62 process.exit();
63});
64```
65
66### Subscription Start (i.e. Replay) Options
67
68NATS Streaming subscriptions are similar to NATS subscriptions, but clients may start their subscription at an earlier point in the message stream, allowing them to receive messages that were published before this client registered interest.
69
70The options are described with examples below:
71
72```javascript
73 // Subscribe starting with the most recently published value
74 var opts = stan.subscriptionOptions();
75 opts.setStartWithLastReceived();
76 var subscription = stan.subscribe('foo', opts);
77
78 // Receive all stored values in order
79 var opts = stan.subscriptionOptions();
80 opts.setDeliverAllAvailable();
81 var subscription = stan.subscribe('foo', opts);
82
83 // Receive all messages starting at a specific sequence number
84 var opts = stan.subscriptionOptions();
85 opts.setStartAtSequence(22);
86 var subscription = stan.subscribe('foo', opts);
87
88 // Subscribe starting at a specific time
89 var d = new Date(2016, 7, 8); // August 8th, 2016
90 var opts = stan.subscriptionOptions();
91 opts.setStartTime(d);
92 var subscription = stan.subscribe('foo', opts);
93
94 // Subscribe starting at a specific amount of time in the past (e.g. 30 seconds ago)
95 var opts = stan.subscriptionOptions();
96 opts.setStartAtTimeDelta(30*1000); // 30 seconds ago
97 var subscription = stan.subscribe('foo', opts);
98```
99
100### Wildcard Subscriptions
101
102NATS Streaming subscriptions **do not** support wildcards.
103
104### Durable Subscriptions
105
106Replay of messages offers great flexibility for clients wishing to begin processing at some earlier point in the data stream.
107However, some clients just need to pick up where they left off from an earlier session, without having to manually track their position in the stream of messages.
108Durable subscriptions allow clients to assign a durable name to a subscription when it is created.
109Doing this causes the NATS Streaming server to track the last acknowledged message for that clientID + durable name, so that only messages since the last acknowledged message will be delivered to the client.
110
111```javascript
112var stan = require('node-nats-streaming').connect('test-cluster', 'client-123');
113
114stan.on('connect', function () {
115 // Subscribe with durable name
116 var opts = stan.subscriptionOptions();
117 opts.setDeliverAllAvailable();
118 opts.setDurableName('my-durable');
119
120 var durableSub = stan.subscribe('foo', opts);
121 durableSub.on('message', function(msg) {
122 console.log('Received a message: ' + msg.getData());
123 });
124
125 //...
126 // client suspends durable subscription
127 //
128 durableSub.close();
129
130 //...
131 // client resumes durable subscription
132 //
133 durableSub = stan.subscribe('foo', opts);
134 durableSub.on('message', function(msg) {
135 console.log('Received a message: ' + msg.getData());
136 });
137
138 // ...
139 // client receives message sequence 1-40, and disconnects
140 stan.close();
141
142 // client reconnects in the future with same clientID
143 var stan = require('node-nats-streaming').connect('test-cluster', 'client-123');
144 var durableSub = stan.subscribe('foo', opts);
145 durableSub.on('message', function(msg) {
146 console.log('Received a message: ' + msg.getData());
147 });
148});
149```
150
151### Queue Groups
152
153Subscriptions with the same queue name will form a queue group. Each message is only delivered to a single subscriber per queue group. You can have as many queue groups as you wish. Normal subscribers are not affected by queue group semantics.
154
155```javascript
156 var opts = stan.subscriptionOptions();
157 opts.setStartWithLastReceived();
158 var subscription = stan.subscribe('foo', 'foo.workers', opts);
159```
160
161### Asynchronous Publishing
162
163For each message published, a [NUID](https://github.com/nats-io/nuid) is generated for the message on creation. When the message is received by the server, the client library is notified on its optional callback:
164
165```javascript
166 var guid = stan.publish('foo', 'Hello World!', function(err, aGuid){
167 // err will be undefined if the message was accepted by the
168 // NATS streaming server
169 if(err) {
170 console.log('Error publishing: ' + aGuid + ' - ' + err);
171 }
172 });
173```
174
175#### Message Acknowledgements and Redelivery
176
177NATS Streaming offers At-Least-Once delivery semantics, meaning that once a message has been delivered to an eligible subscriber, if an acknowledgement is not received within the configured timeout interval, NATS Streaming will attempt redelivery of the message.
178This timeout interval is specified by the subscription option `SubscriptionOptions#setAckWait(millis)`, which defaults to 30 seconds.
179
180By default, messages are automatically acknowledged by the NATS Streaming client library after the subscriber's message handler is invoked. However, there may be cases in which the subscribing client wishes to accelerate or defer acknowledgement of the message.
181To do this, the client must set manual acknowledgement mode on the subscription, and invoke `Message#ack()` on the `Message`.
182
183```javascript
184 var opts = stan.subscriptionOptions();
185 opts.setManualAckMode(true);
186 opts.setAckWait(60*1000); //60s
187
188 var sub = stan.subscribe('Foo', opts);
189
190 sub.on('message', function (msg) {
191 // do something with the message
192 msg.ack();
193 });
194```
195
196### Synchronous Publishing
197
198The Nodejs client does not support synchronous publishing.
199
200
201### Rate limiting/matching
202
203A classic problem of publish-subscribe messaging is matching the rate of message producers with the rate of message consumers.
204Message producers can often outpace the speed of the subscribers that are consuming their messages.
205This mismatch is commonly called a "fast producer/slow consumer" problem, and may result in dramatic resource utilization spikes in the underlying messaging system as it tries to buffer messages until the slow consumer(s) can catch up.
206
207Under Nodejs, this is even more important, as in Nodejs is a single-threaded environment. This means that if your application is CPU bound, it is possible for your application to block the processing of outgoing or incoming messages.
208
209### Publisher rate limiting
210
211NATS Streaming provides a connection option called `maxPubAcksInflight` that effectively limits the number of unacknowledged messages that a publisher may have in-flight at any given time. When this maximum is reached, your publisher's callback will be invoked with an error. If not callback was defined, an error will be thrown until the number of unacknowledged messages fall below the specified limit.
212
213### Subscriber rate limiting
214
215Rate limiting may also be accomplished on the subscriber side, on a per-subscription basis, using a subscription option called `SubscriptionOptions#setMaxInFlight(number)`. This option specifies the maximum number of outstanding acknowledgements (messages that have been delivered but not acknowledged) that NATS Streaming will allow for a given subscription.
216When this limit is reached, NATS Streaming will suspend delivery of messages to this subscription until the number of unacknowledged messages falls below the specified limit.
217
218## Supported Node Versions
219
220Support policy for Nodejs versions follows
221[Nodejs release support]( https://github.com/nodejs/Release).
222We will support and build node-nats-streaming on even Nodejs versions that are current
223or in maintenance.
224
225## License
226
227Unless otherwise noted, the NATS source files are distributed under the Apache Version 2.0 license found in the LICENSE file.
228