kafkaHelper

Documentation

Index

• Constants
• func GetElements(topic int, offset int64, nbElements int) ([]interface{}, error)
• func GetLastElement(topic int) (interface{}, error)
• func GetLastElementWithRetryOnError(topic int) interface{}
• func GetTopic(topic int) string
• func IsTopicEmpty(topic int) bool
• func NewReader(topic int) *kafka.Reader
• func NewReaderNextMessage(topic int) *kafka.Reader
• func NewReaderXElementsBeforeLastMessage(topic int, x int64) *kafka.Reader
• func NewSyncWriter(topic int) *kafka.Writer
• func NewSyncWriterWithCompression(topic int) *kafka.Writer
• func NewWriter(topic int) *kafka.Writer
• func ReadOffset(topic int) (offset int64, err error)
• func ReadOffsetWithRetryOnError(topic int) (offset int64)
• func WriteMessage(w *kafka.Writer, m KafkaMessage) error
• type Config
• type KafkaMessage
• type KafkaMessageWithAHash

Constants

const (
	TopicIndexBlock = 0

	TopicFiltersBlock = 1
	TopicTrades       = 2
	TopicTradesBlock  = 3

	// The replica topics can be used to forward trades and blocks to other services in parallel.
	TopicFiltersBlockReplica = 4
	TopicTradesReplica       = 5
	TopicTradesBlockReplica  = 6

	TopicTradesEstimation = 7

	TopicFiltersBlockDone = 14

	TopicFiltersBlockTest = 21
	TopicTradesTest       = 22
	TopicTradesBlockTest  = 23
	TopicNFTTrades        = 24
	TopicNFTTradesTest    = 25
)

Functions

func GetElements

func GetElements(topic int, offset int64, nbElements int) ([]interface{}, error)

func GetLastElement

func GetLastElement(topic int) (interface{}, error)

func GetLastElementWithRetryOnError

func GetLastElementWithRetryOnError(topic int) interface{}

func GetTopic

func GetTopic(topic int) string

func IsTopicEmpty

func IsTopicEmpty(topic int) bool

func NewReader

func NewReader(topic int) *kafka.Reader

func NewReaderNextMessage

func NewReaderNextMessage(topic int) *kafka.Reader

func NewReaderXElementsBeforeLastMessage

func NewReaderXElementsBeforeLastMessage(topic int, x int64) *kafka.Reader

func NewSyncWriter

func NewSyncWriter(topic int) *kafka.Writer

func NewSyncWriterWithCompression

func NewSyncWriterWithCompression(topic int) *kafka.Writer

func NewWriter

func NewWriter(topic int) *kafka.Writer

func ReadOffset

func ReadOffset(topic int) (offset int64, err error)

WithRetryOnError

func ReadOffsetWithRetryOnError

func ReadOffsetWithRetryOnError(topic int) (offset int64)

func WriteMessage

func WriteMessage(w *kafka.Writer, m KafkaMessage) error

Types

type Config

type Config struct {
	KafkaUrl []string
}

var (
	KafkaConfig Config
)

type KafkaMessage

type KafkaMessage interface {
	MarshalBinary() ([]byte, error)
}

type KafkaMessageWithAHash

type KafkaMessageWithAHash interface {
	Hash() string
}