// メッセージを送信する。
func Publish(client *MQTT.Client, topic string, qos byte, retain bool, message string) {
	token := client.Publish(topic, qos, retain, message)

	if token.Wait() && token.Error() != nil {
		fmt.Printf("Publish error: %s\n", token.Error())
	}
}

func loop(reader io.Reader, client *MQTT.Client) {
	channel := make(chan data.SensorData)
	scanner := bufio.NewScanner(reader)
	go func() {
		for scanner.Scan() {
			line := strings.SplitN(scanner.Text(), ":", 2)
			if nil == sensors.Sensors[line[0]] {
				jww.DEBUG.Println(scanner.Text())
			} else {
				d := sensors.Sensors[line[0]].Parse(line[0], line[1])
				channel <- d
			}
		}
		close(channel)
	}()

	for data := range channel {
		topic := "/gowx/sample"
		buf := new(bytes.Buffer)
		encoder := json.NewEncoder(buf)
		encoder.Encode(data)
		payload := buf.Bytes()
		if token := client.Publish(topic, 0, false, payload); token.Wait() && token.Error() != nil {
			jww.ERROR.Println("Failed to send message.", token.Error())
		}
		jww.DEBUG.Printf("Publishing %s -> %s\n", topic, buf.Bytes())
	}
}

func (d *Device) reportEvent(client *MQTT.Client) {
	for {
		time.Sleep(3 * time.Second)

		event := protocol.Event{}

		params, err := tlv.MakeTLVs([]interface{}{"hello event."})
		if err != nil {
			fmt.Println(err)
			return
		}

		event.Params = params
		event.Head.No = 1
		event.Head.SubDeviceid = 1
		event.Head.ParamsCount = uint16(len(params))

		payload, err := event.Marshal()
		if err != nil {
			fmt.Println(err)
			return
		}

		client.Publish("e", 1, false, payload)

	}
}

func setupSubscriptions(client *mqtt.Client, wg *sync.WaitGroup) {
	if topicFiltersString != "" {

		wg.Add(numberMessagesExpected)

		topicFilters := strings.Split(topicFiltersString, ",")
		for _, filter := range topicFilters {
			client.Subscribe(filter, qos, func(client *mqtt.Client, msg mqtt.Message) {
				if !msg.Retained() || !ignoreRetained {
					payload := bytes.NewBuffer(msg.Payload()).String()

					if ignorePayload {
						fmt.Printf("r:%v, t:%v, s:%v\n", msg.Retained(), msg.Topic(), len(payload))
					} else {
						infoF("r:%v, t:%v, s:%v", msg.Retained(), msg.Topic(), len(payload))
						fmt.Printf("%v\n", payload)
					}
					wg.Done()
				} else {
					info("ignoring retained msg")
				}
			})
		}

		infoF("waiting for %v msgs.", numberMessagesExpected)
	} else {
		info("No subscriptions...")
	}
}

func (m *MQTTClient) SubscribeOnConnect(client *MQTT.Client) {
	log.Infof("client connected")

	if len(m.Subscribed) > 0 {
		token := client.SubscribeMultiple(m.Subscribed, m.onMessageReceived)
		token.Wait()
		if token.Error() != nil {
			log.Error(token.Error())
		}
	}
}

func publishMessage(client *mqtt.Client) {
	if pubTopic != "" {
		token := client.Publish(pubTopic, qos, false, pubMessage)
		token.WaitTimeout(10 * time.Second)
		if token.Error() != nil {
			infoF("FATAL: Could not publish: %v", token.Error())
			os.Exit(-1)
		}
	} else {
		info("nothing to publish")
	}
}

func (b *Broker) SubscribeOnConnect(client *MQTT.Client) {
	log.Infof("client connected")
	b.connected = true

	if b.Subscribed.Length() > 0 {
		// subscribe
		token := client.SubscribeMultiple(b.Subscribed.List(), b.onMessageReceived)
		token.Wait()
		if token.Error() != nil {
			log.Error(token.Error())
		}
	}
}

// メッセージを受信する。
func Subscribe(client *MQTT.Client, topic string, qos byte) *SubscribeResult {
	var result *SubscribeResult = &SubscribeResult{}
	result.Count = 0

	var handler MQTT.MessageHandler = func(client *MQTT.Client, msg MQTT.Message) {
		result.Count++
		if Debug {
			fmt.Printf("Received message : topic=%s, message=%s\n", msg.Topic(), msg.Payload())
		}
	}

	token := client.Subscribe(topic, qos, handler)

	if token.Wait() && token.Error() != nil {
		fmt.Printf("Subscribe error: %s\n", token.Error())
	}

	return result
}

func SendData(client *MQTT.Client, endChan chan struct{}) {
	ticker := time.NewTicker(1 * time.Second)
	for {
		select {
		case <-endChan:
			return
		case <-ticker.C:
			var p Payload
			p.DataPoints.Load1, p.DataPoints.Load5, p.DataPoints.Load15 = getLoadAvg()
			p.DataPoints.CPUTemp = getCPUTemp(host.CPUTempFile())
			p.Timestamp = time.Now()
			payloadBytes, err := json.Marshal(p)
			if err == nil {
				client.Publish(config.PubTopic, 0, false, payloadBytes)
			} else {
				fmt.Println(err.Error())
			}
		}
	}
}

func (d *Device) reportStatus(client *MQTT.Client) {

	payloadHead := protocol.DataHead{
		Flag:      0,
		Timestamp: uint64(time.Now().Unix() * 1000),
	}
	param := []interface{}{uint8(1)}
	params, err := tlv.MakeTLVs(param)
	if err != nil {
		fmt.Println(err)
		return
	}
	sub := protocol.SubData{
		Head: protocol.SubDataHead{
			SubDeviceid: uint16(1),
			PropertyNum: uint16(1),
			ParamsCount: uint16(len(params)),
		},
		Params: params,
	}

	status := protocol.Data{
		Head:    payloadHead,
		SubData: []protocol.SubData{},
	}

	status.SubData = append(status.SubData, sub)

	payload, err := status.Marshal()
	if err != nil {
		fmt.Println(err)
		return
	}

	client.Publish("s", 1, false, payload)

}

// Brokerとの接続を切断する。
func Disconnect(client *MQTT.Client) {
	client.Disconnect(10)
}

func makeHandler(c *MQTT.Client, cmd string) func(event lirc.Event) {
	return func(event lirc.Event) {
		c.Publish("/lights/set", 0, false, cmd)
	}
}

// Brokerとの接続を切断する。
func Disconnect(client *MQTT.Client) {
	client.ForceDisconnect()
}