nats

Documentation

Index

• Constants
• Variables
• func NewReceiver(connection Connection, codec communication.Codec, topic string) *receiverNATS
• func NewSender(connection Connection, codec communication.Codec) *senderNATS
• func ParseServerURIs(serverURIs []string) ([]*url.URL, error)
• func ParseServerURL(serverURI string) (*url.URL, error)
• func SignedSubject(signer identity.Signer, topic string) (string, error)
• type BrokerConnector
  • func NewBrokerConnector(dialer requests.DialContext, resolveContext resolver.ResolveContext) *BrokerConnector
  • func (b *BrokerConnector) Connect(serverURLs ...*url.URL) (Connection, error)
• type Connection
• type ConnectionMock
  • func NewConnectionMock() *ConnectionMock
  • func StartConnectionMock() *ConnectionMock
  • func (conn *ConnectionMock) Check() error
  • func (conn *ConnectionMock) Close()
  • func (conn *ConnectionMock) GetLastMessage() []byte
  • func (conn *ConnectionMock) GetLastMessageSubject() string
  • func (conn *ConnectionMock) GetLastRequest() []byte
  • func (conn *ConnectionMock) MessageWait(waitChannel chan interface{}) (interface{}, error)
  • func (conn *ConnectionMock) MockError(message string)
  • func (conn *ConnectionMock) MockResponse(subject string, payload []byte)
  • func (conn *ConnectionMock) Open() error
  • func (conn *ConnectionMock) Publish(subject string, payload []byte) error
  • func (conn *ConnectionMock) Request(subject string, payload []byte, timeout time.Duration) (*nats.Msg, error)
  • func (conn *ConnectionMock) RequestWithContext(ctx context.Context, subject string, payload []byte) (*nats.Msg, error)
  • func (conn *ConnectionMock) Servers() []string
  • func (conn *ConnectionMock) Subscribe(subject string, handler nats.MsgHandler) (*nats.Subscription, error)
• type ConnectionWrap
  • func (c *ConnectionWrap) Close()
  • func (c *ConnectionWrap) Open() (err error)
  • func (c *ConnectionWrap) Servers() []string

Constants

const (
	// DefaultBrokerScheme broker scheme.
	DefaultBrokerScheme = "nats"
	// DefaultBrokerPort broker port.
	DefaultBrokerPort = 4222
)

Variables

var (
	// SignatureEncoding represents base64 variant chosen for signature
	// encoding
	SignatureEncoding = base64.RawURLEncoding
)

Functions

func NewReceiver

func NewReceiver(connection Connection, codec communication.Codec, topic string) *receiverNATS

NewReceiver constructs new Receiver's instance which works through NATS connection. Codec packs/unpacks messages to byte payloads. Topic (optional) if need to send messages prefixed topic.

func NewSender

func NewSender(connection Connection, codec communication.Codec) *senderNATS

NewSender constructs new Sender's instance which works thru NATS connection. Codec packs/unpacks messages to byte payloads. Topic (optional) if need to send messages prefixed topic.

func ParseServerURIs

func ParseServerURIs(serverURIs []string) ([]*url.URL, error)

ParseServerURIs validates given list of NATS server addresses.

func ParseServerURL

func ParseServerURL(serverURI string) (*url.URL, error)

ParseServerURL validates given NATS server address.

func SignedSubject

func SignedSubject(signer identity.Signer, topic string) (string, error)

SignedSubject signs topic to pass command through nats-proxy

Types

type BrokerConnector

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

BrokerConnector establishes new connections to NATS servers and handles reconnects.

func NewBrokerConnector

func NewBrokerConnector(dialer requests.DialContext, resolveContext resolver.ResolveContext) *BrokerConnector

NewBrokerConnector creates a new BrokerConnector.

func (*BrokerConnector) Connect

func (b *BrokerConnector) Connect(serverURLs ...*url.URL) (Connection, error)

Connect establishes a new connection to the broker(s).

type Connection

type Connection interface {
	Open() error
	Close()
	Servers() []string
	Publish(subject string, payload []byte) error
	Subscribe(subject string, handler nats.MsgHandler) (*nats.Subscription, error)
	Request(subject string, payload []byte, timeout time.Duration) (*nats.Msg, error)
	RequestWithContext(ctx context.Context, subj string, data []byte) (*nats.Msg, error)
}

Connection represents is publish-subscriber instance which can deliver messages

type ConnectionMock

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

ConnectionMock acts as a local connection implementation

func NewConnectionMock

func NewConnectionMock() *ConnectionMock

NewConnectionMock constructs new NATS connection which delivers published messages to local subscribers

func StartConnectionMock

func StartConnectionMock() *ConnectionMock

StartConnectionMock creates connection and starts it immediately

func (*ConnectionMock) Check

func (conn *ConnectionMock) Check() error

Check checks the connection

func (*ConnectionMock) Close

func (conn *ConnectionMock) Close()

Close destructs the connection

func (*ConnectionMock) GetLastMessage

func (conn *ConnectionMock) GetLastMessage() []byte

GetLastMessage returns the last message received

func (*ConnectionMock) GetLastMessageSubject

func (conn *ConnectionMock) GetLastMessageSubject() string

GetLastMessageSubject returns the last message subject

func (*ConnectionMock) GetLastRequest

func (conn *ConnectionMock) GetLastRequest() []byte

GetLastRequest gets last request data

func (*ConnectionMock) MessageWait

func (conn *ConnectionMock) MessageWait(waitChannel chan interface{}) (interface{}, error)

MessageWait waits for a message to arrive

func (*ConnectionMock) MockError

func (conn *ConnectionMock) MockError(message string)

MockError mocks the error

func (*ConnectionMock) MockResponse

func (conn *ConnectionMock) MockResponse(subject string, payload []byte)

MockResponse mocks the response

func (*ConnectionMock) Open

func (conn *ConnectionMock) Open() error

Open starts the connection

func (*ConnectionMock) Publish

func (conn *ConnectionMock) Publish(subject string, payload []byte) error

Publish publishes a new message

func (*ConnectionMock) Request

func (conn *ConnectionMock) Request(subject string, payload []byte, timeout time.Duration) (*nats.Msg, error)

Request sends a new request

func (*ConnectionMock) RequestWithContext

func (conn *ConnectionMock) RequestWithContext(ctx context.Context, subject string, payload []byte) (*nats.Msg, error)

RequestWithContext Request sends a new request with context

func (*ConnectionMock) Servers

func (conn *ConnectionMock) Servers() []string

Servers returns list of currently connected servers

func (*ConnectionMock) Subscribe

func (conn *ConnectionMock) Subscribe(subject string, handler nats.MsgHandler) (*nats.Subscription, error)

Subscribe subscribes to a topic

type ConnectionWrap

type ConnectionWrap struct {
	*nats_lib.Conn
	// contains filtered or unexported fields
}

ConnectionWrap defines wrapped connection to NATS server(s).

func (*ConnectionWrap) Close

func (c *ConnectionWrap) Close()

Close destructs the connection.

func (*ConnectionWrap) Open

func (c *ConnectionWrap) Open() (err error)

Open starts the connection: left for test compatibility. Deprecated: Use nats.BrokerConnector#Connect() instead.

func (*ConnectionWrap) Servers

func (c *ConnectionWrap) Servers() []string

Servers returns list of currently connected servers.