kafkaworker

README

Example

Create delivery handler

package workerhandler

import (
	"context"
	"encoding/json"
	
	"github.com/golangid/candi/candishared"
	"github.com/golangid/candi/codebase/factory/types"
	"github.com/golangid/candi/tracer"
)

// KafkaHandler struct
type KafkaHandler struct {
}

// NewKafkaHandler constructor
func NewKafkaHandler() *KafkaHandler {
	return &KafkaHandler{}
}

// MountHandlers return group map topic to handler func
func (h *KafkaHandler) MountHandlers(group *types.WorkerHandlerGroup) {

	group.Add("example-topic", h.handleExampleTopic) // handling consume topic "example-topic"
}

func (h *KafkaHandler) handleExampleTopic(eventContext *candishared.EventContext) error {
	trace := tracer.StartTrace(eventContext.Context(), "DeliveryKafkaConsumer:HandleExampleTopic")
	defer trace.Finish()

	// process usecase
	return nil
}

Register in module

package examplemodule

import (

	"example.service/internal/modules/examplemodule/delivery/workerhandler"

	"github.com/golangid/candi/codebase/factory/dependency"
	"github.com/golangid/candi/codebase/factory/types"
	"github.com/golangid/candi/codebase/interfaces"
)

type Module struct {
	// ...another delivery handler
	workerHandlers map[types.Worker]interfaces.WorkerHandler
}

func NewModules(deps dependency.Dependency) *Module {
	return &Module{
		workerHandlers: map[types.Worker]interfaces.WorkerHandler{
			// ...another worker handler
			// ...
			types.Kafka: workerhandler.NewKafkaHandler(),
		},
	}
}

// ...another method

Documentation

Index

• func NewWorker(service factory.ServiceFactory, bk interfaces.Broker, opts ...OptionFunc) factory.AppServerFactory
• type OptionFunc
  • func SetConsumerGroup(consumerGroup string) OptionFunc
  • func SetDebugMode(debugMode bool) OptionFunc
  • func SetMaxGoroutines(maxGoroutines int) OptionFunc

Functions

func NewWorker

func NewWorker(service factory.ServiceFactory, bk interfaces.Broker, opts ...OptionFunc) factory.AppServerFactory

NewWorker create new kafka consumer

Types

type OptionFunc

type OptionFunc func(*option)

OptionFunc type

func SetConsumerGroup

func SetConsumerGroup(consumerGroup string) OptionFunc

SetConsumerGroup option func, for log when startup

func SetDebugMode

func SetDebugMode(debugMode bool) OptionFunc

SetDebugMode option func

func SetMaxGoroutines

func SetMaxGoroutines(maxGoroutines int) OptionFunc

SetMaxGoroutines option func