nmq

Documentation

Index

• Variables
• func CloseNatsMQ()
• func MustInitNatsMQ(cfg *NatsMQConfig)
• func NewClient(url string, options ...nats.Option) (*nmqClient, error)
• func NewClientWithDefaultOptions(ctx context.Context) (mqwrapper.Client, error)
• func NewNmqID(id MessageIDType) mqcommon.MessageID
• func SerializeNmqID(messageID MessageIDType) []byte
• type Consumer
  • func (nc *Consumer) Ack(message common.Message)
  • func (nc *Consumer) Chan() <-chan common.Message
  • func (nc *Consumer) CheckTopicValid(topic string) error
  • func (nc *Consumer) Close()
  • func (nc *Consumer) GetLatestMsgID() (common.MessageID, error)
  • func (nc *Consumer) Seek(id common.MessageID, inclusive bool) error
  • func (nc *Consumer) Subscription() string
• type MessageIDType
  • func DeserializeNmqID(messageID []byte) MessageIDType
• type NatsMQConfig
  • func ParseServerOption(params *paramtable.ComponentParam) *NatsMQConfig

Variables

var Nmq *server.Server

Nmq is global natsmq instance that will be initialized only once

Functions

func CloseNatsMQ

func CloseNatsMQ()

CloseNatsMQ is used to close global natsmq

func MustInitNatsMQ

func MustInitNatsMQ(cfg *NatsMQConfig)

MustInitNatsMQ init global local natsmq instance. Panic if initailizing operation failed.

func NewClient

func NewClient(url string, options ...nats.Option) (*nmqClient, error)

NewClient returns a new nmqClient object

func NewClientWithDefaultOptions

func NewClientWithDefaultOptions(ctx context.Context) (mqwrapper.Client, error)

NewClientWithDefaultOptions returns a new NMQ client with default options. It retrieves the NMQ client URL from the server configuration.

func NewNmqID

func NewNmqID(id MessageIDType) mqcommon.MessageID

NewNmqID creates and returns a new instance of the nmqID struct with the given MessageID.

func SerializeNmqID

func SerializeNmqID(messageID MessageIDType) []byte

SerializeNmqID is used to serialize a message ID to byte array

Types

type Consumer

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

Consumer is a client that used to consume messages from natsmq

func (*Consumer) Ack

func (nc *Consumer) Ack(message common.Message)

Ack is used to ask a natsmq message

func (*Consumer) Chan

func (nc *Consumer) Chan() <-chan common.Message

Chan returns a channel to read messages from natsmq

func (*Consumer) CheckTopicValid

func (nc *Consumer) CheckTopicValid(topic string) error

CheckTopicValid verifies if the given topic is valid for this consumer. 1. topic should exist.

func (*Consumer) Close

func (nc *Consumer) Close()

Close is used to free the resources of this consumer

func (*Consumer) GetLatestMsgID

func (nc *Consumer) GetLatestMsgID() (common.MessageID, error)

GetLatestMsgID returns the ID of the most recent message processed by the consumer.

func (*Consumer) Seek

func (nc *Consumer) Seek(id common.MessageID, inclusive bool) error

Seek is used to seek the position in natsmq topic

func (*Consumer) Subscription

func (nc *Consumer) Subscription() string

Subscription returns the subscription name of this consumer

type MessageIDType

type MessageIDType = uint64

MessageIDType is a type alias for server.UniqueID that represents the ID of a Nmq message.

func DeserializeNmqID

func DeserializeNmqID(messageID []byte) MessageIDType

DeserializeNmqID is used to deserialize a message ID from byte array

type NatsMQConfig

type NatsMQConfig struct {
	Opts              server.Options
	InitializeTimeout time.Duration
}

NatsMQConfig is used to initialize NatsMQ.

func ParseServerOption

func ParseServerOption(params *paramtable.ComponentParam) *NatsMQConfig

ParseServerOption get nats server option from paramstable.