How to use Run method of mqtt Package

Best Venom code snippet using mqtt.Run

devemulator.go

Source:devemulator.go

...109// transmits. The simulation will call TX randomly110type AdvancedDevice interface {111	// funtion that will be called once to initialize each new device112	Init(devid string)113	Run(devrx <-chan []byte, devtx chan<- []byte)114}115func runSimpleDevice(client MQTT.Client, devCfg deviceConfig) {116	devCfg.devInstance.(SimpleDevice).Init(devCfg.mqttNode)117	// Sleep some random time to emulate random turn on time118	time.Sleep(time.Duration(rand.Intn(device_max_sleep)) * time.Millisecond)119	client.Subscribe(devCfg.rxTopic(), byte(mqttQos), func(c MQTT.Client, m MQTT.Message) {120		bytes, err := base64.StdEncoding.DecodeString(string(m.Payload()))121		if err != nil {122			log.Println("Error base64 decoding message for " + m.Topic())123			return124		}125		resp := devCfg.devInstance.(SimpleDevice).RX(bytes)126		if resp != nil {127			resp64 := base64.StdEncoding.EncodeToString(resp)128			client.Publish(devCfg.txTopic(), byte(mqttQos), false, resp64)129		}130	})131	for {132		tx := devCfg.devInstance.(SimpleDevice).TX()133		if tx != nil {134			tx64 := base64.StdEncoding.EncodeToString(tx)135			client.Publish(devCfg.txTopic(), byte(mqttQos), false, tx64)136		}137		time.Sleep(time.Duration(devCfg.txPeriodMs) * time.Millisecond)138	}139}140func runAdvancedDevice(client MQTT.Client, devCfg deviceConfig) {141	log.Fatalln("AdvancedDevice functionality is not implemented")142	// devrxdata := make(chan []byte)143	// devtxdata := make(chan []byte)144	// quit := make(chan bool)145	// go func() {146	// 	for {147	// 		select {148	// 		// case tx <- devtxdata:149	// 		case <-quit:150	// 			return151	// 		}152	// 	}153	// }()154	// devCfg.devInstance.(AdvancedDevice).Init(devCfg.mqttNode)155	// devCfg.devInstance.(AdvancedDevice).Run(devrxdata, devtxdata)156}157func main() {158	/* Parse Arguments */159	flag.Parse()160	fmt.Println("# Starting up")161	/* Setup basic MQTT connection */162	opts := MQTT.NewClientOptions().AddBroker(mqttBroker)163	opts.SetClientID(genclientid())164	opts.SetUsername(mqttUser)165	opts.SetPassword(mqttPass)166	/* Create and start a client using the above ClientOptions */167	c := MQTT.NewClient(opts)168	if token := c.Connect(); token.Wait() && token.Error() != nil {169		panic(token.Error())170	}171	defer c.Disconnect(250)172	/* Start a go routine for each virtual device */173	for _, dev := range devices {174		fmt.Print("# Running device " + dev.mqttNode + " type ")175		fmt.Println(reflect.TypeOf(dev.devInstance))176		if _, ok := dev.devInstance.(SimpleDevice); ok {177			go runSimpleDevice(c, dev)178		} else if _, ok := dev.devInstance.(AdvancedDevice); ok {179			go runAdvancedDevice(c, dev)180		} else {181			log.Fatal("Device " + dev.mqttNode + " is of invalid type")182		}183	}184	/* Setup Stuff to Manage a Safe Exit */185	timeout := make(chan bool)186	// signal for SIGINT187	signals := make(chan os.Signal)188	if runtime > 0 {189		fmt.Println("# Running for ", runtime, " second(s)")190		// run for argument seconds if specified191		go func() {192			<-time.After(time.Duration(runtime) * time.Second)193			timeout <- true194		}()195	}196	signal.Notify(signals, os.Interrupt)197	select {198	case <-timeout:199		fmt.Println("# Reached end of timed life")200	case sig := <-signals:201		fmt.Println("# Caught signal ", sig)202	}203	fmt.Println("# Ending")...

mac2mqtt.go

Source:mac2mqtt.go

1package main2import (3	"fmt"4	"gopkg.in/yaml.v2"5	"io/ioutil"6	"log"7	"os"8	"os/exec"9	"regexp"10	"strconv"11	"strings"12	"sync"13	"time"14	mqtt "github.com/eclipse/paho.mqtt.golang"15)16var hostname string17type config struct {18	Ip       string `yaml:"mqtt_ip"`19	Port     string `yaml:"mqtt_port"`20	User     string `yaml:"mqtt_user"`21	Password string `yaml:"mqtt_password"`22}23func (c *config) getConfig() *config {24	configContent, err := ioutil.ReadFile("mac2mqtt.yaml")25	if err != nil {26		log.Fatal(err)27	}28	err = yaml.Unmarshal(configContent, c)29	if err != nil {30		log.Fatal(err)31	}32	if c.Ip == "" {33		log.Fatal("Must specify mqtt_ip in mac2mqtt.yaml")34	}35	if c.Port == "" {36		log.Fatal("Must specify mqtt_port in mac2mqtt.yaml")37	}38	if c.User == "" {39		log.Fatal("Must specify mqtt_user in mac2mqtt.yaml")40	}41	if c.Password == "" {42		log.Fatal("Must specify mqtt_password in mac2mqtt.yaml")43	}44	return c45}46func getHostname() string {47	hostname, err := os.Hostname()48	if err != nil {49		log.Fatal(err)50	}51	// "name.local" => "name"52	firstPart := strings.Split(hostname, ".")[0]53	// remove all symbols, but [a-zA-Z0-9_-]54	reg, err := regexp.Compile("[^a-zA-Z0-9_-]+")55	if err != nil {56		log.Fatal(err)57	}58	firstPart = reg.ReplaceAllString(firstPart, "")59	return firstPart60}61func getCommandOutput(name string, arg ...string) string {62	cmd := exec.Command(name, arg...)63	stdout, err := cmd.Output()64	if err != nil {65		log.Fatal(err)66	}67	stdoutStr := string(stdout)68	stdoutStr = strings.TrimSuffix(stdoutStr, "\n")69	return stdoutStr70}71func getMuteStatus() bool {72	output := getCommandOutput("/usr/bin/osascript", "-e", "output muted of (get volume settings)")73	b, err := strconv.ParseBool(output)74	if err != nil {75		log.Fatal(err)76	}77	return b78}79func getCurrentVolume() int {80	output := getCommandOutput("/usr/bin/osascript", "-e", "output volume of (get volume settings)")81	i, err := strconv.Atoi(output)82	if err != nil {83		log.Fatal(err)84	}85	return i86}87func runCommand(name string, arg ...string) {88	cmd := exec.Command(name, arg...)89	_, err := cmd.Output()90	if err != nil {91		log.Fatal(err)92	}93}94// from 0 to 10095func setVolume(i int) {96	runCommand("/usr/bin/osascript", "-e", "set volume output volume "+strconv.Itoa(i))97}98// true - turn mute on99// false - turn mute off100func setMute(b bool) {101	runCommand("/usr/bin/osascript", "-e", "set volume output muted "+strconv.FormatBool(b))102}103func commandSleep() {104	runCommand("pmset", "sleepnow")105}106func commandDisplaySleep() {107	runCommand("pmset", "displaysleepnow")108}109func commandShutdown() {110	if os.Getuid() == 0 {111		// if the program is run by root user we are doing the most powerfull shutdown - that always shuts down the computer112		runCommand("shutdown", "-h", "now")113	} else {114		// if the program is run by ordinary user we are trying to shutdown, but it may fail if the other user is logged in115		runCommand("/usr/bin/osascript", "-e", "tell app \"System Events\" to shut down")116	}117}118var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {119	log.Printf("Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())120}121var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {122	log.Println("Connected to MQTT")123	token := client.Publish(getTopicPrefix()+"/status/alive", 0, true, "true")124	token.Wait()125	log.Println("Sending 'true' to topic: " + getTopicPrefix() + "/status/alive")126	listen(client, getTopicPrefix()+"/command/#")127}128var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {129	log.Printf("Disconnected from MQTT: %v", err)130}131func getMQTTClient(ip, port, user, password string) mqtt.Client {132	opts := mqtt.NewClientOptions()133	opts.AddBroker(fmt.Sprintf("tcp://%s:%s", ip, port))134	opts.SetUsername(user)135	opts.SetPassword(password)136	opts.OnConnect = connectHandler137	opts.OnConnectionLost = connectLostHandler138	opts.SetWill(getTopicPrefix()+"/status/alive", "false", 0, true)139	client := mqtt.NewClient(opts)140	if token := client.Connect(); token.Wait() && token.Error() != nil {141		panic(token.Error())142	}143	return client144}145func getTopicPrefix() string {146	return "mac2mqtt/" + hostname147}148func listen(client mqtt.Client, topic string) {149	token := client.Subscribe(topic, 0, func(client mqtt.Client, msg mqtt.Message) {150		if msg.Topic() == getTopicPrefix()+"/command/volume" {151			i, err := strconv.Atoi(string(msg.Payload()))152			if err == nil && i >= 0 && i <= 100 {153				setVolume(i)154				updateVolume(client)155				updateMute(client)156			} else {157				log.Println("Incorrect value")158			}159		}160		if msg.Topic() == getTopicPrefix()+"/command/mute" {161			b, err := strconv.ParseBool(string(msg.Payload()))162			if err == nil {163				setMute(b)164				updateVolume(client)165				updateMute(client)166			} else {167				log.Println("Incorrect value")168			}169		}170		if msg.Topic() == getTopicPrefix()+"/command/sleep" {171			if string(msg.Payload()) == "sleep" {172				commandSleep()173			}174		}175		if msg.Topic() == getTopicPrefix()+"/command/displaysleep" {176			if string(msg.Payload()) == "displaysleep" {177				commandDisplaySleep()178			}179		}180		if msg.Topic() == getTopicPrefix()+"/command/shutdown" {181			if string(msg.Payload()) == "shutdown" {182				commandShutdown()183			}184		}185	})186	token.Wait()187	if token.Error() != nil {188		log.Printf("Token error: %s\n", token.Error())189	}190}191func updateVolume(client mqtt.Client) {192	token := client.Publish(getTopicPrefix()+"/status/volume", 0, false, strconv.Itoa(getCurrentVolume()))193	token.Wait()194}195func updateMute(client mqtt.Client) {196	token := client.Publish(getTopicPrefix()+"/status/mute", 0, false, strconv.FormatBool(getMuteStatus()))197	token.Wait()198}199func getBatteryChargePercent() string {200	output := getCommandOutput("/usr/bin/pmset", "-g", "batt")201	// $ /usr/bin/pmset -g batt202	// Now drawing from 'Battery Power'203	//  -InternalBattery-0 (id=4653155)        100%; discharging; 20:00 remaining present: true204	r := regexp.MustCompile(`(\d+)%`)205	percent := r.FindStringSubmatch(output)[1]206	return percent207}208func updateBattery(client mqtt.Client) {209	token := client.Publish(getTopicPrefix()+"/status/battery", 0, false, getBatteryChargePercent())210	token.Wait()211}212func main() {213	log.Println("Started")214	var c config215	c.getConfig()216	var wg sync.WaitGroup217	hostname = getHostname()218	mqttClient := getMQTTClient(c.Ip, c.Port, c.User, c.Password)219	volumeTicker := time.NewTicker(2 * time.Second)220	batteryTicker := time.NewTicker(60 * time.Second)221	wg.Add(1)222	go func() {223		for {224			select {225			case _ = <-volumeTicker.C:226				updateVolume(mqttClient)227				updateMute(mqttClient)228			case _ = <-batteryTicker.C:229				updateBattery(mqttClient)230			}231		}232	}()233	wg.Wait()234}...

main.go

Source:main.go

...9	"github.com/markbradley27/henrietta/src/daemons/van_state/edge_handler"10	"github.com/warthog618/gpiod"11)12const (13	engineRunPinDebouncePeriod = 1 * time.Second14	engineRunPinNumber         = 1715	gpioChipDevName            = "gpiochip0"16)17func main() {18	logFilePath := flag.String("log_file", "", "Log file.")19	mqttHost := flag.String("mqtt_host", consts.RpiIp, "MQTT broker host.")20	mqttPort := flag.Int("mqtt_port", consts.MosquittoPort, "MQTT broker port.")21	mqttClientID := flag.String("mqtt_client_id", consts.ClientIdVanState, "MQTT client id.")22	flag.Parse()23	log_util.SetupLogFile(*logFilePath)24	log.Printf("Connecting to MQTT server at %s:%d as %#v...", *mqttHost, *mqttPort, *mqttClientID)25	mqttClient, err := mqtt_util.ConnectMQTTClient(*mqttHost, *mqttPort, *mqttClientID)26	if err != nil {27		log.Fatalf("Connecting to MQTT: %v", err)28	}29	defer mqttClient.Disconnect(0)30	log.Print("Connected.")31	edgeHandler := edge_handler.NewEdgeHandler(mqttClient)32	log.Printf("Subscribing to engine run pin %d edges...", engineRunPinNumber)33	engineRunPin, err := gpiod.RequestLine(gpioChipDevName, engineRunPinNumber, gpiod.WithEventHandler(edgeHandler.HandleEngineRunEdge), gpiod.WithBothEdges, gpiod.WithDebounce(engineRunPinDebouncePeriod))34	if err != nil {35		log.Fatalf("Requesting engine run pin: %v", err)36	}37	defer engineRunPin.Close()38	log.Print("Subscribed.")39	select {}40}...

Run

Using AI Code Generation

1import (2func main() {3	opts.SetClientID("go-simple")4	opts.SetUsername("test")5	opts.SetPassword("test")6	opts.SetCleanSession(true)7	opts.SetDefaultPublishHandler(func(client mqtt.Client, msg mqtt.Message) {8		fmt.Printf("TOPIC: %s9", msg.Topic())10		fmt.Printf("MSG: %s11", msg.Payload())12	})13	c := mqtt.NewClient(opts)14	if token := c.Connect(); token.Wait() && token.Error() != nil {15		panic(token.Error())16	}17	if token := c.Subscribe("go-simple", 0, nil); token.Wait() && token.Error() != nil {18		fmt.Println(token.Error())19		os.Exit(1)20	}21	for {22		time.Sleep(1 * time.Second)23	}24}25import (26func main() {27	opts.SetClientID("go-simple")28	opts.SetUsername("test")29	opts.SetPassword("test")30	opts.SetCleanSession(true)31	opts.SetDefaultPublishHandler(func(client mqtt.Client, msg mqtt.Message) {32		fmt.Printf("TOPIC: %s33", msg.Topic())34		fmt.Printf("MSG: %s35", msg.Payload())36	})37	c := mqtt.NewClient(opts)38	if token := c.Connect(); token.Wait() && token.Error() != nil {39		panic(token.Error())40	}41	if token := c.Subscribe("go-simple", 0, nil); token.Wait() && token.Error() != nil {42		fmt.Println(token.Error())43		os.Exit(1)44	}45	for {46		time.Sleep(1 * time.Second)47	}48}49import (50func main() {51	opts.SetClientID("go-simple")52	opts.SetUsername("test")53	opts.SetPassword("test")54	opts.SetCleanSession(true

Run

Using AI Code Generation

1import (2func main() {3	opts.SetClientID("go-simple")4	opts.SetUsername("test")5	opts.SetPassword("test")6	opts.SetKeepAlive(2 * time.Second)7	opts.SetPingTimeout(1 * time.Second)8	c := mqtt.NewClient(opts)9	if token := c.Connect(); token.Wait() && token.Error() != nil {10		panic(token.Error())11	}12	if token := c.Subscribe("go-simple", 0, nil); token.Wait() && token.Error() != nil {13		fmt.Println(token.Error())14		os.Exit(1)15	}16	time.Sleep(3 * time.Second)17	if token := c.Unsubscribe("go-simple"); token.Wait() && token.Error() != nil {18		fmt.Println(token.Error())19		os.Exit(1)20	}21	c.Disconnect(250)22}23import (24func main() {25	opts.SetClientID("go-simple")26	opts.SetUsername("test")27	opts.SetPassword("test")28	opts.SetKeepAlive(2 * time.Second)29	opts.SetPingTimeout(1 * time.Second)30	c := mqtt.NewClient(opts)31	if token := c.Connect(); token.Wait() && token.Error() != nil {32		panic(token.Error())33	}34	if token := c.Subscribe("go-simple", 0, nil); token.Wait() && token.Error() != nil {35		fmt.Println(token.Error())36		os.Exit(1)37	}38	time.Sleep(3 * time.Second)39	if token := c.Unsubscribe("go-simple"); token.Wait() && token.Error() != nil {40		fmt.Println(token.Error())41		os.Exit(1)42	}43	c.Disconnect(250)44}45import (46func main() {47	opts.SetClientID("go

Run

Using AI Code Generation

1import (2func main() {3	mqtt.Start()4	time.Sleep(10 * time.Second)5	mqtt.Stop()6}7import (8func main() {9	mqtt.Start()10	time.Sleep(10 * time.Second)11	mqtt.Stop()12}13import (14func main() {15	mqtt.Start()16	time.Sleep(10 * time.Second)17	mqtt.Stop()18}19import (20func main() {21	mqtt.Start()22	time.Sleep(10 * time.Second)23	mqtt.Stop()24}25import (26func main() {27	mqtt.Start()28	time.Sleep(10 * time.Second)29	mqtt.Stop()30}31import (32func main() {33	mqtt.Start()34	time.Sleep(10 * time.Second)35	mqtt.Stop()36}37import (38func main() {39	mqtt.Start()40	time.Sleep(10 * time.Second)41	mqtt.Stop()42}43import (44func main() {45	mqtt.Start()46	time.Sleep(10 * time.Second)47	mqtt.Stop()48}49import (50func main() {51	mqtt.Start()52	time.Sleep(10 * time.Second)53	mqtt.Stop()54}55import (

Run

Using AI Code Generation

1import (2func main() {3	fmt.Println("Hello World!")4	mqtt.Run()5	time.Sleep(time.Second * 5)6}7import (8func main() {9	fmt.Println("Hello World!")10	mqtt.Run()11	time.Sleep(time.Second * 5)12}13import (14func main() {15	fmt.Println("Hello World!")16	mqtt.Run()17	time.Sleep(time.Second * 5)18}19import (20func main() {21	fmt.Println("Hello World!")22	mqtt.Run()23	time.Sleep(time.Second * 5)24}25import (26func main() {27	fmt.Println("Hello World!")28	mqtt.Run()29	time.Sleep(time.Second * 5)30}31import (32func main() {33	fmt.Println("Hello World!")34	mqtt.Run()35	time.Sleep(time.Second * 5)36}37import (

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.