nats

Documentation

Index

• Constants
• Variables
• func NewDefaultServerOptions() server.Options
• type AsyncPublisher
  • func NewAsyncPublisher(log *zap.Logger, clientID string, addr string) *AsyncPublisher
  • func (p *AsyncPublisher) Open() error
  • func (p *AsyncPublisher) Publish(subject string, r io.Reader) error
• type Handler
• type LogHandler
  • func (lh *LogHandler) Process(s Subscription, m Message)
• type Message
• type Publisher
• type QueueSubscriber
  • func NewQueueSubscriber(clientID string, addr string) *QueueSubscriber
  • func (s *QueueSubscriber) Open() error
  • func (s *QueueSubscriber) Subscribe(subject, group string, handler Handler) error
• type Server
  • func NewServer(opts *server.Options) *Server
  • func (s *Server) Close()
  • func (s *Server) Open() error
• type Subscriber
• type Subscription

Constants

const ServerName = "platform"

Variables

var ErrNoNatsConnection = errors.New("nats connection has not been established. Call Open() first")

Functions

func NewDefaultServerOptions

func NewDefaultServerOptions() server.Options

NewDefaultServerOptions returns the default NATS server options, allowing the caller to override specific fields.

Types

type AsyncPublisher

type AsyncPublisher struct {
	ClientID   string
	Connection stan.Conn

	Addr string
	// contains filtered or unexported fields
}

func NewAsyncPublisher

func NewAsyncPublisher(log *zap.Logger, clientID string, addr string) *AsyncPublisher

func (*AsyncPublisher) Open

func (p *AsyncPublisher) Open() error

Open creates and maintains a connection to NATS server

func (*AsyncPublisher) Publish

func (p *AsyncPublisher) Publish(subject string, r io.Reader) error

type Handler

type Handler interface {
	// Process does something with a received subscription message, then acks it.
	Process(s Subscription, m Message)
}

type LogHandler

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

func (*LogHandler) Process

func (lh *LogHandler) Process(s Subscription, m Message)

type Message

type Message interface {
	Data() []byte
	Ack() error
}

type Publisher

type Publisher interface {
	// Publish a new message to channel
	Publish(subject string, r io.Reader) error
}

type QueueSubscriber

type QueueSubscriber struct {
	ClientID   string
	Connection stan.Conn
	Addr       string
}

func NewQueueSubscriber

func NewQueueSubscriber(clientID string, addr string) *QueueSubscriber

func (*QueueSubscriber) Open

func (s *QueueSubscriber) Open() error

Open creates and maintains a connection to NATS server

func (*QueueSubscriber) Subscribe

func (s *QueueSubscriber) Subscribe(subject, group string, handler Handler) error

type Server

type Server struct {
	Server *sserver.StanServer
	// contains filtered or unexported fields
}

Server wraps a connection to a NATS streaming server

func NewServer

func NewServer(opts *server.Options) *Server

NewServer creates a new streaming server with the provided server options.

func (*Server) Close

func (s *Server) Close()

Close stops the embedded NATS server.

func (*Server) Open

func (s *Server) Open() error

Open starts a NATS streaming server

type Subscriber

type Subscriber interface {
	// Subscribe listens to a channel, handling messages with Handler
	Subscribe(subject, group string, handler Handler) error
}

type Subscription

type Subscription interface {
	// Pending returns the number of queued messages and queued bytes for this subscription.
	Pending() (int64, int64, error)

	// Delivered returns the number of delivered messages for this subscription.
	Delivered() (int64, error)

	// Close removes this subscriber
	Close() error
}