Function TestMessageSetReader

message_test.go:239–536

(t *testing.T)

}

func TestMessageSetReader(t *testing.T) {
	const startOffset = 1000
	const highWatermark = 5000
	const topic = "test-topic"
	msgs := make([]Message, 100)
	for i := 0; i < 100; i++ {
		msgs[i] = Message{
			Time:   time.Now(),
			Offset: int64(i + startOffset),
			Key:    []byte(fmt.Sprintf("key-%d", i)),
			Value:  []byte(fmt.Sprintf("val-%d", i)),
			Headers: []Header{
				{
					Key:   fmt.Sprintf("header-key-%d", i),
					Value: []byte(fmt.Sprintf("header-value-%d", i)),
				},
			},
		}
	}
	defaultHeader := fetchResponseHeader{
		highWatermarkOffset: highWatermark,
		lastStableOffset:    highWatermark,
		topic:               topic,
	}
	for _, tc := range []struct {
		name    string
		builder fetchResponseBuilder
		err     error
		debug   bool
	}{
		{
			name: "empty",
			builder: fetchResponseBuilder{
				header: defaultHeader,
			},
			err: errShortRead,
		},
		{
			name: "v0",
			builder: fetchResponseBuilder{
				header: defaultHeader,
				msgSets: []messageSetBuilder{
					v0MessageSetBuilder{
						msgs: []Message{msgs[0]},
					},
				},
			},
		},
		{
			name: "v0 compressed",
			builder: fetchResponseBuilder{
				header: defaultHeader,
				msgSets: []messageSetBuilder{
					v0MessageSetBuilder{
						codec: new(gzip.Codec),
						msgs:  []Message{msgs[0]},
					},
				},