nats

README

nats

nats

Documentation

Index

• Variables
• type Config
• type Handler
• type JetstreamClient
  • func NewJetstream(config *Config, opts ...Option) (*JetstreamClient, error)
  • func (c *JetstreamClient) Close()
  • func (c *JetstreamClient) Connect() (err error)
  • func (c *JetstreamClient) Drain() error
  • func (c *JetstreamClient) Publish(ctx context.Context, subject string, data []byte) (*PubAck, error)
  • func (c *JetstreamClient) Subscribe(subject string, handler Handler, opts ...nats.SubOpt) (*nats.Subscription, error)
  • func (c *JetstreamClient) Unsubscribe() error
• type Logger
• type Message
• type NatsClient
  • func New(config *Config, opts ...Option) (*NatsClient, error)
  • func (c *NatsClient) Close()
  • func (c *NatsClient) Connect() (err error)
  • func (c *NatsClient) Drain() error
  • func (c *NatsClient) Publish(ctx context.Context, subject string, data []byte) error
  • func (c *NatsClient) Subscribe(subject string, handler Handler) (*nats.Subscription, error)
  • func (c *NatsClient) Unsubscribe() error
• type NatsOption
• type Option
  • func WithNatsOptions(opts ...NatsOption) Option
  • func WithStreamConfig(cfg *StreamConfig) Option
  • func WithTracer(tracer trace.Tracer) Option
  • func WithWarnHandler(logger Logger) Option
• type PubAck
• type StreamConfig

Variables

var (
	ErrDrainFailed       = errors.New("drain failed")
	ErrUnsubscribeFailed = errors.New("unsubscribe failed")
	ErrInvalidConfig     = errors.New("invalid config value")
)

Types

type Config

type Config struct {
	Addr       string // "nats://localhost:4222, nats://localhost:4223, nats://localhost:4224"
	StreamName string
	Subjects   []string
}

type Handler

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

Handler to handle sub message.

type JetstreamClient

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

func NewJetstream

func NewJetstream(config *Config, opts ...Option) (*JetstreamClient, error)

func (*JetstreamClient) Close

func (c *JetstreamClient) Close()

func (*JetstreamClient) Connect

func (c *JetstreamClient) Connect() (err error)

func (*JetstreamClient) Drain

func (c *JetstreamClient) Drain() error

func (*JetstreamClient) Publish

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

func (*JetstreamClient) Subscribe

func (c *JetstreamClient) Subscribe(subject string, handler Handler, opts ...nats.SubOpt) (*nats.Subscription, error)

func (*JetstreamClient) Unsubscribe

func (c *JetstreamClient) Unsubscribe() error

type Logger

type Logger interface {
	Warn(ctx context.Context, args ...interface{})
	Warnf(ctx context.Context, template string, args ...interface{})
	Error(ctx context.Context, args ...interface{})
	Errorf(ctx context.Context, template string, args ...interface{})
}

type Message

type Message = nats.Msg

Message is nats.Msg.

type NatsClient

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

func New

func New(config *Config, opts ...Option) (*NatsClient, error)

func (*NatsClient) Close

func (c *NatsClient) Close()

func (*NatsClient) Connect

func (c *NatsClient) Connect() (err error)

func (*NatsClient) Drain

func (c *NatsClient) Drain() error

func (*NatsClient) Publish

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

func (*NatsClient) Subscribe

func (c *NatsClient) Subscribe(subject string, handler Handler) (*nats.Subscription, error)

func (*NatsClient) Unsubscribe

func (c *NatsClient) Unsubscribe() error

type NatsOption

type NatsOption = nats.Option // nolint:golint,revive // need nats option.

type Option

type Option func(*options)

func WithNatsOptions

func WithNatsOptions(opts ...NatsOption) Option

func WithStreamConfig

func WithStreamConfig(cfg *StreamConfig) Option

func WithTracer

func WithTracer(tracer trace.Tracer) Option

func WithWarnHandler

func WithWarnHandler(logger Logger) Option

type PubAck

type PubAck = nats.PubAck

type StreamConfig

type StreamConfig = nats.StreamConfig