Function readRecords

fetch_test.go:170–207

(records RecordReader)

}

func readRecords(records RecordReader) ([]memoryRecord, error) {
	list := []memoryRecord{}

	for {
		rec, err := records.ReadRecord()

		if err != nil {
			if errors.Is(err, io.EOF) {
				return list, nil
			}
			return nil, err
		}

		var (
			offset      = rec.Offset
			key         = rec.Key
			value       = rec.Value
			headers     = rec.Headers
			bytesKey    []byte
			bytesValues []byte
		)

		if key != nil {
			bytesKey, _ = ioutil.ReadAll(key)
		}

		if value != nil {
			bytesValues, _ = ioutil.ReadAll(value)
		}

		list = append(list, memoryRecord{
			offset:  offset,
			key:     bytesKey,
			value:   bytesValues,
			headers: headers,
		})
	}
}

func TestClientPipeline(t *testing.T) {
	client, topic, shutdown := newLocalClientAndTopic()