gousukafka

README

Kafka-Integration for Go Universal Service Utilities

Full docu for go-gousu on https://github.com/indece-official/go-gousu

Requirements

Services & Controllers

Services

Name	Description	Interface	Factory
kafka	Kafka client service for consuming & producing kafka messages	gousukafka.IService	gousukafka.NewService

Controllers

Name	Description	Interface	Struct
kafka	Base controller for consuming one kafka topic	gousukafka.IController	gousukafka.NewControllerBase

Usage

TODO

Documentation

Index

• Constants
• func NewService(ctx gousu.IContext) gousu.IService
• type IService
• type MockService
  • func NewMockService() *MockService
  • func (s *MockService) Done(msg *kafka.Message, err error)
  • func (s *MockService) Produce(topic string, value []byte) error
  • func (s *MockService) Subscribe(topic string) (chan *kafka.Message, error)
• type Service
  • func (s *Service) Done(msg *kafka.Message, err error)
  • func (s *Service) Health() error
  • func (s *Service) Name() string
  • func (s *Service) Produce(topic string, value []byte) error
  • func (s *Service) Start() error
  • func (s *Service) Stop() error
  • func (s *Service) Subscribe(topic string) (chan *kafka.Message, error)

Constants

const ServiceName = "kafka"

ServiceName is the name of the kafka service

Functions

func NewService

func NewService(ctx gousu.IContext) gousu.IService

NewService creates a new initialized instance of Service

Types

type IService

type IService interface {
	gousu.IService

	Subscribe(topic string) (chan *kafka.Message, error)
	Produce(topic string, value []byte) error
	Done(msg *kafka.Message, err error)
}

IService defines the interface of the kafka service

type MockService

type MockService struct {
	gousu.MockService

	DoneFunc            func(msg *kafka.Message, err error)
	SubscribeFunc       func(topic string) (chan *kafka.Message, error)
	ProduceFunc         func(topic string, value []byte) error
	DoneFuncCalled      int
	SubscribeFuncCalled int
	ProduceFuncCalled   int
}

MockService for simply mocking IService

func NewMockService

func NewMockService() *MockService

NewMockService creates a new initialized instance of MockService

func (*MockService) Done

func (s *MockService) Done(msg *kafka.Message, err error)

Done calls DoneFunc and increases DoneFuncCalled

func (*MockService) Produce

func (s *MockService) Produce(topic string, value []byte) error

Produce calls ProduceFunc and increases ProduceFuncCalled

func (*MockService) Subscribe

func (s *MockService) Subscribe(topic string) (chan *kafka.Message, error)

Subscribe calls SubscribeFunc and increases SubscribeFuncCalled

type Service

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

Service provides a service for basic kafka client functionality

func (*Service) Done

func (s *Service) Done(msg *kafka.Message, err error)

Done must be called after receiving a message via Subscribe(...)

func (*Service) Health

func (s *Service) Health() error

Health provides information about if the Service is healthy

func (*Service) Name

func (s *Service) Name() string

Name returns the name of the kafka service from ServiceName

func (*Service) Produce

func (s *Service) Produce(topic string, value []byte) error

Produce emits a message to a topic The topic is automatically created (when enabled on the kafka server) if it doesn't exist

func (*Service) Start

func (s *Service) Start() error

Start starts the KafkaService and connects the Kafka consumer

func (*Service) Stop

func (s *Service) Stop() error

Stop closes kafka consumer & producer

func (*Service) Subscribe

func (s *Service) Subscribe(topic string) (chan *kafka.Message, error)

Subscribe subscribes to a topic and returns a channel

Important: After receiving a message on the returned channel,

Done(...) must be called, else the function will block