<commit_before>package main

import (
	"crypto/rand"
	"crypto/tls"
	"fmt"
	"time"

	MQTT "git.eclipse.org/gitroot/paho/org.eclipse.paho.mqtt.golang.git"
	log "github.com/Sirupsen/logrus"
	"github.com/codegangsta/cli"
)

var MaxClientIdLen = 8

type MQTTClient struct {
	Client *MQTT.MqttClient
	Opts   *MQTT.ClientOptions
}

// Connects connect to the MQTT broker with Options.
func (m *MQTTClient) Connect() (*MQTT.MqttClient, error) {
	m.Client = MQTT.NewClient(m.Opts)
	_, err := m.Client.Start()
	if err != nil {
		return nil, err
	}
	return m.Client, nil
}

func (m *MQTTClient) Publish(topic string, payload []byte, qos int, retain bool) error {
	mqttmsg := MQTT.NewMessage(payload)
	// FIXME: validate qos number
	mqttmsg.SetQoS(MQTT.QoS(qos))
	mqttmsg.SetRetainedFlag(retain)

	receipt := m.Client.PublishMessage(topic, mqttmsg)
	<-receipt

	return nil
}

func onMessageReceived(client *MQTT.MqttClient, message MQTT.Message) {
	fmt.Println(string(message.Payload()))
}

func (m *MQTTClient) Subscribe(topic string, qos int) error {
	topicFilter, err := MQTT.NewTopicFilter(topic, byte(qos))
	if err != nil {
		return err
	}
	_, err = m.Client.StartSubscription(onMessageReceived, topicFilter)
	if err != nil {
		return err
	}
<commit_msg>show comming message topic when verbose mode.<commit_after>package main

import (
	"crypto/rand"
	"crypto/tls"
	"fmt"
	"time"

	MQTT "git.eclipse.org/gitroot/paho/org.eclipse.paho.mqtt.golang.git"
	log "github.com/Sirupsen/logrus"
	"github.com/codegangsta/cli"
)

var MaxClientIdLen = 8

type MQTTClient struct {
	Client *MQTT.MqttClient
	Opts   *MQTT.ClientOptions
}

// Connects connect to the MQTT broker with Options.
func (m *MQTTClient) Connect() (*MQTT.MqttClient, error) {
	m.Client = MQTT.NewClient(m.Opts)
	_, err := m.Client.Start()
	if err != nil {
		return nil, err
	}
	return m.Client, nil
}

func (m *MQTTClient) Publish(topic string, payload []byte, qos int, retain bool) error {
	mqttmsg := MQTT.NewMessage(payload)
	// FIXME: validate qos number
	mqttmsg.SetQoS(MQTT.QoS(qos))
	mqttmsg.SetRetainedFlag(retain)

	receipt := m.Client.PublishMessage(topic, mqttmsg)
	<-receipt

	return nil
}

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

func (m *MQTTClient) Subscribe(topic string, qos int) error {
	topicFilter, err := MQTT.NewTopicFilter(topic, byte(qos))
	if err != nil {
		return err
	}
	_, err = m.Client.StartSubscription(onMessageReceived, topicFilter)
	if err != nil {
		return err
	}

{'fraction_non_alphanumeric': 0.13371868978805396, 'fraction_numerical': 0.0007707129094412332, 'mean_word_length': 4.4423480083857445, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1926573', 'n_tokens_mistral': 1039, 'n_tokens_neox': 1003, 'n_words': 222}