// Consume fetches the values. s is a Stream of T.
func (b *Buffer) Consume(s functional.Stream) {
	defer s.Close()
	b.idx, b.err = readStreamIntoSlice(s, b.buffer, b.addrFunc)
	if b.err == functional.Done {
		b.err = nil
	}
}

func readStreamIntoSlice(
	s functional.Stream,
	aSlice reflect.Value,
	addrFunc func(reflect.Value) interface{}) (numRead int, err error) {
	l := aSlice.Len()
	for numRead = 0; numRead < l; numRead++ {
		err = s.Next(addrFunc(aSlice.Index(numRead)))
		if err != nil {
			break
		}
	}
	return
}

// FirstOnly reads the first value from stream storing it in ptr.
// FirstOnly closes the stream.
// FirstOnly returns emptyError if no values were on stream.
func FirstOnly(stream functional.Stream, emptyError error, ptr interface{}) (err error) {
	defer func() {
		closeError := stream.Close()
		if err == nil {
			err = closeError
		}
	}()
	err = stream.Next(ptr)
	if err == functional.Done {
		err = emptyError
		return
	}
	return
}

// Consume prints all Entry values that s emits to stdout. On error, prints
// the error encountered and halts.
func (p Printer) Consume(s functional.Stream) {
	var entry Entry
	err := s.Next(&entry)
	for ; err == nil; err = s.Next(&entry) {
		fmt.Println(&entry)
	}
	if err != functional.Done {
		fmt.Printf("Error happened: %v\n", err)
		s.Close()
	}
}

func (c *errorReportingConsumerForTesting) Consume(s functional.Stream) {
	var x int
	for s.Next(&x) != functional.Done {
		c.count++
	}
}