package scanproto

import (
	"log"

	"code.justin.tv/ear/kinesis/multilangclient"
)

// implements scanproto.EventSource and multilangclient.Consumer
type kinesisEventSource struct {
	events chan *TimestampedEvent
	errors chan error
}

func NewKinesisEventSource() EventSource {
	return &kinesisEventSource{
		events: make(chan *TimestampedEvent, readBuffer),
		errors: make(chan error, 1),
	}
}

// Multilang consumer methods
func (s *kinesisEventSource) Initialize(shardID string) error {
	return nil
}

func (s *kinesisEventSource) Process(record []byte) error {
	events, err := decodeSingleEventSet(record)
	if err != nil {
		return err
	}
	for _, e := range events {
		s.events <- e
	}
	return nil
}

func (s *kinesisEventSource) Shutdown() error {
	s.Stop()
	return nil
}

// EventSource methods
func (s *kinesisEventSource) Run() {
	err := multilangclient.Run(s)
	if err != nil {
		log.Fatalf("multilangclient err=%q", err)
	}
}

func (s *kinesisEventSource) Events() <-chan *TimestampedEvent {
	return s.events
}
func (s *kinesisEventSource) Errors() <-chan error {
	return s.errors
}

// Stop will close the kinesisEventSource's output channels, marking
// it as having finished processing. In general, this should only be
// called when the MultiLangDaemon has issued a shutdown command to
// the client, which means that it should really only be called
// through kinesisEventSource.Shutdown. Calling it from elsewhere is
// likely to cause a panic as the kinesisEventSource will attempt to
// write to a closed channel if it receives a record from the
// MultiLangDaemon.
func (s *kinesisEventSource) Stop() {
	close(s.events)
	close(s.errors)
}
