messaging

Documentation

Index

• Variables
• type Broker
• type MessageBroker
  • func NewMessageBroker(url, userName, password, serviceName string) (*MessageBroker, error)
  • func (mb *MessageBroker) CloseConnection() error
  • func (mb *MessageBroker) Consume(consumer jetstream.Consumer, handler func(msg jetstream.Msg)) (jetstream.ConsumeContext, error)
  • func (mb *MessageBroker) CreatePersistentMessageReceiver(name, id, address, port, cType string, stream jetstream.Stream) (jetstream.Consumer, error)
  • func (mb *MessageBroker) CreatePersistentMessageStore(name string, subjects []string) (jetstream.Stream, error)
  • func (mb *MessageBroker) DesubscribeAsync(topic string) error
  • func (mb *MessageBroker) GetConsumer() jetstream.Consumer
  • func (mb *MessageBroker) GetStream() jetstream.Stream
  • func (mb *MessageBroker) PersistedPublish(topic string, payload []byte) error
  • func (mb *MessageBroker) Publish(topic string, payload []byte) error
  • func (mb *MessageBroker) PublishRequestAndWait(topic string, payload []byte, timeout time.Duration) (*nats.Msg, error)
  • func (mb *MessageBroker) RetrieveRegisteredConsumers(stream jetstream.Stream) ([]ResolvedService, error)
  • func (mb *MessageBroker) SetConsumer(consumer jetstream.Consumer)
  • func (mb *MessageBroker) SetStream(stream jetstream.Stream)
  • func (mb *MessageBroker) SubscribeAsync(topic string, callback nats.MsgHandler) error
  • func (mb *MessageBroker) SubscribeQueueAsync(topic string, queue string, callback nats.MsgHandler) error
• type ResolvedService

Variables

var ErrorMetadataConfigIncomplete = errors.New("service metadata config incomplete")

Types

type Broker

type Broker interface {
	CreatePersistentMessageStore(name string, subjects []string) (jetstream.Stream, error)
	CreatePersistentMessageReceiver(name, id, address, port, ctype string, stream jetstream.Stream) (jetstream.Consumer, error)
	RetrieveRegisteredConsumers(stream jetstream.Stream) ([]ResolvedService, error)
	PersistedPublish(topic string, payload []byte) error
	Consume(consumer jetstream.Consumer, handler func(msg jetstream.Msg)) (jetstream.ConsumeContext, error)
	Subscribe(topic string, callback nats.MsgHandler) error
	QueueGroupSubscribe(topic, queue string, callback nats.MsgHandler) error
	Publish(topic string, payload []byte) error
	PublishSync(topic string, payload []byte, timeout time.Duration) (*nats.Msg, error)
	Unsubscribe(topic string) error
	Close() error
}

type MessageBroker

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

func NewMessageBroker

func NewMessageBroker(url, userName, password, serviceName string) (*MessageBroker, error)

func (*MessageBroker) CloseConnection

func (mb *MessageBroker) CloseConnection() error

func (*MessageBroker) Consume

func (mb *MessageBroker) Consume(consumer jetstream.Consumer, handler func(msg jetstream.Msg)) (jetstream.ConsumeContext, error)

func (*MessageBroker) CreatePersistentMessageReceiver

func (mb *MessageBroker) CreatePersistentMessageReceiver(name, id, address, port, cType string, stream jetstream.Stream) (jetstream.Consumer, error)

func (*MessageBroker) CreatePersistentMessageStore

func (mb *MessageBroker) CreatePersistentMessageStore(name string, subjects []string) (jetstream.Stream, error)

func (*MessageBroker) DesubscribeAsync

func (mb *MessageBroker) DesubscribeAsync(topic string) error

func (*MessageBroker) GetConsumer

func (mb *MessageBroker) GetConsumer() jetstream.Consumer

func (*MessageBroker) GetStream

func (mb *MessageBroker) GetStream() jetstream.Stream

func (*MessageBroker) PersistedPublish

func (mb *MessageBroker) PersistedPublish(topic string, payload []byte) error

func (*MessageBroker) Publish

func (mb *MessageBroker) Publish(topic string, payload []byte) error

func (*MessageBroker) PublishRequestAndWait

func (mb *MessageBroker) PublishRequestAndWait(topic string, payload []byte, timeout time.Duration) (*nats.Msg, error)

func (*MessageBroker) RetrieveRegisteredConsumers

func (mb *MessageBroker) RetrieveRegisteredConsumers(stream jetstream.Stream) ([]ResolvedService, error)

func (*MessageBroker) SetConsumer

func (mb *MessageBroker) SetConsumer(consumer jetstream.Consumer)

func (*MessageBroker) SetStream

func (mb *MessageBroker) SetStream(stream jetstream.Stream)

func (*MessageBroker) SubscribeAsync

func (mb *MessageBroker) SubscribeAsync(topic string, callback nats.MsgHandler) error

func (*MessageBroker) SubscribeQueueAsync

func (mb *MessageBroker) SubscribeQueueAsync(topic string, queue string, callback nats.MsgHandler) error

type ResolvedService

type ResolvedService struct {
	ID      string
	Address string
	Port    int
}