package main
import (
"context"
"errors"
"fmt"
"github.com/tuanuet/retry-kafka/producer"
"github.com/tuanuet/retry-kafka/retriable"
"os"
"os/signal"
"syscall"
"time"
"github.com/tuanuet/retry-kafka/consumer"
)
// User example models
type User struct {
ID uint32
Name string
Age int32
}
// UserEvent should implement Event interface
type UserEvent struct {
User
}
func (u UserEvent) GetTopicName() string {
return "evt.kafka.user"
}
func (u UserEvent) GetPartitionValue() string {
return fmt.Sprintf("%d", u.ID)
}
func main() {
publisher := producer.NewProducer(&UserEvent{}, []string{"localhost:9092"})
for i := 0; i < 100; i++ {
if err := publisher.SendMessage(&UserEvent{
User{
ID: uint32(i),
Name: "tuan",
Age: 27,
},
}, nil); err != nil {
panic(err)
}
}
c := consumer.NewConsumer(
"test_consumer",
&UserEvent{},
[]string{"localhost:9092"},
consumer.WithBatchFlush(10, 500*time.Millisecond),
consumer.WithRetries([]consumer.RetryOption{
{Pending: 10 * time.Second},
{Pending: 15 * time.Second},
}),
)
//err := c.Consume(context.Background(), func(evt retriable.Event, headers []*retriable.Header) error {
// u := evt.(*UserEvent)
// fmt.Println(u.Age)
//
// return errors.New("aaaa")
//})
//err := c.Consume(context.Background(), func(evt retriable.Event, headers []*retriable.Header) error {
// fmt.Println("===========================================================")
// u := evt.(*UserEvent)
// fmt.Println(u.ID)
// return errors.New("errr")
//})
go func() {
err := c.BatchConsume(context.Background(), func(evts []retriable.Event, headers [][]*retriable.Header) error {
fmt.Println("============================================")
fmt.Println(len(evts))
fmt.Println(len(evts))
us := make([]*UserEvent, 0)
for _, evt := range evts {
u := evt.(*UserEvent)
us = append(us, u)
}
return errors.New("aaaa")
})
fmt.Println("[2] doing")
if err != nil {
panic(err)
}
}()
signals := make(chan os.Signal, 1)
signal.Notify(signals, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
<-signals
c.Close()
println("close function!")
}
tuanuet / retry-kafka Goto Github PK
View Code? Open in Web Editor NEWLicense: Apache License 2.0