package sink

import (
	"io"

	"code.justin.tv/ear/kinesis/combiner"
	"code.justin.tv/release/trace/internal/pipeline"
	"golang.org/x/net/context"
)

type writerSink struct {
	w   io.Writer
	ctx context.Context
	eh  pipeline.ErrorHandler
}

func (s *writerSink) run(data <-chan *combiner.Blob) {
	for {
		select {
		case <-s.ctx.Done():
			return
		case b, ok := <-data:
			if !ok {
				return
			}
			_, err := s.w.Write(b.Data)
			if err != nil {
				s.eh(err)
			}
		}
	}

}

// WriterSink writes blob data to an io.Writer, without any separator.
func WriterSink(ctx context.Context, w io.Writer, eh pipeline.ErrorHandler) Sink {
	s := &writerSink{
		w:   w,
		ctx: ctx,
		eh:  eh,
	}
	return s.run
}
