simplified doc #506
simplified doc #506
Conversation
jetstream.md
Outdated
| specific time. The configuration also specifies if the server should require for | ||
| a message to be acknowledged, and how long to wait for acknowledgements. The |
There was a problem hiding this comment.
| specific time. The configuration also specifies if the server should require for | |
| a message to be acknowledged, and how long to wait for acknowledgements. The | |
| specific time. The configuration also specifies if the server should require messages | |
| to be acknowledged and how long to wait for acknowledgements. The |
| @@ -47,7 +47,7 @@ const stream = "mystream"; | |||
| const subj = `mystream.*`; | |||
| await jsm.streams.add({ name: stream, subjects: [subj] }); | |||
|
|
|||
| // publish a reg nats message directly to the stream | |||
| // jetstream can capture nats core messages | |||
There was a problem hiding this comment.
I suggest this comment be moved to when the stream is defined. The wording I typically use is "one or more subjects can be bound to a stream". I know "binding" may sound exclusive which we know is not technically true since i can still spin up a sub on the same subject. (Thinking out loud) I also describe a stream as a "responder" to messages on a specific subject whose job is to store them.
| The JetStream client functionality presents a JetStream view on a NATS client. | ||
| While you can implement your own API to interact with streams, the JetStream | ||
| APIs make this convenient. | ||
| The JetStream client presents an API for working with messages stored on a |
There was a problem hiding this comment.
| The JetStream client presents an API for working with messages stored on a | |
| The JetStream client presents an API for working with messages stored in a |
| const jsm = await nc.jetstreamManager(); | ||
| await jsm.streams.add({ name: "a", subjects: ["a.*"] }); | ||
|
|
||
| // create a jetstream client: | ||
| const js = nc.jetstream(); | ||
|
|
||
| // to publish messages to a stream: | ||
| // to publish messages to a stream |
There was a problem hiding this comment.
| // to publish messages to a stream | |
| // publish a message received by a stream |
| let pa = await js.publish("a.b"); | ||
| // the jetstream returns an acknowledgement with the | ||
| // jetstream returns an acknowledgement with the |
There was a problem hiding this comment.
One subtle improvement I am attempting with communication is to not refer to jetstream and simply refer to streams and consumers (KV, etc.) as part of NATS. JS is a subsystem, but people should not have to care. Obviously a larger team-wide point to make, but just noting here 😄
| // for the same ID for a configured amount of time (specified | ||
| // by the stream configuration), and reject messages that | ||
| // sport the same ID: |
There was a problem hiding this comment.
| // for the same ID for a configured amount of time (specified | |
| // by the stream configuration), and reject messages that | |
| // sport the same ID: | |
| // for the same ID for a configured amount of time (within a | |
| // configurable time window), and reject messages that have | |
| // the same ID: |
| while (true) { | ||
| const messages = await c.consume({ max_messages: 1 }); | ||
|
|
||
| // watch the to see if the consume operation misses heartbeats |
There was a problem hiding this comment.
| // watch the to see if the consume operation misses heartbeats | |
| // watch to see if the consume operation misses heartbeats |
| const n = s.data as number; | ||
| console.log(`${n} heartbeats missed`); | ||
| if (n === 2) { | ||
| // by calling `stop()` the message processing loop ends |
There was a problem hiding this comment.
| // by calling `stop()` the message processing loop ends | |
| // by calling `stop()` the message processing loop ends. |
| (async () => { | ||
| for await (const s of await messages.status()) { | ||
| if (s.type === ConsumerEvents.HeartbeatsMissed) { | ||
| // you can decide how many heartbeats you are willing to miss |
There was a problem hiding this comment.
What do the heartbeats semantically correspond to? Availability to the consumer to respond? Not just availability of messages?
There was a problem hiding this comment.
The server should be sending heartbeats whenever messages are not being sent. This allows the client to know that the server request is alive.
| const js = nc.jetstream(); | ||
| const c = await js.consumers.get(stream, consumer); | ||
|
|
||
| // this example uses a consume that is monitors a rate limiter |
There was a problem hiding this comment.
(?)
| // this example uses a consume that is monitors a rate limiter | |
| // this example uses consume that monitors a rate limiter |
There was a problem hiding this comment.
Changed to // this example controls parallel processing of the messages // by only allowing 5 concurrent messages to be processed // and then only allowing additional processing as others complete
No description provided.