nats

Documentation

Index

• type Client
  • func NewClient(url string, opts ClientOpts) (*Client, error)
  • func (nc *Client) Close(ctx context.Context) error
• type ClientOpts
• type ConsumerManager
• type CreateStoreArgs
• type JetstreamClient
  • func NewJetstreamClient(nc *Client) (*JetstreamClient, error)
  • func (jc *JetstreamClient) DeleteConsumer(ctx context.Context, stream string, consumer string) error
  • func (jc *JetstreamClient) GetConsumerInfo(ctx context.Context, stream string, consumer string) (*jetstream.ConsumerInfo, error)
  • func (jc *JetstreamClient) ListConsumers(ctx context.Context, stream string) ([]*jetstream.ConsumerInfo, error)
• type KeyValueManager
  • func NewKeyValueManager(jc *JetstreamClient, bucketName string) (*KeyValueManager, error)
  • func (kvm KeyValueManager) CreateStore(ctx context.Context, store string, args CreateStoreArgs) error
  • func (kvm KeyValueManager) DeleteStore(ctx context.Context, store string) error
  • func (kvm KeyValueManager) ListStores(ctx context.Context) []string
• type Options

Types

type Client

type Client struct {
	Conn *nats.Conn
	// contains filtered or unexported fields
}

func NewClient

func NewClient(url string, opts ClientOpts) (*Client, error)

func (*Client) Close

func (nc *Client) Close(ctx context.Context) error

Close implements Client.

type ClientOpts

type ClientOpts struct {
	CrdentialsFile string

	Name string
	// https://pkg.go.dev/github.com/nats-io/nats.go#Options
	Servers []string
	Logger  *slog.Logger

	DisconnectedCB func()
	ReconnectedCB  func()
	ConnectedCB    func()
	ClosedCB       func()
}

type ConsumerManager

type ConsumerManager interface {
	GetConsumerInfo(ctx context.Context, stream string, consumer string) (*jetstream.ConsumerInfo, error)
	ListConsumers(ctx context.Context, stream string) ([]*jetstream.ConsumerInfo, error)
	DeleteConsumer(ctx context.Context, stream string, consumer string) error
}

type CreateStoreArgs

type CreateStoreArgs struct {
	Replicas     int
	TTL          *time.Duration
	MaxValueSize *int32
	Description  *string
}

type JetstreamClient

type JetstreamClient struct {
	Jetstream jetstream.JetStream
	Logger    *slog.Logger
}

func NewJetstreamClient

func NewJetstreamClient(nc *Client) (*JetstreamClient, error)

func (*JetstreamClient) DeleteConsumer

func (jc *JetstreamClient) DeleteConsumer(ctx context.Context, stream string, consumer string) error

DeleteConsumer implements ConsumerManager.

func (*JetstreamClient) GetConsumerInfo

func (jc *JetstreamClient) GetConsumerInfo(ctx context.Context, stream string, consumer string) (*jetstream.ConsumerInfo, error)

GetConsumerInfo implements ConsumerManager

func (*JetstreamClient) ListConsumers

func (jc *JetstreamClient) ListConsumers(ctx context.Context, stream string) ([]*jetstream.ConsumerInfo, error)

ListConsumers implements ConsumerManager.

type KeyValueManager

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

func NewKeyValueManager

func NewKeyValueManager(jc *JetstreamClient, bucketName string) (*KeyValueManager, error)

func (KeyValueManager) CreateStore

func (kvm KeyValueManager) CreateStore(ctx context.Context, store string, args CreateStoreArgs) error

func (KeyValueManager) DeleteStore

func (kvm KeyValueManager) DeleteStore(ctx context.Context, store string) error

func (KeyValueManager) ListStores

func (kvm KeyValueManager) ListStores(ctx context.Context) []string

type Options

type Options nats.Options

expose these nats types, so that we can use them directly, without having to import nats.io/nats.go