nats

package module
v0.0.0-...-c973774 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 17, 2026 License: BSD-3-Clause Imports: 3 Imported by: 0

README

go-ruby-nats/nats

nats — go-ruby-nats

Docs License Go Coverage

A pure-Go (no cgo) port of the Ruby nats-pure / nats client's public surface — the messaging client Ruby programs use to talk to a NATS server.

The Ruby client speaks the NATS protocol over a socket it owns. This module keeps the same object model — NATS::Client, NATS::Msg, NATS::Subscription and the NATS::Error tree — but delegates the wire protocol to the official pure-Go client nats.go. It does not reimplement the protocol, and it links statically with CGO_ENABLED=0 on every 64-bit target the go-* ecosystem supports (amd64, arm64, riscv64, loong64, ppc64le and big-endian s390x).

It is the NATS backend for go-embedded-ruby, but is a standalone, reusable module — a sibling of go-ruby-redis and go-ruby-sqlite3.

Example

nc, _ := nats.Connect(nats.Servers("nats://127.0.0.1:4222"))
defer nc.Close()

// Subscribe with a responder.
nc.Subscribe("greet.*", func(m *nats.Msg) {
    m.Respond([]byte("hello, " + string(m.Data)))
})

// Publish.
nc.Publish("greet.world", []byte("world"))

// Request/reply.
rep, _ := nc.Request("greet.joe", []byte("joe"), time.Second)
fmt.Println(string(rep.Data)) // hello, joe

nc.Drain()

Surface

Faithful port of the gem's client API:

  • Client — Connect (NATS.connect), Publish / PublishRequest / PublishMsg, Subscribe / QueueSubscribe (subscribe(queue:)), Unsubscribe, Request / RequestMsg, Flush / FlushTimeout, Drain, Close, IsClosed, and the connection callbacks OnError, OnReconnect, OnDisconnect, OnClose.
  • Msg — Subject, Data, Reply, Header / Headers, Respond / RespondMsg.
  • Subscription — Unsubscribe, AutoUnsubscribe, Drain.
  • Header — Get / Values / Set / Add / Del.
  • Error tree — ErrTimeout (NATS::IO::Timeout), ErrNoResponders (NoRespondersError), ErrConnectionClosed, ErrConnectionDraining, ErrBadSubscription, ErrBadSubject, ErrNoServers, ErrMaxPayload, … — each a sentinel matched with errors.Is.

Connection options mirror the gem's NATS.connect keywords: Servers, Name, UserInfo, Token, MaxReconnects, ReconnectWait, Timeout, and the *Callback registrations. NATSOption threads a raw nats.go option through for transport features (TLS, JWT, in-process server) not surfaced by the gem-shaped options.

Host seam

The transport is injectable. Connect dials a real server through nats.go by default, but the dialer is a seam, so the whole client can run against an in-process broker or an embedded nats-server started with nats.InProcessServer — no external socket in unit tests.

Tests & coverage

go test ./... runs two suites, both verifying real round-trips (delivery, queue groups, request/reply, timeout, unsubscribe, drain, headers) rather than asserting on internals:

  • a deterministic in-process broker that drives the client with no socket and runs on every arch (including under qemu), and
  • an embedded nats-server (in-process, DontListen) that validates the real nats.go adapter end to end on the native lanes.

Coverage is 100% of statements, enforced in CI with -race across three OSes and all six 64-bit arches (CGO_ENABLED=0, s390x big-endian under qemu).

License

BSD-3-Clause — see LICENSE. Copyright (c) 2026, the go-ruby-nats/nats authors.

WebAssembly

Being pure Go (CGO=0), this library also compiles to WebAssembly — both GOOS=js GOARCH=wasm (browser / Node.js) and GOOS=wasip1 GOARCH=wasm (WASI). CI builds both targets on every push, alongside the six 64-bit native/qemu arches.

GOOS=js     GOARCH=wasm go build ./...   # browser / Node
GOOS=wasip1 GOARCH=wasm go build ./...   # WASI (wasmtime, wasmer, wasmedge, …)

Documentation

Overview

Package nats is a pure-Go (CGO_ENABLED=0) port of the Ruby nats-pure / nats client's public surface — the messaging client Ruby programs use to talk to a NATS server.

Upstream, the Ruby client speaks the NATS protocol over a socket it owns; this module keeps the same object model — Client, Msg, Subscription and the [Error] tree — but delegates the wire protocol to the official pure-Go client github.com/nats-io/nats.go. It does not reimplement the protocol.

MRI-faithful surface 

nc, _ := nats.Connect(nats.Servers("nats://127.0.0.1:4222"))
sub, _ := nc.Subscribe("greet.*", func(m *nats.Msg) {
	m.Respond([]byte("hello, " + string(m.Data)))
})
rep, _ := nc.Request("greet.joe", []byte("joe"), time.Second)
fmt.Println(string(rep.Data)) // hello, joe
sub.Unsubscribe()
nc.Drain()
nc.Close()

The method names mirror the gem: Client.Publish, Client.Subscribe / Client.QueueSubscribe, Client.Request, Client.Flush, Client.Drain, Client.Close, and the connection callbacks Client.OnError, Client.OnReconnect and Client.OnDisconnect. Msg exposes Subject / Data / Reply / Header and Msg.Respond; errors map to the gem's classes (ErrTimeout is NATS::IO::Timeout, ErrNoResponders is NoRespondersError).

Host seam

The transport is injectable. Connect dials a real server through nats.go by default, but the dialer is a seam: tests drive the whole client against an in-process broker (no socket) or an embedded nats-server started with github.com/nats-io/nats.go.InProcessServer. The core opens no external socket in unit tests.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrConnectionClosed is returned when an operation is attempted on a closed
	// connection (NATS::IO::ConnectionClosedError).
	ErrConnectionClosed = errors.New("nats: connection closed")

	// ErrConnectionDraining is returned when publishing on a draining connection
	// (NATS::IO::ConnectionDrainingError).
	ErrConnectionDraining = errors.New("nats: connection draining")

	// ErrConnectionReconnecting is returned while the connection is reconnecting
	// (NATS::IO::ConnectionReconnectingError).
	ErrConnectionReconnecting = errors.New("nats: connection reconnecting")

	// ErrTimeout is returned when a request or flush does not complete in time
	// (NATS::IO::Timeout).
	ErrTimeout = errors.New("nats: timeout")

	// ErrNoResponders is returned by a request when no subscriber is listening on
	// the subject (NATS::IO::NoRespondersError).
	ErrNoResponders = errors.New("nats: no responders available for request")

	// ErrBadSubscription is returned for an invalid or already-removed
	// subscription (NATS::IO::BadSubscription).
	ErrBadSubscription = errors.New("nats: invalid subscription")

	// ErrBadSubject is returned for an empty or malformed subject
	// (NATS::IO::BadSubject).
	ErrBadSubject = errors.New("nats: invalid subject")

	// ErrNoServers is returned when no server is reachable at connect time
	// (NATS::IO::NoServersError).
	ErrNoServers = errors.New("nats: no servers available for connection")

	// ErrMaxPayload is returned when a message exceeds the server's max payload
	// (NATS::IO::MaxPayloadError).
	ErrMaxPayload = errors.New("nats: maximum payload exceeded")

	// ErrInvalidCallback is returned when a subscription is created without a
	// message-handler block.
	ErrInvalidCallback = errors.New("nats: invalid callback")

	// ErrNoReplySubject is returned by [Msg.Respond] when the message carries no
	// reply subject to answer on.
	ErrNoReplySubject = errors.New("nats: message has no reply subject")
)

The error tree mirrors the Ruby client's exception classes. Upstream they are Ruby classes under NATS::Error / NATS::IO::*; here each is a sentinel value callers match with errors.Is. [mapError] translates the errors nats.go returns into these, so downstream code — and the Ruby host that raises the matching class — sees a stable set independent of the nats.go version.

Functions

This section is empty.

Types

type Client 

type Client struct {
	// contains filtered or unexported fields
}

Client is a NATS connection, mirroring NATS::Client. It publishes and subscribes over a transport [conn] and does not own the socket itself — the dialer does. Construct one with Connect.

func Connect 

func Connect(opts ...Option) (*Client, error)

Connect opens a connection to a NATS server, mirroring NATS.connect. The transport is a real nats.go connection; configure it with Option values such as Servers, Name and UserInfo.

func (*Client) Close 

func (c *Client) Close()

Close closes the connection immediately, mirroring NATS::Client#close.

func (*Client) Drain 

func (c *Client) Drain() error

Drain unsubscribes all subscriptions, flushes pending messages and closes the connection, mirroring NATS::Client#drain.

func (*Client) Flush 

func (c *Client) Flush() error

Flush waits for the server to process all buffered messages, mirroring NATS::Client#flush.

func (*Client) FlushTimeout 

func (c *Client) FlushTimeout(timeout time.Duration) error

FlushTimeout is Client.Flush bounded by timeout, returning ErrTimeout if it does not complete in time.

func (*Client) IsClosed 

func (c *Client) IsClosed() bool

IsClosed reports whether the connection is closed, mirroring NATS::Client#closed?.

func (*Client) OnClose 

func (c *Client) OnClose(cb func())

OnClose registers a callback fired when the connection closes, mirroring NATS::Client#on_close.

func (*Client) OnDisconnect 

func (c *Client) OnDisconnect(cb func())

OnDisconnect registers a callback fired when the client disconnects, mirroring NATS::Client#on_disconnect.

func (*Client) OnError 

func (c *Client) OnError(cb func(error))

OnError registers a callback for asynchronous errors, mirroring NATS::Client#on_error.

func (*Client) OnReconnect 

func (c *Client) OnReconnect(cb func())

OnReconnect registers a callback fired after the client reconnects, mirroring NATS::Client#on_reconnect.

func (*Client) Publish 

func (c *Client) Publish(subject string, data []byte) error

Publish sends data to subject, mirroring NATS::Client#publish.

func (*Client) PublishMsg 

func (c *Client) PublishMsg(m *Msg) error

PublishMsg publishes a fully-formed Msg (carrying Reply and Header), mirroring NATS::Client#publish_msg.

func (*Client) PublishRequest 

func (c *Client) PublishRequest(subject, reply string, data []byte) error

PublishRequest publishes data to subject with reply set as the response subject, mirroring NATS::Client#publish(subject, data, reply:).

func (*Client) QueueSubscribe 

func (c *Client) QueueSubscribe(subject, queue string, cb func(*Msg)) (*Subscription, error)

QueueSubscribe registers a queue-group subscription: the server load-balances matching messages across members of queue, mirroring NATS::Client#subscribe(subject, queue:).

func (*Client) Request 

func (c *Client) Request(subject string, data []byte, timeout time.Duration) (*Msg, error)

Request publishes data to subject and waits up to timeout for a single reply, mirroring NATS::Client#request. It returns ErrNoResponders if nobody is listening and ErrTimeout if no reply arrives in time.

func (*Client) RequestMsg 

func (c *Client) RequestMsg(m *Msg, timeout time.Duration) (*Msg, error)

RequestMsg is Client.Request for a pre-built Msg (carrying Header), mirroring NATS::Client#request with a message.

func (*Client) Subscribe 

func (c *Client) Subscribe(subject string, cb func(*Msg)) (*Subscription, error)

Subscribe registers interest in subject and delivers each matching message to cb, mirroring NATS::Client#subscribe(subject) { |msg| }.

func (*Client) Unsubscribe 

func (c *Client) Unsubscribe(sub *Subscription) error

Unsubscribe removes sub, mirroring NATS::Client#unsubscribe. 

type Header map[string][]string

Header carries a message's optional headers, mirroring NATS::Msg#headers — a case-preserving multimap of string keys to string values (the same shape as nats.go's Header and net/textproto's MIMEHeader). A nil Header is a valid empty header for reads.

func (Header) Add 

func (h Header) Add(key, value string)

Add appends a value to the values for key.

func (Header) Del 

func (h Header) Del(key string)

Del removes all values for key.

func (Header) Get 

func (h Header) Get(key string) string

Get returns the first value for key, or "" if absent.

func (Header) Set 

func (h Header) Set(key, value string)

Set replaces any existing values for key with a single value.

func (Header) Values 

func (h Header) Values(key string) []string

Values returns all values for key (nil if absent).

type Msg 

type Msg struct {
	// Subject the message was published to (or delivered on).
	Subject string
	// Reply is the subject a responder should answer on, if any.
	Reply string
	// Data is the message payload.
	Data []byte
	// Header holds optional headers.
	Header Header
	// Sub is the subscription that delivered this message (nil for messages the
	// caller builds).
	Sub *Subscription
	// contains filtered or unexported fields
}

Msg is a NATS message, mirroring NATS::Msg. Subscribers receive one; publishers may build one for Client.PublishMsg to carry a Reply subject or Header.

func (*Msg) Headers 

func (m *Msg) Headers() Header

Headers returns the message headers, mirroring NATS::Msg#headers.

func (*Msg) Respond 

func (m *Msg) Respond(data []byte) error

Respond publishes data to the message's Reply subject, mirroring NATS::Msg#respond. It returns ErrNoReplySubject if the message has no reply subject and ErrConnectionClosed if it is not bound to a connection.

func (*Msg) RespondMsg 

func (m *Msg) RespondMsg(resp *Msg) error

RespondMsg publishes resp (with its Header) to the message's Reply subject, mirroring NATS::Msg#respond_msg.

type Option 

type Option func(*options)

Option configures a Connect call. The provided options mirror the gem's NATS.connect keyword arguments.

func ClosedCallback 

func ClosedCallback(cb func()) Option

ClosedCallback registers the on_close callback at connect time.

func DisconnectCallback 

func DisconnectCallback(cb func()) Option

DisconnectCallback registers the on_disconnect callback at connect time.

func ErrorCallback 

func ErrorCallback(cb func(error)) Option

ErrorCallback registers the on_error callback at connect time.

func MaxReconnects 

func MaxReconnects(n int) Option

MaxReconnects sets the reconnect attempt limit (NATS.connect max_reconnect_attempts:).

func NATSOption 

func NATSOption(opt natsgo.Option) Option

NATSOption threads a raw nats.go github.com/nats-io/nats.go.Option into the underlying connection — an escape hatch for transport features (TLS, JWT, in-process server) not surfaced by the gem-shaped options above.

func Name 

func Name(name string) Option

Name sets the connection name (NATS.connect name:).

func ReconnectCallback 

func ReconnectCallback(cb func()) Option

ReconnectCallback registers the on_reconnect callback at connect time.

func ReconnectWait 

func ReconnectWait(d time.Duration) Option

ReconnectWait sets the delay between reconnect attempts (NATS.connect reconnect_time_wait:).

func Servers 

func Servers(urls ...string) Option

Servers sets the server URLs to connect to (NATS.connect servers:).

func Timeout 

func Timeout(d time.Duration) Option

Timeout sets the connect timeout (NATS.connect connect_timeout:).

func Token 

func Token(token string) Option

Token sets a token credential (NATS.connect auth_token:).

func UserInfo 

func UserInfo(user, password string) Option

UserInfo sets username/password credentials (NATS.connect user:/password:).

type Subscription 

type Subscription struct {
	// Subject is the subscribed subject (may contain wildcards).
	Subject string
	// Queue is the queue group, or "" for a plain subscription.
	Queue string
	// contains filtered or unexported fields
}

Subscription is an interest registration, mirroring NATS::Subscription. It is returned by Client.Subscribe / Client.QueueSubscribe and delivers messages to the callback given there.

func (*Subscription) AutoUnsubscribe 

func (s *Subscription) AutoUnsubscribe(max int) error

AutoUnsubscribe removes the subscription after max more messages are delivered, mirroring the gem's subscribe(max:) auto-unsubscribe.

func (*Subscription) Drain 

func (s *Subscription) Drain() error

Drain removes the subscription after its buffered messages are processed, mirroring NATS::Subscription#drain.

func (*Subscription) Unsubscribe 

func (s *Subscription) Unsubscribe() error

Unsubscribe removes the subscription, mirroring NATS::Subscription#unsubscribe.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.