nats

README

= NATS High-Level Wrapper & Proxy for Golang

== Overview

This repository provides a high-level wrapper for the https://github.com/nats-io/nats.go[official NATS Go client], enabling scalable and efficient message processing. It dynamically adjusts the number of goroutines to optimize performance by increasing worker count under load and reducing idle workers automatically.

Additionally, the repository includes a proxy implementation that balances messages between pods based on a given rule, ensuring efficient distribution across multiple instances.

== Features

* *Automatic Scaling of Workers:* Adjusts goroutine count based on message load.
* *High-Level API:* Simplifies integration with NATS Core.
* *Proxy Implementation:* Balances messages across pods based on predefined rules.
* *Efficient Resource Utilization:* Reduces idle workers to conserve CPU and memory.
* *Resilient and Fault-Tolerant:* Designed for high availability in distributed environments.

Documentation

Overview

Package nats provides NATS publishing and subscription helpers.

New code should use core-owned message and boundary contracts at consumer edges:

import "github.com/InsideGallery/core/queue/nats"

publisher := nats.NewPublisher(legacyPublisher)
result, err := publisher.Publish(ctx, nats.PublishOptions{
	Message: nats.Message{Subject: "events"},
})

Prefer Publisher, Subscriber, Message, Headers, PublishOptions, RequestOptions, SubscribeOptions, and their result types where application code should not depend on the NATS SDK message type.

Compatibility: the queue/nats/client, publisher, and subscriber sub-packages remain available for existing SDK-shaped call sites. Prefer ConnectClient and the adapters in this package for new code.

Index

• Variables
• type Headers
• type Message
• type MessageHandler
• type PublishOptions
• type PublishResult
• type Publisher
• type PublisherAdapter
  • func NewPublisher(p *publisher.Publisher) *PublisherAdapter
  • func (p *PublisherAdapter) Publish(ctx context.Context, options PublishOptions) (PublishResult, error)
  • func (p *PublisherAdapter) Request(ctx context.Context, options RequestOptions) (RequestResult, error)
• type RequestOptions
• type RequestResult
• type SubscribeOptions
• type SubscribeResult
• type Subscriber
• type SubscriberAdapter
  • func NewSubscriber(s *legacysubscriber.Subscriber) *SubscriberAdapter
  • func (s *SubscriberAdapter) Close(ctx context.Context) error
  • func (s *SubscriberAdapter) Subscribe(ctx context.Context, options SubscribeOptions, handler MessageHandler) (SubscribeResult, error)
  • func (s *SubscriberAdapter) Wait(ctx context.Context) error

Variables

var ErrHandlerNotSet = errors.New("handler is not set")

ErrHandlerNotSet reports a missing subscriber handler.

var ErrMissingSubject = errors.New("subject is not set")

ErrMissingSubject reports a message without a subject.

var ErrPublisherNotSet = errors.New("publisher is not set")

ErrPublisherNotSet reports a missing publisher dependency.

var ErrSubscriberNotSet = errors.New("subscriber is not set")

ErrSubscriberNotSet reports a missing subscriber dependency.

Types

type Headers

type Headers map[string][]string

Headers is the core-owned queue header shape.

type Message

type Message struct {
	Subject string
	Reply   string
	Data    []byte
	Header  Headers
}

Message is the core-owned queue message shape.

type MessageHandler

type MessageHandler func(ctx context.Context, msg Message) error

MessageHandler handles a message without exposing the NATS SDK message type.

type PublishOptions

type PublishOptions struct {
	Message Message
	Timeout time.Duration
}

PublishOptions is the core-owned input for publishing a message.

type PublishResult

type PublishResult struct {
	Subject string
	Bytes   int
}

PublishResult is the core-owned publish result.

type Publisher

type Publisher interface {
	Publish(ctx context.Context, options PublishOptions) (PublishResult, error)
	Request(ctx context.Context, options RequestOptions) (RequestResult, error)
}

Publisher is the core-owned NATS publishing contract for new consumers.

type PublisherAdapter

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

PublisherAdapter adapts the legacy NATS publisher to the core-owned Publisher contract.

func NewPublisher

func NewPublisher(p *publisher.Publisher) *PublisherAdapter

NewPublisher wraps a legacy NATS publisher with the core-owned Publisher contract.

func (*PublisherAdapter) Publish

func (p *PublisherAdapter) Publish(ctx context.Context, options PublishOptions) (PublishResult, error)

Publish publishes a message through the core-owned Publisher contract.

func (*PublisherAdapter) Request

func (p *PublisherAdapter) Request(ctx context.Context, options RequestOptions) (RequestResult, error)

Request sends a request and returns a core-owned response message.

type RequestOptions

type RequestOptions struct {
	Message Message
}

RequestOptions is the core-owned input for request/reply messaging.

type RequestResult

type RequestResult struct {
	Message Message
}

RequestResult is the core-owned request/reply result.

type SubscribeOptions

type SubscribeOptions struct {
	Subject string
	Queue   string
	Buffer  int
	Timeout time.Duration
}

SubscribeOptions is the core-owned input for NATS queue subscriptions.

type SubscribeResult

type SubscribeResult struct {
	Subject string
	Queue   string
}

SubscribeResult reports the registered subscription identity.

type Subscriber

type Subscriber interface {
	Subscribe(ctx context.Context, options SubscribeOptions, handler MessageHandler) (SubscribeResult, error)
	Close(ctx context.Context) error
	Wait(ctx context.Context) error
}

Subscriber is the core-owned NATS subscriber contract for new consumers.

type SubscriberAdapter

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

SubscriberAdapter adapts the legacy NATS subscriber to the core-owned Subscriber contract.

func NewSubscriber

func NewSubscriber(s *legacysubscriber.Subscriber) *SubscriberAdapter

NewSubscriber wraps a legacy NATS subscriber with the core-owned Subscriber contract.

func (*SubscriberAdapter) Close

func (s *SubscriberAdapter) Close(ctx context.Context) error

Close closes the wrapped subscriber.

func (*SubscriberAdapter) Subscribe

func (s *SubscriberAdapter) Subscribe(
	ctx context.Context,
	options SubscribeOptions,
	handler MessageHandler,
) (SubscribeResult, error)

Subscribe registers a queue subscriber through the core-owned Subscriber contract.

func (*SubscriberAdapter) Wait

func (s *SubscriberAdapter) Wait(ctx context.Context) error

Wait waits until the wrapped subscriber stops.