Work with streams and consumers on CloudFerro Cloud
The live subscription in the connection guide receives a message only while it is running. A dashboard and a reporting application also need events that arrived while they were away. This exercise creates a small stream, publishes job-status updates, and gives each reader its own consumer.
Prerequisites
No. 1 A working NATS client connection
Complete Connect with the NATS CLI on CloudFerro Cloud. In the terminal used below, set NATS_URL and NATS_CREDS to the tested account endpoint and credentials path. Every command keeps --tlsfirst.
No. 2 Capacity and unused example names
Use a test account or names reserved for this exercise. It creates JOB_STATUS on jobs.status, with dashboard and reporting consumers. Check available capacity using Prepare your Message Streaming account on CloudFerro Cloud: the example requests 1 MiB of file storage, up to 1,000 messages, and 10 minutes of history. Reduce these values if the account’s limits are lower.
The message counts below assume a new stream, no other publishers or readers, and completion before its retention period expires. Do not delete an existing stream to reproduce these counts; choose unused names instead.
Plan the event flow
The exercise uses these resources:
Resource |
Example purpose |
|---|---|
Subject: jobs.status |
Workers publish status changes for processing jobs. |
Stream: JOB_STATUS |
Retain messages published to the job-status subject. |
Consumer: dashboard |
Read status events for the progress display. |
Consumer: reporting |
Read retained events to calculate job durations, with its own progress. |
Creating the account does not create these resources. The following commands do.
Create a bounded stream
List the account’s streams before using the example name:
nats --tlsfirst stream ls
Create the stream:
nats --tlsfirst stream add JOB_STATUS \
--subjects=jobs.status \
--storage=file --retention=limits --discard=old \
--max-age=10m --max-bytes=1MiB --max-msgs=1000 \
--replicas=1 --defaults
The command should report that JOB_STATUS was created and show Messages: 0. Check that the subject and limits match the command. Limits retention keeps events until an age, size, or count limit removes them; acknowledging a message does not remove it for the other reader.
This small exercise uses one replica. Replication is a stream setting managed through the client; it does not mean that the platform automatically copies resources between regions.
Create two independent readers
Create a named pull consumer for each application:
nats --tlsfirst consumer add JOB_STATUS dashboard \
--pull --deliver=all --ack=explicit \
--replay=instant --filter=jobs.status --defaults
nats --tlsfirst consumer add JOB_STATUS reporting \
--pull --deliver=all --ack=explicit \
--replay=instant --filter=jobs.status --defaults
Each consumer starts from all available messages and requests them when ready. Explicit acknowledgment lets the reader confirm which deliveries it has processed.
Publish a job starting and finishing
Send two updates for job-001:
nats --tlsfirst pub jobs.status \
'{"job_id":"job-001","status":"started"}' --jetstream
nats --tlsfirst pub jobs.status \
'{"job_id":"job-001","status":"finished"}' --jetstream
The --jetstream option waits for a storage acknowledgment. For a new stream, the two responses identify JOB_STATUS and sequence numbers 1 and 2. Check the stored count:
nats --tlsfirst stream info JOB_STATUS
The stream should now contain two messages.
Let the dashboard read first
Fetch and acknowledge the two events with the dashboard consumer:
nats --tlsfirst consumer next JOB_STATUS dashboard \
--count=2 --ack --raw
The payloads should appear in publication order:
{"job_id":"job-001","status":"started"}
{"job_id":"job-001","status":"finished"}
Inspect the other reader:
nats --tlsfirst consumer info JOB_STATUS reporting
Unprocessed Messages should still be 2. Reading with dashboard did not advance reporting.
Let the reporting reader catch up
Leave the reporting consumer idle and publish one more event:
nats --tlsfirst pub jobs.status \
'{"job_id":"job-002","status":"started"}' --jetstream
nats --tlsfirst consumer next JOB_STATUS dashboard \
--count=1 --ack --raw
The dashboard receives only the new event. Now let the reporting consumer request the three messages it has not read:
nats --tlsfirst consumer next JOB_STATUS reporting \
--count=3 --ack --raw
nats --tlsfirst consumer info JOB_STATUS reporting
It should receive both job-001 updates followed by the job-002 update. Its Unprocessed Messages and Outstanding Acks should then be 0. The stream still holds the three events until retention removes them.
These CLI commands acknowledge messages as they are received for demonstration. Application code should acknowledge a delivery after its processing succeeds.
Retain and recover event history
A consumer can catch up after an interruption only while the messages it needs remain in the stream. Purging messages or reaching a retention limit can remove that history. Logical backups and restores are client operations; plan them separately from the account’s portal configuration.
The CLI and SDKs also support editing stream settings, deleting consumers, purging streams, and inspecting messages. My Portal manages the account and credentials rather than those stream operations. Keep JOB_STATUS and its consumers for the integration and monitoring examples that follow.
What to do next
The NATS stream guide and pull consumer guide explain the underlying behavior in more detail.
Use Message Streaming with other products on CloudFerro Cloud describes application patterns with other products. For account usage and management actions, continue with Monitor and remove a NATS account on CloudFerro Cloud.