package kafka

import (
	"fmt"
	"strconv"
	"testing"
)

func Test_partitionConsumer_Messages(t *testing.T) {
	var addrs = []string{"10.105.36.200:9092", "10.105.4.216:9092"}
	var config = Config{
		Version: "V1_1_1_0",
	}
	config.Producer.Return.Successes = true
	client, err := NewClient(addrs, &config)
	if err != nil {
		t.Errorf("newClient err:%v", err)
		return
	}
	consumer, err := client.Consumer()
	if err != nil {
		t.Errorf("Consumer() err:%v", err)
		return
	}
	topic := "scrm-test-demo-1"
	partitions, err := consumer.Partitions(topic)
	if err != nil {
		t.Errorf("Partitions() err:%v", err)
		return
	}
	for _, p := range partitions {
		pc, err := consumer.ConsumePartition(topic, p, OffsetOldest)
		if err != nil {
			t.Errorf("ConsumePartition() err:%v", err)
			return
		}
		msgs, err := pc.Messages(2)
		if err != nil {
			t.Errorf("Messages() err:%v", err)
			return
		}
		for _, v := range msgs {
			fmt.Println("msgs========")
			fmt.Println(string(v.Value), "-", strconv.FormatInt(v.Offset, 10))
		}

		msgs, err = pc.Messages(2)
		if err != nil {
			t.Errorf("Messages() err:%v", err)
			return
		}
		for _, v := range msgs {
			fmt.Println("msgs========")
			fmt.Println(string(v.Value), "-", strconv.FormatInt(v.Offset, 10))
		}
		pc.AsyncClose()
	}
}