kafkaHelper

package

v1.4.482 Latest Latest Go to latest Published: May 6, 2024 License: GPL-2.0 Imports: 11 Imported by: 7

Details

Valid go.mod file

The Go module system was introduced in Go 1.11 and is the official dependency management solution for Go.
Redistributable license

Redistributable licenses place minimal restrictions on how software can be used, modified, and redistributed.
Tagged version

Modules with tagged versions give importers more predictable builds.
Stable version

When a project reaches major version v1 it is considered stable.
Learn more about best practices

Repository

github.com/diadata-org/diadata

Links

Open Source Insights

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 ¶

View Source

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
)

Variables ¶

This section is empty.

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 ¶ added in v1.4.6

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
}

Source Files ¶

View all Source files

Kafka.go

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL