package source

import (
	"fmt"
	MQTT "github.com/eclipse/paho.mqtt.golang"
	streams "gitee.com/qw_1215/glink"
	"gitee.com/qw_1215/glink/flow"
	"time"
)

type MqttSource struct {
	client MQTT.Client
	topic  string
	out    chan interface{}
}

//获取新的mqtt源
func NewMqttSource(config *MQTT.ClientOptions, topic string) (*MqttSource, error) {
	if clientID := config.ClientID; clientID == "" {
		config.SetClientID(fmt.Sprintf("defaultClientFromGlink:%d", time.Now().UnixNano()))
	}
	//创建连接
	c := MQTT.NewClient(config)
	if token := c.Connect(); token.WaitTimeout(config.WriteTimeout) && token.Wait() && token.Error() != nil {
		return nil, token.Error()
	}
	source := &MqttSource{
		c,
		topic,
		make(chan interface{}),
	}
	go source.init()
	return source, nil
}

// start main loop
func (ms *MqttSource) init() {
	for {
		//if token := ms.client.Subscribe("exp/zw", 0, ms.f); token.Wait() && token.Error() != nil {
		//	log.Fatal(token.Error())
		//	ms.out <-"werwerwer"
		//}
		ms.out <- "werwerwer"
	}
	ms.client.Disconnect(250)
}
func (ms *MqttSource) f(client MQTT.Client, msg MQTT.Message) {
	fmt.Printf("topic: %s\n", msg.Topic())
	fmt.Printf("date: %s\n", msg.Payload())
	ms.out <- msg.Payload()
}

// Via streams data through given flow
func (ms *MqttSource) Via(_flow *flow.Map) streams.Flow {
	flow.DoStream(ms, _flow)
	return _flow
}

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