kafka

Documentation

Index

• Variables
• func CheckConnector(config ConnectorConfig)
• func NewKafkaProducer() (*kafkaProducer, error)
• func RegisterConnector(config ConnectorConfig) *http.Response
• type ConnectorConfig
• type KafkaConsumer
  • func NewKafkaConsumer() (KafkaConsumer, error)
• type KafkaProducer
• type Payload
• type Schema

Variables

var KafkaConsumerSet = wire.NewSet(NewKafkaConsumer)

var KafkaProducerSet = wire.NewSet(NewKafkaProducer)

Functions

func CheckConnector

func CheckConnector(config ConnectorConfig)

func NewKafkaProducer

func NewKafkaProducer() (*kafkaProducer, error)

func RegisterConnector

func RegisterConnector(config ConnectorConfig) *http.Response

Types

type ConnectorConfig

type ConnectorConfig struct {
	ServerUrl       string
	DataType        string
	ConnectorConfig string
	Connector       string
}

type KafkaConsumer

type KafkaConsumer interface {
	Subscribe(topic string, rebalanceCb kafka.RebalanceCb) error
	SubscribeTopics(topics []string, rebalanceCb kafka.RebalanceCb) error
	ReadMessage(timeout time.Duration) (*kafka.Message, error)
	Close() error
}

func NewKafkaConsumer

func NewKafkaConsumer() (KafkaConsumer, error)

type KafkaProducer

type KafkaProducer interface {
	Close()
	Produce(schema *Schema[Payload]) error
}

type Payload

type Payload interface {
	any
}

Payload Struct : For Any Data We Want to Sent To Broker

type Schema

type Schema[T Payload] struct {
	Id        int       `json:"id"`  // You Can Use ID Or Not Use
	Key       string    `json:"key"` // Key or Type of event, we will set as a part of metadata
	Timestamp time.Time `json:"timestamp"`
	Payload   T         `json:"payload"`
}

Schema Event Data When a producer send a message to kafka topic It Can Set Key as a Part Of Message Metadata Kafka uses this key to determine the partition to which the message will be written Ensuring that related messages or events are processed in the order they were produced