kafka_connect

Documentation

Index

• Variables
• func CloseConnection()
• func Connect()
• func ConnectTopic(TopicName, GroupID string) *kafka.Reader
• func Connect_err() error
• func CreateClient() error
• func FillSettings()
• func GetAddr() net.Addr
• func GetOffsetFromGroupID(TopicName, GroupID string) (int64, error)
• func LogInfo_Connected(err error)
• func StartKafka()
• func Start_ctx(ctx *context.Context, WaitGroup *sync.WaitGroup) error
• func WaitStop()
• type SettingsINI

Variables

var Client *kafka.Client

Client - клиент для Kafka

var Conn *kafka.Conn

Conn - соединение к серверу nats

Functions

func CloseConnection

func CloseConnection()

CloseConnection - закрывает соединение с сервером Kafka

func Connect

func Connect()

Connect - подключается к серверу Kafka

func ConnectTopic

func ConnectTopic(TopicName, GroupID string) *kafka.Reader

ConnectTopic - подключает кафку к нужному топику

func Connect_err

func Connect_err() error

Connect_err - подключается к серверу Kafka и возвращает ошибку

func CreateClient

func CreateClient() error

CreateClient - создаёт клиент для Kafka

func FillSettings

func FillSettings()

FillSettings загружает переменные окружения в структуру из файла или из переменных окружения

func GetAddr

func GetAddr() net.Addr

GetAddr - создаёт Addr

func GetOffsetFromGroupID

func GetOffsetFromGroupID(TopicName, GroupID string) (int64, error)

GetOffsetFromGroupID - получает оффсет группы для конкретного топика, партиция 0

func LogInfo_Connected

func LogInfo_Connected(err error)

LogInfo_Connected - выводит сообщение в Лог, или паника при ошибке

func StartKafka

func StartKafka()

StartKafka - необходимые процедуры для подключения к серверу Kafka

func Start_ctx

func Start_ctx(ctx *context.Context, WaitGroup *sync.WaitGroup) error

Start_ctx - необходимые процедуры для подключения к серверу Kafka Свой контекст и WaitGroup нужны для остановки работы сервиса Graceful shutdown Для тех кто пользуется этим репозиторием для старта и останова сервиса можно просто StartKafka()

func WaitStop

func WaitStop()

WaitStop - ожидает отмену глобального контекста

Types

type SettingsINI

type SettingsINI struct {
	KAFKA_HOST     string
	KAFKA_PORT     string
	KAFKA_LOGIN    string
	KAFKA_PASSWORD string
}

SettingsINI - структура для хранения всех нужных переменных окружения

var Settings SettingsINI

Settings хранит все нужные переменные окружения