msgqueue

Documentation

Index

• Constants
• Variables
• func Retry(attempts int, sleep time.Duration, fn func() error) error
• type MsgSender
  • func NewProducer(log log.Logger) MsgSender
  • func NewProducerFromConfig(brokers []string, topics string, featureToggle bool, log log.Logger) MsgSender
• type MsgWriter
  • func NewFileMsgWriter(filePath string) (MsgWriter, error)
  • func NewKafkaMsgWriter(brokers string) (MsgWriter, error)
  • func NewNopMsgWriter() MsgWriter
  • func NewStdOutMsgWriter() MsgWriter

Constants

const (
	FlagBrokers       = "brokers"
	FlagTopics        = "subscribe-modules"
	FlagFeatureToggle = "feature-toggle"
	KafkaPubTopic     = "coinex-dex"
)

const (
	CfgPrefixFile  = "file:"
	CfgPrefixKafka = "kafka:"
	CfgPrefixOS    = "os:"
)

const RetryNum = math.MaxInt64

Variables

var (
	EventTypeMsgQueue = "kafka"
)

Functions

func Retry

func Retry(attempts int, sleep time.Duration, fn func() error) error

Types

type MsgSender

type MsgSender interface {
	SendMsg(key []byte, v []byte)
	IsSubscribed(topic string) bool
	IsOpenToggle() bool
	GetMode() []string
	Close()
}

func NewProducer

func NewProducer(log log.Logger) MsgSender

func NewProducerFromConfig

func NewProducerFromConfig(brokers []string, topics string, featureToggle bool, log log.Logger) MsgSender

type MsgWriter

type MsgWriter interface {
	WriteKV(k, v []byte) error
	Close() error
	String() string
}

func NewFileMsgWriter

func NewFileMsgWriter(filePath string) (MsgWriter, error)

func NewKafkaMsgWriter

func NewKafkaMsgWriter(brokers string) (MsgWriter, error)

func NewNopMsgWriter

func NewNopMsgWriter() MsgWriter

func NewStdOutMsgWriter

func NewStdOutMsgWriter() MsgWriter