package gqueue

import (
	"gitee.com/zr233/gqueue/adapter"
	"github.com/Shopify/sarama"
)

var core = NewQueue()
var producer adapter.Producer

func init() {
	_, _ = core.UseOrigin(adapter.NewOriginConfig())
}

type Queue struct {
	adapter.Adaptor
}

func NewQueue() *Queue {
	q := &Queue{}
	return q
}

func (q *Queue) UseKafka(zkHosts []string, config *sarama.Config) (q_ *Queue, err error) {
	q_ = q
	q.Adaptor, err = adapter.NewKafka(zkHosts, config)
	return
}

func (q *Queue) UseOrigin(cfg *adapter.OriginConfig) (q_ *Queue, err error) {
	q_ = q
	q.Adaptor, err = adapter.NewOrigin(cfg)
	return
}

func NewMessage() *sarama.ProducerMessage {
	return &sarama.ProducerMessage{}
}

func NewProducer() (p adapter.Producer, err error) {
	p, err = core.NewProducer()
	producer = p
	return
}

// Errors is the error output channel back to the user. You MUST read from this
// channel or the Producer will deadlock when the channel is full. Alternatively,
// you can set Producer.Return.Errors in your config to false, which prevents
// errors to be returned.
func Errors() <-chan *sarama.ProducerError {
	return producer.Errors()
}

func NewConsumer(topic, groupId string) (p adapter.Consumer, err error) {
	return core.NewConsumer(topic, groupId)
}

func Input() chan<- *sarama.ProducerMessage {
	return producer.Input()
}