func HandleMqttMessage(client *MQTT.Client, msg MQTT.Message) {
	var r message.Response
	if err := json.Unmarshal(msg.Payload(), &r); err != nil {
		log.Println(err)
		return
	}
	log.Printf("Incoming: %v", r)
	if ch, ok := chanMap[r.RequestId]; ok {
		ch <- r
	}
}

// NewEvent -
func NewEvent(msg MQTT.Message) (Event, error) {
	e := Event{Message: msg}

	decoder := json.NewDecoder(bytes.NewReader(msg.Payload()))

	if err := decoder.Decode(&e); err != nil {
		return e, err
	}

	if err := e.Validate(); err != nil {
		return e, err
	}

	return e, nil
}

func (d *Device) commandHandler(client *MQTT.Client, msg MQTT.Message) {
	cmd := protocol.Command{}

	err := cmd.UnMarshal(msg.Payload())
	if err != nil {
		fmt.Println(err)
		return
	}

	switch cmd.Head.No {
	case commonCmdGetStatus:
		d.reportStatus(client)
	default:
		fmt.Printf("received command : %v: %v", cmd.Head.No, cmd.Params)
	}
}

func (self *Subscriber) publishHandler(client *MQTT.MqttClient, msg MQTT.Message) {
	body := string(msg.Payload())
	event := pubsub.Parse(body)
	if event == nil {
		return
	}
	self.channelsLock.Lock()
	// fmt.Printf("Event: %+v\n", event)
	for _, ch := range self.channels {
		if ch.filter(event) {
			// fmt.Printf("Sending to: %+v\n", ch.topics)
			ch.C <- event
		}
	}
	self.channelsLock.Unlock()
}

// BrokerHandler -
func (c *Connection) BrokerHandler(client *MQTT.Client, msg MQTT.Message) {
	c.Info(
		"Received new mqtt (worker: %s) - (message: %s) for (topic: %s). Building event now ...",
		c.Name(), msg.Payload(), msg.Topic(),
	)

	event, err := events.NewEvent(msg)

	if err != nil {
		c.Error("Could not handle received event due to (err: %s)", err)
		return
	}

	c.Info("Event successfully created (data: %v)", event)
	c.events <- event
}

func (d *Device) statusHandler(client *MQTT.Client, msg MQTT.Message) {
	status := protocol.Data{}

	err := status.UnMarshal(msg.Payload())
	if err != nil {
		fmt.Println(err)
		return
	}

	fmt.Println("device receiving status set : ")

	for _, one := range status.SubData {
		fmt.Println("subdeviceid : ", one.Head.SubDeviceid)
		fmt.Println("no : ", one.Head.PropertyNum)
		fmt.Println("params : ", one.Params)
	}
}

func (b *Broker) onMessageReceived(client *MQTT.Client, m MQTT.Message) {
	log.Debugf("topic:%s / msg:%s", m.Topic(), m.Payload())

	msg := message.Message{
		Sender: b.Name,
		Type:   message.TypeSubscribed,
		Body:   m.Payload(),
		Topic:  m.Topic(),
	}
	b.GwChan <- msg
}

// messageHandler is called when a new message arrives
func handleMessage(client *MQTT.Client, msg MQTT.Message) {

	// Unmarshal JSON to RxPacket
	var packet shared.RxPacket
	err := json.Unmarshal(msg.Payload(), &packet)
	if err != nil {
		log.WithField("topic", msg.Topic()).WithError(err).Warn("Failed to unmarshal JSON.")
		return
	}

	// Filter messages by gateway
	if gateway != "" && packet.GatewayEui != gateway {
		return
	}

	// Decode payload
	data, err := base64.StdEncoding.DecodeString(packet.Data)
	if err != nil {
		log.WithField("topic", msg.Topic()).WithError(err).Warn("Failed to decode Payload.")
		return
	}

	ctx := log.WithFields(log.Fields{
		"devAddr": packet.NodeEui,
	})

	if showMeta {
		ctx = ctx.WithFields(log.Fields{
			"gatewayEui": packet.GatewayEui,
			"time":       packet.Time,
			"frequency":  *packet.Frequency,
			"dataRate":   packet.DataRate,
			"rssi":       *packet.Rssi,
			"snr":        *packet.Snr,
		})
	}

	if showRaw {
		ctx = ctx.WithField("data", fmt.Sprintf("%x", data))
	}

	if showTiming {
		rawData, err := base64.StdEncoding.DecodeString(packet.Data)
		if err == nil {
			airtime, err := util.CalculatePacketTime(len(rawData), packet.DataRate)
			if err == nil {
				ctx = ctx.WithField("airtime", fmt.Sprintf("%.1f ms", airtime))
			}
		}
	}

	// Check for unprintable characters
	unprintable, _ := regexp.Compile(`[^[:print:]]`)
	if unprintable.Match(data) {
		ctx.Debug("Received Message")
	} else {
		ctx.WithField("message", fmt.Sprintf("%s", data)).Info("Received Message")
	}

}

func mqttMsgToWolfMsg(msg MQTT.Message) models.IncomingPacket {
	inMsg := models.IncomingPacket{}
	inMsg.Message = string(msg.Payload())
	inMsg.Topic = msg.Topic()
	inMsg.QoS = int(msg.Qos())
	return inMsg
}

func MsgRecvfun(client *MQTT.Client, msg MQTT.Message) {
	appUpChan <- msg.Topic()
	fmt.Printf("appUp")
	fmt.Printf("topic:[%s]  ", msg.Topic())
	fmt.Printf("Mesg:%s\n", msg.Payload())

}

func messageHandler(client *mqtt.MqttClient, message mqtt.Message) {
	var data interface{}
	if strings.HasSuffix(options.ContentType, "json") {
		err = json.Unmarshal(message.Payload(), &data)
		if err != nil {
			log.Println("ERROR unmarshaling the JSON message:", err.Error())
			return
		}
	} else {
		// TODO: support other content-types
		log.Printf("WARNING processing of %s is not supported", options.ContentType)
		return
	}
	prop, err := propTemplate.Fill(data)

	if err != nil {
		log.Println("ERROR filling template with data: ", err.Error())
		return
	}
	out, _ := json.Marshal(prop)
	propPort.SendMessage(runtime.NewPacket(out))
}

func onMessageReceived(client *MQTT.MqttClient, message MQTT.Message) {
	fmt.Printf("Received message on topic: %s\n", message.Topic())
	fmt.Printf("Message: %s\n", message.Payload())

	if string(message.Payload()) == "ぬるぽ" {
		Publish(client, "say", "ガッ")
	}
}

func (m *MqttClient) onMessageReceived(client *MQTT.Client, message MQTT.Message) {
	log.Infof("topic:%s", message.Topic())

	// Remove topic root
	ct := strings.TrimRight(m.Config.Topic, "#")
	topic := strings.Replace(message.Topic(), ct, "", 1)

	chun := Message{
		Topic:   topic,
		Payload: message.Payload(),
	}

	m.mqttChan <- chun
}

func actionHandler(client *MQTT.Client, message MQTT.Message) {
	fmt.Println("Received action message on", message.Topic(), "-", string(message.Payload()))
	action := strings.ToLower(string(message.Payload()))
	switch action {
	case "off":
		host.LedsOff()
	case "on":
		host.LedsOn()
	case "toggle":
		host.LedsToggle()
	case "slide":
		host.LedsCycle(3)
	}
}

func (d *Device) messageHandler(client *MQTT.Client, msg MQTT.Message) {
	fmt.Printf("TOPIC: %s\n", msg.Topic())
	fmt.Printf("MSG: %x\n", msg.Payload())
	msgtype := msg.Topic()
	fmt.Println(msgtype)

	switch msgtype {
	case "c":
		d.commandHandler(client, msg)
	case "s":
		d.statusHandler(client, msg)
	default:
		fmt.Println("unsuported message type :", msgtype)
	}
}

func messageReceived(client *MQTT.Client, msg MQTT.Message) {
	topics := strings.Split(msg.Topic(), "/")
	msgFrom := topics[len(topics)-1]
	fmt.Print(msgFrom + ": " + string(msg.Payload()))
}

func brokerClientsHandler(client *MQTT.Client, msg MQTT.Message) {
	brokerClients <- true
	fmt.Printf("BrokerClientsHandler      ")
	fmt.Printf("[%s]  ", msg.Topic())
	fmt.Printf("%s\n", msg.Payload())
}

func onMessageReceived(client *MQTT.MqttClient, message MQTT.Message) {
	log.Infof("topic:%s  / msg:%s", message.Topic(), message.Payload())
	fmt.Println(string(message.Payload()))
}

func onMessageReceived(client *MQTT.MqttClient, message MQTT.Message) {
	fmt.Printf("Received message on topic: %s\n", message.Topic())
	fmt.Printf("Message: %s\n", message.Payload())
}

func (b *Bus) decodeRequest(msg *mqtt.Message, data interface{}) error {
	return json.NewDecoder(bytes.NewBuffer(msg.Payload())).Decode(data)
}