package main

import (
	"context"
	"fmt"
	"time"

	kafka "gitee.com/fanfansong/gokafka/gokafka"
)

type ConsumerHandelr struct {
}

func (handler ConsumerHandelr) Handle(ctx context.Context, topic string, messages []byte) error {
	fmt.Println("Topic:", topic)
	fmt.Println("message:", string(messages))
	return nil
}

func complexConsume() {
	//kafkaUrl使用逗号分割
	handler := ConsumerHandelr{}
	consumer, err := kafka.NewConsumer("test", "192.168.150.221:9092")
	if err != nil {
		println(err)
		return
	}
	groupId := "myTest"

	err = consumer.AddGroup(groupId, handler)
	if err != nil {
		println(err)
		return
	}
	defer consumer.Close()

	ctx := context.Background()
	topics := []string{"maggie_test", "test"}
	go consumer.ConsumeTopicsByGroup(ctx, groupId, topics)

	time.Sleep(time.Second * 10)
}

func simpleConsume() {
	handler := ConsumerHandelr{}
	consumer, err := kafka.NewConsumer("test", "192.168.150.221:9092")
	if err != nil {
		println(err)
		return
	}
	defer consumer.Close()

	ctx := context.Background()
	go consumer.ConsumeTopic(ctx, "maggie_test", handler)

	time.Sleep(time.Second * 10)
}

func syncProduct() {
	producer, err := kafka.NewSyncProducer("192.168.150.221:9092")
	if err != nil {
		fmt.Println(err)
	}
	defer producer.Close()

	producer.SendMessage("maggie_test", []byte("!!!!!!test!!!!!!!!"))
}

// func main() {
// 	syncProduct()
// 	return
// }