msgqueue

Documentation

Index

• Constants
• Variables
• func FillMsgs(ctx sdk.Context, key string, msg interface{})
• func GetFileName(dir string, fileIndex int) string
• func GetFilePathAndFileIndexFromDir(dir string, maxFileSize int) (filePath string, fileIndex int, err error)
• func GetFileSize(filePath string) int
• func Retry(attempts int, sleep time.Duration, fn func() error) error
• func SetMkFifoFunc(mkFifo MkFifoFunc)
• type MkFifoFunc
• type MsgSender
  • func NewProducer(log log.Logger) MsgSender
  • func NewProducerFromConfig(brokers []string, topics string, featureToggle bool, log log.Logger) MsgSender
• type MsgWriter
  • func NewDirMsgWriter(dir string) (MsgWriter, error)
  • func NewFileMsgWriter(filePath string) (MsgWriter, error)
  • func NewKafkaMsgWriter(brokers string) (MsgWriter, error)
  • func NewNopMsgWriter() MsgWriter
  • func NewPipeMsgWriter(pipe string) (MsgWriter, error)
  • func NewStdOutMsgWriter() MsgWriter

Constants

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

const (
	CfgPrefixFile  = "file:"
	CfgPrefixKafka = "kafka:"
	CfgPrefixOS    = "os:"
	CfgPrefixDir   = "dir:"
	CfgNamedPipe   = "pipe:"
)

const RetryNum = math.MaxInt64

Variables

var (
	EventTypeMsgQueue = "kafka"
)

var MaxFileSize = 1024 * 1024 * 100

Functions

func FillMsgs

func FillMsgs(ctx sdk.Context, key string, msg interface{})

func GetFileName

func GetFileName(dir string, fileIndex int) string

func GetFilePathAndFileIndexFromDir

func GetFilePathAndFileIndexFromDir(dir string, maxFileSize int) (filePath string, fileIndex int, err error)

func GetFileSize

func GetFileSize(filePath string) int

func Retry

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

func SetMkFifoFunc

func SetMkFifoFunc(mkFifo MkFifoFunc)

Types

type MkFifoFunc

type MkFifoFunc func(path string, mode uint32) (err error)

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 NewDirMsgWriter

func NewDirMsgWriter(dir string) (MsgWriter, error)

func NewFileMsgWriter

func NewFileMsgWriter(filePath string) (MsgWriter, error)

func NewKafkaMsgWriter

func NewKafkaMsgWriter(brokers string) (MsgWriter, error)

func NewNopMsgWriter

func NewNopMsgWriter() MsgWriter

func NewPipeMsgWriter

func NewPipeMsgWriter(pipe string) (MsgWriter, error)

func NewStdOutMsgWriter

func NewStdOutMsgWriter() MsgWriter