natsclient

Documentation

Index

• type Config
• type NatsClient
  • func New(config *Config, bigMsgService messages.BigMsgService) *NatsClient
  • func UnsafeNew(params ...interface{}) *NatsClient
  • func (s *NatsClient) HandleMessages(ctx context.Context, wg *sync.WaitGroup, subj string, ...)
  • func (s *NatsClient) HandleStreamingMessages(ctx context.Context, wg *sync.WaitGroup, subj string, ...)
  • func (s *NatsClient) IsConnected() bool
  • func (s *NatsClient) Run(mainParams *application.MainParams) error
  • func (s *NatsClient) SendMessage(subj string, data []byte) error
  • func (s *NatsClient) SendRequest(subj string, data []byte, timeout time.Duration) (messages.Message, error)
  • func (s *NatsClient) SendStreamingMessage(subj string, data []byte) error
  • func (s *NatsClient) SetConn(nsc stan.Conn) *NatsClient
  • func (s *NatsClient) Stop()
  • func (s *NatsClient) Sub(subj string) (<-chan messages.Message, func() error)
  • func (s *NatsClient) SubStreaming(subj string) (<-chan messages.Message, func() error)
  • func (s *NatsClient) SubStreamingWithContext(ctx context.Context, wg *sync.WaitGroup, subj string) (<-chan messages.Message, func() error)
  • func (s *NatsClient) SubWithContext(ctx context.Context, wg *sync.WaitGroup, subj string) (<-chan messages.Message, func() error)
• type NatsMsg
  • func (m *NatsMsg) Data() []byte
  • func (m *NatsMsg) Respond(data []byte) error
  • func (m *NatsMsg) SetData(data []byte)
• type NatsSubscription
  • func (n *NatsSubscription) Unsub() error
• type StanMsg
  • func (m *StanMsg) Data() []byte
  • func (m *StanMsg) Respond(data []byte) error
  • func (m *StanMsg) SetData(data []byte)

Types

type Config

type Config struct {
	Url              string `arg:"env:NATS_URL"`               // адрес сервер NATS
	StreamingCluster string `arg:"env:NATS_STREAMING_CLUSTER"` // имя кластера NATS Streaming

}

Конфигурация модуля.

type NatsClient

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

Клиент для отправки сообщений через NATS и NATS-Streaming

func New

func New(config *Config, bigMsgService messages.BigMsgService) *NatsClient

Конструктор NatsClient. В случае передачи nil в качестве конфигурации, конфигурация будет получена из переменных окружения.

func UnsafeNew

func UnsafeNew(params ...interface{}) *NatsClient

Конструктор NatsClient c произвольными параметрами.

func (*NatsClient) HandleMessages

func (s *NatsClient) HandleMessages(ctx context.Context, wg *sync.WaitGroup, subj string, handleMessageFunc func(ctx context.Context, wg *sync.WaitGroup, msg messages.Message))

func (*NatsClient) HandleStreamingMessages

func (s *NatsClient) HandleStreamingMessages(ctx context.Context, wg *sync.WaitGroup, subj string, handleMessageFunc func(ctx context.Context, wg *sync.WaitGroup, msg messages.Message))

func (*NatsClient) IsConnected

func (s *NatsClient) IsConnected() bool

Соединение в данный момент установлено

func (*NatsClient) Run

func (s *NatsClient) Run(mainParams *application.MainParams) error

Запуск клиента в работу. Реализация интерфейса application.Runnable.

func (*NatsClient) SendMessage

func (s *NatsClient) SendMessage(subj string, data []byte) error

SendMessage публикация сообщения для NATS

func (*NatsClient) SendRequest

func (s *NatsClient) SendRequest(subj string, data []byte, timeout time.Duration) (messages.Message, error)

SendRequest публикация сообщения с запросом ответа

func (*NatsClient) SendStreamingMessage

func (s *NatsClient) SendStreamingMessage(subj string, data []byte) error

SendREquest публикация сообщения для NATS и NATS Streaming

func (*NatsClient) SetConn

func (s *NatsClient) SetConn(nsc stan.Conn) *NatsClient

передач сервису соединения с NATS

func (*NatsClient) Stop

func (s *NatsClient) Stop()

Завершение работы модуля. Реализация интерфейса service.Runnable.

func (*NatsClient) Sub

func (s *NatsClient) Sub(subj string) (<-chan messages.Message, func() error)

обычная подпика

func (*NatsClient) SubStreaming

func (s *NatsClient) SubStreaming(subj string) (<-chan messages.Message, func() error)

подпика на Streaming

func (*NatsClient) SubStreamingWithContext

func (s *NatsClient) SubStreamingWithContext(ctx context.Context, wg *sync.WaitGroup, subj string) (<-chan messages.Message, func() error)

подпика на Streaming с контекстом

func (*NatsClient) SubWithContext

func (s *NatsClient) SubWithContext(ctx context.Context, wg *sync.WaitGroup, subj string) (<-chan messages.Message, func() error)

обычная подпика с контекстом

type NatsMsg

type NatsMsg struct {
	*nats.Msg
}

обертка для *nats.Msg, реализующая интерфейса Message

func (*NatsMsg) Data

func (m *NatsMsg) Data() []byte

получение данных из сообщения, реализация интерфейса Message

func (*NatsMsg) Respond

func (m *NatsMsg) Respond(data []byte) error

@ Overwrite

func (*NatsMsg) SetData

func (m *NatsMsg) SetData(data []byte)

type NatsSubscription

type NatsSubscription struct {
	MsgChan chan *stan.Msg
	// contains filtered or unexported fields
}

func (*NatsSubscription) Unsub

func (n *NatsSubscription) Unsub() error

type StanMsg

type StanMsg struct {
	*stan.Msg
}

обертка для *StanMsg, реализующая интерфейса Message

func (*StanMsg) Data

func (m *StanMsg) Data() []byte

получение данных из сообщения, реализация интерфейса Message

func (*StanMsg) Respond

func (m *StanMsg) Respond(data []byte) error

Отправка ответа, реализация интерфейса Message. Т.к. в NATS-Streaming отправка сообщений с запросом ответа не поддерживается, метод возвращает ошибку.

func (*StanMsg) SetData

func (m *StanMsg) SetData(data []byte)