nats

NATS Core Knowledge

Full Reference: See advanced.md for JetStream patterns (Node.js, Java), security configuration, TLS setup, and clustering.

Deep Knowledge: Use mcp__documentation__fetch_docs with technology: nats for comprehensive documentation.

Quick Start (Docker)

# docker-compose.yml
services:
  nats:
    image: nats:latest
    ports:
      - "4222:4222"   # Client
      - "8222:8222"   # Monitoring
    command: "-js -m 8222"  # Enable JetStream

Core Concepts

Feature	Core NATS	JetStream
Persistence	No	Yes
Delivery	At-most-once	At-least-once
Replay	No	Yes
Consumer Groups	Queue Groups	Consumer Groups

Subject Patterns

orders.created     → Specific subject
orders.*           → orders.created, orders.updated (single wildcard)
orders.>           → orders.created, orders.us.east (multi wildcard)

Core NATS (Node.js)

import { connect, StringCodec } from 'nats';

const nc = await connect({ servers: 'localhost:4222' });
const sc = StringCodec();

// Simple publish
nc.publish('orders.created', sc.encode(JSON.stringify(order)));

// Subscribe
const sub = nc.subscribe('orders.*');
for await (const msg of sub) {
  const order = JSON.parse(sc.decode(msg.data));
  console.log(`Received on ${msg.subject}:`, order);
}

// Queue group (load balancing)
const qsub = nc.subscribe('orders.process', { queue: 'workers' });

// Request/Reply
const response = await nc.request('orders.validate', sc.encode(JSON.stringify(order)), {
  timeout: 5000,
});
const result = JSON.parse(sc.decode(response.data));

await nc.drain();

Core NATS (Python)

import nats
import json
import asyncio

async def main():
    nc = await nats.connect("nats://localhost:4222")

    # Publish
    await nc.publish("orders.created", json.dumps(order).encode())

    # Subscribe
    async def message_handler(msg):
        order = json.loads(msg.data.decode())
        print(f"Received: {order}")

    await nc.subscribe("orders.*", cb=message_handler)

    # Request/Reply
    response = await nc.request("orders.validate",
                                json.dumps(order).encode(),
                                timeout=5)
    result = json.loads(response.data.decode())

    await nc.drain()

asyncio.run(main())

Core NATS (Go)

nc, _ := nats.Connect(nats.DefaultURL)
defer nc.Drain()

// Publish
body, _ := json.Marshal(order)
nc.Publish("orders.created", body)

// Subscribe
nc.Subscribe("orders.*", func(msg *nats.Msg) {
    var order Order
    json.Unmarshal(msg.Data, &order)
    processOrder(order)
})

// Queue group
nc.QueueSubscribe("orders.process", "workers", func(msg *nats.Msg) {
    // Load balanced
})

// Request/Reply
response, _ := nc.Request("orders.validate", body, 5*time.Second)

When NOT to Use This Skill

• Long-term event replay required - Kafka provides better retention
• Complex routing patterns - RabbitMQ exchanges are more flexible
• JMS compliance needed - Use ActiveMQ
• AWS-native integration - SQS integrates better

Anti-Patterns

Anti-Pattern	Why It's Bad	Solution
Core NATS for critical data	Fire-and-forget	Use JetStream for persistence
No subject hierarchy	Hard to filter	Use dot-separated subjects
No timeout on requests	Hanging requests	Always set timeout
Large payloads	Network strain	Keep messages small
Single server in production	No HA	Deploy 3+ server cluster

Quick Troubleshooting

Issue	Likely Cause	Fix
Messages not received	Wrong subject	Verify subject name, check wildcards
Request timeout	No responders	Verify responders exist
JetStream errors	Not enabled	Start server with -js flag
Consumer lag growing	Slow processing	Add consumers or optimize
ACK timeout	Processing too slow	Increase ack_wait

Production Checklist

• TLS enabled
• Authentication configured
• Authorization (permissions)
• JetStream replicas >= 3
• Stream limits configured
• Consumer max_deliver set
• Ack timeout configured
• Monitoring dashboards
• Cluster health checks

Reference Documentation

Available topics: basics, jetstream, patterns, production