msgqueue

Documentation

Index

• Constants
• Variables
• func FillMsgs(ctx sdk.Context, key string, msg interface{})
• func GetFileLeastHeightInDir(dir string) (string, int64, error)
• 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 FileDeleter
  • func NewFileDeleter(heightCh <-chan int64, dir string) *FileDeleter
  • func (p *FileDeleter) Run()
• type GetFilePathAndFileIndexFromDirCb
• 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, cb GetFilePathAndFileIndexFromDirCb) (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
• type NewHeightInfo
• type RegulateWriteDir
  • func NewRegulateWriteDir(dir string) (*RegulateWriteDir, error)
• type Worker

Constants

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

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

const FILEHEIGHT = 10000

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 GetFileLeastHeightInDir

func GetFileLeastHeightInDir(dir string) (string, int64, error)

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 FileDeleter

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

func NewFileDeleter

func NewFileDeleter(heightCh <-chan int64, dir string) *FileDeleter

func (*FileDeleter) Run

func (p *FileDeleter) Run()

type GetFilePathAndFileIndexFromDirCb

type GetFilePathAndFileIndexFromDirCb func(string, int) (filePath string, fileIndex int, err error)

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, cb GetFilePathAndFileIndexFromDirCb) (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

type NewHeightInfo

type NewHeightInfo struct {
	ChainID       string       `json:"chain_id"`
	Height        int64        `json:"height"`
	TimeStamp     int64        `json:"timestamp"`
	LastBlockHash cmn.HexBytes `json:"last_block_hash"`
}

type RegulateWriteDir

type RegulateWriteDir struct {
	MsgWriter
	// contains filtered or unexported fields
}

func NewRegulateWriteDir

func NewRegulateWriteDir(dir string) (*RegulateWriteDir, error)

type Worker

type Worker interface {
	Work()
}