MQTT

[oedemis]

Have you plans to implement also a requester for MQTT i have no experience with go but i have tested a some mqtt here is a mqtt_requester but im not sure what im doing :)

package requester

import ( "fmt" "os" "github.com/satori/go.uuid" "github.com/tylertreat/bench" MQTT "git.eclipse.org/gitroot/paho/org.eclipse.paho.mqtt.golang.git" )

// MQTTRequesterFactory implements RequesterFactory by creating a Requester // which publishes messages to an AMQP exchange and waits to consume them. type MQTTRequesterFactory struct { URL string TOPICs []string }

// GetRequester returns a new Requester, called for each Benchmark connection. func (r *MQTTRequesterFactory) GetRequester(num uint64) bench.Requester { return &mqttRequester{ url: r.URL, topics: r.TOPICs, } }

// amqpRequester implements Requester by publishing a message to an MQTT // exhcnage and waiting to consume it????. type mqttRequester struct { url string topics []string client *MQTT.Client }

var f MQTT.MessageHandler = func(client *MQTT.Client, msg MQTT.Message) { fmt.Printf("TOPIC: %s\n", msg.Topic()) fmt.Printf("MSG: %s\n", msg.Payload()) }

// Setup prepares the Requester for benchmarking. func (r *mqttRequester) Setup() error { u1 := uuid.NewV4()

opts := MQTT.NewClientOptions().AddBroker(r.url).SetClientID(u1.String())
//opts.SetDefaultPublishHandler(f)

c := MQTT.NewClient(opts)
r.client = c
if token := c.Connect(); token.Wait() && token.Error() != nil {
    panic(token.Error())
}

for i:= 0; i < len(r.topics); i++ {
    if token := c.Subscribe(r.topics[i], 0, nil); token.Wait() && token.Error() != nil {
        fmt.Println(token.Error())
        os.Exit(1)
    }       
}

return nil

}

// Request performs a synchronous request to the system under test. func (r *mqttRequester) Request() error {

/*if token := r.client.Subscribe("/go-mqtt/sample", 0, nil); token.Wait() && token.Error() != nil {
    fmt.Println(token.Error())
    os.Exit(1)
}*/

text := fmt.Sprintf("this is msg #%d!", 1)
//token := r.client.Publish("/go-mqtt/sample", 0, false, text)
//r.client.Publish("/go-mqtt/sample", 0, false, text)

for i:= 0; i < len(r.topics); i++ {
    r.client.Publish(r.topics[i], 0, false, text)
}

return nil

}

// Teardown is called upon benchmark completion. func (r *mqttRequester) Teardown() error { r.client.Disconnect(250) return nil }

and main.go

package main

import ( "fmt" "time"

"github.com/tylertreat/bench"
"./requester"

)

//"github.com/tylertreat/bench/requester" func main() { /r := &requester.WebRequesterFactory{ URL: "http://localhost:8080/", }/

r := &requester.MQTTRequesterFactory{
    URL:         "tcp://127.0.0.1:1883",
    TOPICs:      []string{"topic1", "topic2"},
}

benchmark := bench.NewBenchmark(r, 10000, 1, 30*time.Second)
summary, err := benchmark.Run()
if err != nil {
    panic(err)
}

fmt.Println(summary)
summary.GenerateLatencyDistribution(bench.Logarithmic, "mqtt.txt")

}