package source

import (
	streams "gitee.com/qw_1215/glink"
	"gitee.com/qw_1215/glink/flow"
	"github.com/nsqio/go-nsq"
	"time"
)

type NsqSource struct {
	consumer *nsq.Consumer
	url      string
	out      chan interface{}
}

func NewNsqSource(url string, topic string, channel string) (*NsqSource, error) {
	cfg := nsq.NewConfig()
	cfg.LookupdPollInterval = time.Second          //设置重连时间
	c, err := nsq.NewConsumer(topic, channel, cfg) // 新建一个消费者
	sq := &NsqSource{
		consumer: c,
		url:      url,
		out:      make(chan interface{}),
	}
	streams.Check(err)

	go sq.init()
	return sq, nil
}

func (rs *NsqSource) init() error {
	rs.consumer.SetLogger(nil, 0) //屏蔽系统日志
	rs.consumer.AddHandler(rs)
	// 建立一个nsqd连接
	if err := rs.consumer.ConnectToNSQD(rs.url); err != nil {
		return err
	}
	return nil
}

//处理消息
func (rs *NsqSource) HandleMessage(msg *nsq.Message) error {
	rs.out <- msg
	return nil
}

// Via streams data through given flow
func (rs *NsqSource) Via(_flow streams.Flow) streams.Flow {
	flow.DoStream(rs, _flow)
	return _flow
}

// Out returns channel for sending data
func (rs *NsqSource) Out() <-chan interface{} {
	return rs.out
}