client

package
v0.0.0-...-989ae23 Latest Latest
Warning

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

 Go to latest Published: Aug 16, 2026 License: MIT Imports: 14 Imported by: 1

Documentation

Overview

Package client provides a NATS JetStream client with connection management, KV stores, KV-backed streams, and consumer helpers.

Index

Constants

This section is empty.

Variables

View Source 
var GetJetStream = func(nc *nats.Conn) (jetstream.JetStream, error) {
	return jetstream.New(nc)
}

GetJetStream is a public variable function wrapping jetstream.New.

Functions

This section is empty.

Types

type AuthOptions 

type AuthOptions struct {
	// AuthType specifies the authentication method to use (NoAuth, UserPassAuth, or NKeyAuth).
	AuthType AuthType
	// Username is required for UserPassAuth and represents the NATS username.
	Username string
	// Password is required for UserPassAuth and represents the NATS password.
	Password string
	// NKeyFile is required for NKeyAuth and specifies the path to the NKEY private seed file.
	// This file should contain an Ed25519 private key (starting with "S").
	NKeyFile string
}

AuthOptions holds authentication-related settings for connecting to NATS.

type AuthType 

type AuthType int

AuthType defines the different authentication methods supported by the client. 

const (
	// NoAuth represents a connection with no authentication.
	NoAuth AuthType = iota
	// UserPassAuth represents authentication using a username and password.
	UserPassAuth
	// NKeyAuth represents authentication using NATS NKEYs (Ed25519 public-private key pairs).
	NKeyAuth
)

type Client 

type Client struct {

	// NC underlying connection.
	NC NATSConnector
	// ExtJS is the extended JetStream API for high-level operations (e.g. retrieving streams/consumers).
	ExtJS jetstream.JetStream
	// Opts configuration options used to create the client
	Opts *Options
	// KeyPair allows injecting a mock `nkeys.KeyPair` for testing authentication logic.
	KeyPair nkeys.KeyPair
	// contains filtered or unexported fields
}

Client provides an implementation for interacting with an embedded NATS server.

func New 

func New(
	logger *slog.Logger,
	opts *Options,
) *Client

New creates a new Client instance using the provided logger and Options.

func (*Client) Close 

func (c *Client) Close()

Close closes the underlying NATS connection.

func (*Client) Connect 

func (c *Client) Connect() error

Connect establishes the connection to the NATS server and JetStream context. This method returns an error if there are any issues during connection.

func (*Client) ConnectedServerVersion 

func (c *Client) ConnectedServerVersion() string

ConnectedServerVersion returns the version of the connected NATS server. Returns an empty string if not connected or the version is unavailable.

func (*Client) ConnectedURL 

func (c *Client) ConnectedURL() string

ConnectedURL returns the URL of the NATS server the client is connected to. Returns an empty string if not connected.

func (*Client) ConsumeMessages 

func (c *Client) ConsumeMessages(
	ctx context.Context,
	streamName string,
	consumerName string,
	handler JetStreamMessageHandler,
	opts *ConsumeOptions,
) error

ConsumeMessages subscribes to a JetStream consumer and processes messages with the provided handler. This provides a clean abstraction for message consumption with proper context handling.

func (*Client) CreateOrUpdateConsumerWithConfig 

func (c *Client) CreateOrUpdateConsumerWithConfig(
	ctx context.Context,
	streamName string,
	consumerConfig jetstream.ConsumerConfig,
) error

CreateOrUpdateConsumerWithConfig creates or updates a JetStream consumer with the provided configuration.

func (*Client) CreateOrUpdateJetStreamWithConfig 

func (c *Client) CreateOrUpdateJetStreamWithConfig(
	ctx context.Context,
	streamConfig jetstream.StreamConfig,
	consumerConfigs ...jetstream.ConsumerConfig,
) error

CreateOrUpdateJetStreamWithConfig configures a JetStream stream and its consumers. This is a convenience method that creates both stream and consumers in one call.

func (*Client) CreateOrUpdateKVBucket 

func (c *Client) CreateOrUpdateKVBucket(
	ctx context.Context,
	bucketName string,
) (jetstream.KeyValue, error)

CreateOrUpdateKVBucket creates or updates a KV bucket using the jetstream API, which natively supports upsert semantics. Returns the jetstream.KeyValue interface.

func (*Client) CreateOrUpdateKVBucketWithConfig 

func (c *Client) CreateOrUpdateKVBucketWithConfig(
	ctx context.Context,
	config jetstream.KeyValueConfig,
) (jetstream.KeyValue, error)

CreateOrUpdateKVBucketWithConfig creates or updates a KV bucket with the provided configuration.

NATS does not allow changing the storage type on an existing stream. If CreateOrUpdateKeyValue fails with a "can not change storage type" error, this method retries without the storage field so that other mutable settings (TTL, MaxBytes, Replicas) are still applied.

func (*Client) CreateOrUpdateObjectStore 

func (c *Client) CreateOrUpdateObjectStore(
	ctx context.Context,
	cfg jetstream.ObjectStoreConfig,
) (jetstream.ObjectStore, error)

CreateOrUpdateObjectStore creates or updates a NATS Object Store bucket with the provided configuration.

NATS does not allow changing the storage type on an existing stream. If CreateOrUpdateObjectStore fails with a "can not change storage type" error, this method retries without the storage field.

func (*Client) CreateOrUpdateStreamWithConfig 

func (c *Client) CreateOrUpdateStreamWithConfig(
	ctx context.Context,
	streamConfig jetstream.StreamConfig,
) error

CreateOrUpdateStreamWithConfig creates or updates a JetStream stream with the provided configuration.

NATS does not allow changing the storage type on an existing stream. If CreateOrUpdateStream fails with a "can not change storage type" error, this method retries without the storage field so that other mutable settings are still applied.

func (*Client) GetStreamInfo 

func (c *Client) GetStreamInfo(
	ctx context.Context,
	streamName string,
) (*jetstream.StreamInfo, error)

GetStreamInfo retrieves information about a JetStream stream.

func (*Client) KVDelete 

func (c *Client) KVDelete(
	bucket string,
	key string,
) error

KVDelete removes a key from the specified KV bucket.

func (*Client) KVGet 

func (c *Client) KVGet(
	bucket string,
	key string,
) ([]byte, error)

KVGet retrieves a value from the specified KV bucket.

func (*Client) KVKeys 

func (c *Client) KVKeys(
	bucket string,
) ([]string, error)

KVKeys returns all keys from the specified KV bucket.

func (*Client) KVPut 

func (c *Client) KVPut(
	bucket string,
	key string,
	value []byte,
) error

KVPut stores a value in the specified KV bucket.

func (*Client) KVPutAndPublish 

func (c *Client) KVPutAndPublish(
	ctx context.Context,
	kvBucket string,
	key string,
	data []byte,
	notifySubject string,
) (uint64, error)

KVPutAndPublish implements the pattern of storing data in KV and sending a notification via stream. This is useful for workflow systems where you want persistent storage + event notification.

func (*Client) KeyValue 

func (c *Client) KeyValue(
	ctx context.Context,
	bucket string,
) (jetstream.KeyValue, error)

KeyValue returns a handle to a JetStream Key-Value bucket by name.

func (*Client) ObjectStore 

func (c *Client) ObjectStore(
	ctx context.Context,
	name string,
) (jetstream.ObjectStore, error)

ObjectStore returns an existing NATS Object Store handle by name.

func (*Client) Publish 

func (c *Client) Publish(
	ctx context.Context,
	subject string,
	data []byte,
) error

Publish publishes a message to a JetStream subject. If the context carries an OpenTelemetry span, the trace context is automatically propagated via NATS message headers.

func (*Client) PublishAndWaitKV 

func (c *Client) PublishAndWaitKV(
	ctx context.Context,
	subject string,
	data []byte,
	kvBucket jetstream.KeyValue,
	opts *RequestReplyOptions,
) ([]byte, error)

PublishAndWaitKV publishes a message and waits for a response in a KV bucket. This implements an async request/reply pattern using KV for response storage.

func (*Client) PublishCore 

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

PublishCore publishes a message using core NATS (not JetStream). Unlike Publish, this does not require JetStream stream routing and does not inject OpenTelemetry trace headers. Use this for simple fire-and-forget messaging like enrollment responses and key rotation.

func (*Client) Stream 

func (c *Client) Stream(
	ctx context.Context,
	name string,
) (jetstream.Stream, error)

Stream returns a handle to a JetStream stream by name.

func (*Client) Subscribe 

func (c *Client) Subscribe(
	subject string,
	handler nats.MsgHandler,
) (*nats.Subscription, error)

Subscribe creates a core NATS subscription on the given subject. Messages are delivered to the handler callback. This uses plain NATS pub/sub (not JetStream) — suitable for fire-and-forget messaging like enrollment requests and key rotation notifications.

func (*Client) WatchKV 

func (c *Client) WatchKV(
	ctx context.Context,
	kv jetstream.KeyValue,
	pattern string,
) (<-chan jetstream.KeyValueEntry, error)

WatchKV watches a KV bucket for responses matching a pattern. This is useful for collecting responses from multiple workers.

Note: Test coverage for this function is intentionally limited to error paths and setup logic. The goroutine's event forwarding logic is not covered due to the complexity of testing async behavior without introducing flaky tests.

type ConsumeOptions 

type ConsumeOptions struct {
	// QueueGroup for load balancing across multiple consumers (optional)
	QueueGroup string
	// MaxInFlight limits the number of unacknowledged messages
	MaxInFlight int
}

ConsumeOptions configures message consumption behavior.

type JetStreamMessageHandler 

type JetStreamMessageHandler func(msg jetstream.Msg) error

JetStreamMessageHandler defines the signature for JetStream message handling functions.

type NATSConnWrapper 

type NATSConnWrapper struct {
	Conn *nats.Conn
}

NATSConnWrapper is a concrete implementation of NATSConnector, wrapping a *nats.Conn.

func (*NATSConnWrapper) Close 

func (n *NATSConnWrapper) Close()

Close wraps the Close method of nats.Conn.

func (*NATSConnWrapper) Connect 

func (n *NATSConnWrapper) Connect(
	url string,
	opts ...nats.Option,
) (*nats.Conn, error)

Connect establishes a connection to the NATS server with the given URL and options.

func (*NATSConnWrapper) ConnectedUrl 

func (n *NATSConnWrapper) ConnectedUrl() string

ConnectedUrl wraps the ConnectedUrl method of nats.Conn.

func (*NATSConnWrapper) JetStream 

func (n *NATSConnWrapper) JetStream(
	opts ...nats.JSOpt,
) (nats.JetStreamContext, error)

JetStream wraps the JetStream method of nats.Conn.

func (*NATSConnWrapper) Publish 

func (n *NATSConnWrapper) Publish(
	subject string,
	data []byte,
) error

Publish wraps the core NATS Publish method of nats.Conn.

func (*NATSConnWrapper) QueueSubscribe 

func (n *NATSConnWrapper) QueueSubscribe(
	subject, queue string,
	handler nats.MsgHandler,
) (*nats.Subscription, error)

QueueSubscribe wraps the QueueSubscribe method of nats.Conn.

func (*NATSConnWrapper) Subscribe 

func (n *NATSConnWrapper) Subscribe(
	subject string,
	handler nats.MsgHandler,
) (*nats.Subscription, error)

Subscribe wraps the Subscribe method of nats.Conn.

type NATSConnector 

type NATSConnector interface {
	JetStream(opts ...nats.JSOpt) (nats.JetStreamContext, error)
	Close()
	ConnectedUrl() string
	Connect(url string, opts ...nats.Option) (*nats.Conn, error)
	Subscribe(subject string, handler nats.MsgHandler) (*nats.Subscription, error)
	QueueSubscribe(subject, queue string, handler nats.MsgHandler) (*nats.Subscription, error)
	Publish(subject string, data []byte) error
}

NATSConnector defines an interface for managing a NATS connection.

type Options 

type Options struct {
	// Host specifies the NATS server hostname or IP address.
	Host string
	// Port specifies the NATS server port.
	Port int
	// Auth contains authentication settings for connecting to the NATS server.
	Auth AuthOptions
	// Name is a human-readable name for this client connection.
	Name string
}

Options holds the configuration for connecting to a NATS server, including connection details and authentication settings.

type RequestReplyOptions 

type RequestReplyOptions struct {
	// RequestID to use (default: generated UUID)
	RequestID string
	// Timeout for waiting for response (default: 30s)
	Timeout time.Duration
	// PollInterval for checking KV store (default: 100ms)
	PollInterval time.Duration
}

RequestReplyOptions configures request/reply behavior.

Source Files

View all Source files

Directories

Path Synopsis
Package mocks provides generated mock implementations for testing.
 Package mocks provides generated mock implementations for testing.

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.