shovel

Documentation

Index

• type Config
• type Shovel
  • func New(cfg Config) *Shovel
  • func (consumer *Shovel) ConsumeWithTimeout(wg *sync.WaitGroup, timeout time.Duration)
  • func (consumer *Shovel) Process(msg *kafka.Message) error

Types

type Config

type Config struct {
	SourceTopicConsumer *kafka.Consumer
	Logger              *zerolog.Logger
	Rdb                 *redis.Client
	Producer            *producer.CommonProducer
	SourceTopicName     string
	TargetTopicName     string
	PoisonedTopicName   string
	Name                string
	MaxErrorCount       int
	IsPoisonedTopic     bool
}

type Shovel

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

func New

func New(cfg Config) *Shovel

func (*Shovel) ConsumeWithTimeout

func (consumer *Shovel) ConsumeWithTimeout(wg *sync.WaitGroup, timeout time.Duration)

func (*Shovel) Process

func (consumer *Shovel) Process(msg *kafka.Message) error