nats

Documentation

Index

• Constants
• Variables
• type AsyncPublisher
  • func NewAsyncPublisher(clientID string) *AsyncPublisher
  • func (p *AsyncPublisher) Open() error
  • func (p *AsyncPublisher) Publish(subject string, r io.Reader) error
• type Config
• type Handler
• type LogHandler
  • func (lh *LogHandler) Process(s Subscription, m Message)
• type Message
• type Publisher
• type QueueSubscriber
  • func NewQueueSubscriber(clientID string) *QueueSubscriber
  • func (s *QueueSubscriber) Open() error
  • func (s *QueueSubscriber) Subscribe(subject, group string, handler Handler) error
• type Server
  • func NewServer(c Config) *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")

Types

type AsyncPublisher

type AsyncPublisher struct {
	ClientID   string
	Connection stan.Conn
	Logger     *zap.Logger
}

func NewAsyncPublisher

func NewAsyncPublisher(clientID 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 Config

type Config struct {
	// The directory where nats persists message information
	FilestoreDir string
}

Config is the configuration for the NATS streaming server

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 {
	Logger *zap.Logger
}

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
}

func NewQueueSubscriber

func NewQueueSubscriber(clientID 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 *stand.StanServer
	// contains filtered or unexported fields
}

Server wraps a connection to a NATS streaming server

func NewServer

func NewServer(c Config) *Server

NewServer creates and returns a new server struct from the provided config

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
}