Function TestInterceptors

functional_producer_test.go:869–939

(t *testing.T)

}

func TestInterceptors(t *testing.T) {
	config := NewFunctionalTestConfig()
	setupFunctionalTest(t)
	defer teardownFunctionalTest(t)

	config.Producer.Return.Successes = true
	config.Consumer.Return.Errors = true
	config.Producer.Interceptors = []ProducerInterceptor{&appendInterceptor{i: 0}, &appendInterceptor{i: 100}}
	config.Consumer.Interceptors = []ConsumerInterceptor{&appendInterceptor{i: 20}}

	client, err := NewClient(FunctionalTestEnv.KafkaBrokerAddrs, config)
	if err != nil {
		t.Fatal(err)
	}
	defer safeClose(t, client)

	initialOffset, err := client.GetOffset("test.1", 0, OffsetNewest)
	if err != nil {
		t.Fatal(err)
	}

	producer, err := NewAsyncProducerFromClient(client)
	if err != nil {
		t.Fatal(err)
	}

	for i := 0; i < 10; i++ {
		producer.Input() <- &ProducerMessage{Topic: "test.1", Key: nil, Value: StringEncoder(TestMessage)}
	}

	for i := 0; i < 10; i++ {
		select {
		case msg := <-producer.Errors():
			t.Error(msg.Err)
		case msg := <-producer.Successes():
			v, _ := msg.Value.Encode()
			expected := TestMessage + strconv.Itoa(i) + strconv.Itoa(i+100)
			if string(v) != expected {
				t.Errorf("Interceptor should have incremented the value, got %s, expected %s", v, expected)
			}
		}
	}
	safeClose(t, producer)

	master, err := NewConsumerFromClient(client)
	if err != nil {
		t.Fatal(err)
	}
	consumer, err := master.ConsumePartition("test.1", 0, initialOffset)
	if err != nil {
		t.Fatal(err)
	}

	for i := 0; i < 10; i++ {
		select {
		case <-time.After(10 * time.Second):
			t.Fatal("Not received any more events in the last 10 seconds.")
		case err := <-consumer.Errors():