Unlock 30% off on Manual Testing Annual Plans this Holiday Season. 
Use Code: LCMYEARENDEXCL30
Use Code: LCMYEARENDEXCL30Check icon
Copied to Clipboard!
How to use session method of mqtt Package
Best Venom code snippet using mqtt.session
main.go
Source:main.go 
...10 "github.com/eclipse/paho.mqtt.golang"11 "golang.org/x/sys/unix"12)13var (14 sessionLock sync.Mutex15 carInfo *leaf.VehicleInfo16 carBattery *leaf.BatteryRecords17 carBatteryLock sync.Mutex18 carTemperature *leaf.TemperatureRecords19 carTemperatureLock sync.Mutex20 carLocation *leaf.Location21 carLocationLock sync.Mutex22 carHVAC bool23 carHVACExpiry time.Time24 carHVACLock sync.Mutex25 carHVACReady bool26 mqttBinaryTopic string27 mqttSensorTopic string28 mqttSwitchTopic string29 mqttTrackerTopic string30)31func main() {32 log.SetOutput(os.Stdout)33 err := run()34 if err != nil {35 fmt.Printf("Error: %v\n", err)36 os.Exit(1)37 }38}39func run() error {40 // Setup error handling for background Go routines.41 chError := make(chan error, 0)42 // Setup MQTT connection details.43 mqttOpts := mqtt.NewClientOptions()44 mqttOpts.AddBroker(os.Getenv("MQTT_HOST"))45 mqttOpts.SetClientID("leaf2mqtt")46 mqttOpts.SetUsername(os.Getenv("MQTT_USERNAME"))47 mqttOpts.SetPassword(os.Getenv("MQTT_PASSWORD"))48 mqttOpts.SetAutoReconnect(true)49 mqttOpts.SetCleanSession(false)50 mqttSession := mqtt.NewClient(mqttOpts)51 if token := mqttSession.Connect(); token.Wait() && token.Error() != nil {52 return token.Error()53 }54 // Setup Nissan connection details.55 leafSession := &leaf.Session{56 Username: os.Getenv("LEAF_USERNAME"),57 Password: os.Getenv("LEAF_PASSWORD"),58 Country: os.Getenv("LEAF_COUNTRY"),59 }60 // Login.61 info, _, _, err := leafSession.Login()62 if err != nil {63 return err64 }65 carInfo = info66 mqttBinaryTopic = fmt.Sprintf("homeassistant/binary_sensor/leaf2mqtt_%s", carInfo.VIN)67 mqttTrackerTopic = fmt.Sprintf("homeassistant/device_tracker/leaf2mqtt_%s", carInfo.VIN)68 mqttSensorTopic = fmt.Sprintf("homeassistant/sensor/leaf2mqtt_%s", carInfo.VIN)69 mqttSwitchTopic = fmt.Sprintf("homeassistant/switch/leaf2mqtt_%s", carInfo.VIN)70 // Create the MQTT location topics.71 data := fmt.Sprintf(`{72 "name": "%s (location)",73 "state_topic": "%s/location/state",74 "payload_home": "home",75 "payload_not_home": "not_home",76 "json_attributes_topic": "%s/location/attributes",77 "icon": "mdi:car",78 "source_type": "gps"79}`, carInfo.VIN, mqttTrackerTopic, mqttTrackerTopic)80 if token := mqttSession.Publish(fmt.Sprintf("%s/location/config", mqttTrackerTopic), 0, true, data); token.Wait() && token.Error() != nil {81 return token.Error()82 }83 if token := mqttSession.Publish(fmt.Sprintf("%s/location/state", mqttTrackerTopic), 0, true, "not_home"); token.Wait() && token.Error() != nil {84 return token.Error()85 }86 // Create the MQTT battery topics.87 data = fmt.Sprintf(`{88 "name": "%s (charge)",89 "state_topic": "%s/charge/state",90 "device_class": "battery",91 "unit_of_measurement": "%%"92}`, carInfo.VIN, mqttSensorTopic)93 if token := mqttSession.Publish(fmt.Sprintf("%s/charge/config", mqttSensorTopic), 0, true, data); token.Wait() && token.Error() != nil {94 return token.Error()95 }96 data = fmt.Sprintf(`{97 "name": "%s (plugged)",98 "state_topic": "%s/plugged/state",99 "payload_on": "on",100 "payload_off": "off",101 "device_class": "plug"102}`, carInfo.VIN, mqttBinaryTopic)103 if token := mqttSession.Publish(fmt.Sprintf("%s/plugged/config", mqttBinaryTopic), 0, true, data); token.Wait() && token.Error() != nil {104 return token.Error()105 }106 data = fmt.Sprintf(`{107 "name": "%s (charging)",108 "state_topic": "%s/charging/state",109 "payload_on": "on",110 "payload_off": "off",111 "device_class": "battery_charging"112}`, carInfo.VIN, mqttBinaryTopic)113 if token := mqttSession.Publish(fmt.Sprintf("%s/charging/config", mqttBinaryTopic), 0, true, data); token.Wait() && token.Error() != nil {114 return token.Error()115 }116 // Create the MQTT range topics117 data = fmt.Sprintf(`{118 "name": "%s (range with AC)",119 "state_topic": "%s/range_ac/state",120 "unit_of_measurement": "km"121}`, carInfo.VIN, mqttSensorTopic)122 if token := mqttSession.Publish(fmt.Sprintf("%s/range_ac/config", mqttSensorTopic), 0, true, data); token.Wait() && token.Error() != nil {123 return token.Error()124 }125 data = fmt.Sprintf(`{126 "name": "%s (range without AC)",127 "state_topic": "%s/range_noac/state",128 "unit_of_measurement": "km"129}`, carInfo.VIN, mqttSensorTopic)130 if token := mqttSession.Publish(fmt.Sprintf("%s/range_noac/config", mqttSensorTopic), 0, true, data); token.Wait() && token.Error() != nil {131 return token.Error()132 }133 // Create the MQTT temperature topics.134 data = fmt.Sprintf(`{135 "name": "%s (temperature)",136 "state_topic": "%s/temperature/state",137 "device_class": "temperature",138 "unit_of_measurement": "C"139}`, carInfo.VIN, mqttSensorTopic)140 if token := mqttSession.Publish(fmt.Sprintf("%s/temperature/config", mqttSensorTopic), 0, true, data); token.Wait() && token.Error() != nil {141 return token.Error()142 }143 // Create the MQTT climate topics.144 data = fmt.Sprintf(`{145 "name": "%s (climate)",146 "state_topic": "%s/climate/state",147 "command_topic": "%s/climate/set",148 "payload_on": "on",149 "payload_off": "off",150 "retain": true,151 "state_on": "on",152 "state_off": "off"153}`, carInfo.VIN, mqttSwitchTopic, mqttSwitchTopic)154 if token := mqttSession.Publish(fmt.Sprintf("%s/climate/config", mqttSwitchTopic), 0, true, data); token.Wait() && token.Error() != nil {155 return token.Error()156 }157 if token := mqttSession.Publish(fmt.Sprintf("%s/climate/state", mqttSwitchTopic), 0, true, "off"); token.Wait() && token.Error() != nil {158 return token.Error()159 }160 callback := func(client mqtt.Client, msg mqtt.Message) {161 mqttClimateSet(leafSession, client, msg, chError)162 }163 if token := mqttSession.Subscribe(fmt.Sprintf("%s/climate/set", mqttSwitchTopic), 0, callback); token.Wait() && token.Error() != nil {164 return token.Error()165 }166 // Start the goroutines167 go updateBattery(leafSession, mqttSession, chError)168 go updateLocation(leafSession, mqttSession, chError)169 go printStatus(leafSession, mqttSession, chError)170 go signalHandling(leafSession, chError)171 return <-chError172}173func mqttClimateSet(session *leaf.Session, mqttSession mqtt.Client, msg mqtt.Message, chError chan error) {174 value := msg.Payload()175 if !carHVACReady {176 carHVACReady = true177 log.Printf("[climate] Skipping initial request")178 return179 }180 if string(value) == "on" {181 err := turnClimateOn(session)182 if err == nil {183 if token := mqttSession.Publish(fmt.Sprintf("%s/climate/state", mqttSwitchTopic), 0, true, "on"); token.Wait() && token.Error() != nil {184 log.Printf("[climate] Failed to update MQTT: %v", token.Error())185 return186 }187 }188 } else if string(value) == "off" {189 err := turnClimateOff(session)190 if err == nil {191 if token := mqttSession.Publish(fmt.Sprintf("%s/climate/state", mqttSwitchTopic), 0, true, "off"); token.Wait() && token.Error() != nil {192 log.Printf("[climate] Failed to update MQTT: %v", token.Error())193 return194 }195 }196 } else {197 log.Printf("[climate] Invalid climate state: %v", value)198 }199}200func turnClimateOn(session *leaf.Session) error {201 log.Printf("[climate] Turning HVAC on")202 carHVACLock.Lock()203 defer carHVACLock.Unlock()204 if carHVAC {205 // Reset by attempting to turn off first.206 sessionLock.Lock()207 session.ClimateOff()208 sessionLock.Unlock()209 }210 sessionLock.Lock()211 err := session.ClimateOn()212 sessionLock.Unlock()213 if err != nil {214 log.Printf("[climate] Failed to turn on: %v", err)215 return err216 }217 carHVAC = true218 carHVACExpiry = time.Now().Add(15*time.Minute)219 log.Printf("[climate] Turned HVAC on")220 return nil221}222func turnClimateOff(session *leaf.Session) error {223 log.Printf("[climate] Turning HVAC off")224 carHVACLock.Lock()225 defer carHVACLock.Unlock()226 sessionLock.Lock()227 err := session.ClimateOff()228 sessionLock.Unlock()229 if err != nil {230 log.Printf("[climate] Failed to turn off: %v", err)231 return err232 }233 carHVAC = false234 log.Printf("[climate] Turned HVAC off")235 return nil236}237func signalHandling(session *leaf.Session, chError chan error) {238 sigs := make(chan os.Signal, 1)239 signal.Notify(sigs, unix.SIGUSR1, unix.SIGUSR2)240 for {241 sig := <-sigs242 // Turn on the HVAC system.243 if sig == unix.SIGUSR1 {244 turnClimateOn(session)245 }246 // Turn off the HVAC system.247 if sig == unix.SIGUSR2 {248 turnClimateOff(session)249 }250 }251}252func printStatus(session *leaf.Session, mqttSession mqtt.Client, chError chan error) {253 printCurrent := func() {254 carBatteryLock.Lock()255 carLocationLock.Lock()256 carTemperatureLock.Lock()257 carHVACLock.Lock()258 defer carBatteryLock.Unlock()259 defer carLocationLock.Unlock()260 defer carTemperatureLock.Unlock()261 defer carHVACLock.Unlock()262 log.Printf("[info] Start of status update")263 log.Printf("Car information:")264 log.Printf(" Name: %s", carInfo.Nickname)265 log.Printf(" VIN: %s", carInfo.VIN)266 if carBattery != nil {267 hvacState := "off"268 if carHVAC {269 if carBattery.BatteryStatus.BatteryChargingStatus.IsCharging() {270 hvacState = "on"271 } else if carHVACExpiry.After(time.Now()) {272 hvacState = fmt.Sprintf("on (%s remaining)", time.Until(carHVACExpiry))273 }274 if hvacState == "off" {275 carHVAC = false276 if token := mqttSession.Publish(fmt.Sprintf("%s/climate/state", mqttSwitchTopic), 0, true, "off"); token.Wait() && token.Error() != nil {277 log.Printf("[climate] Failed to update MQTT: %v", token.Error())278 }279 }280 }281 log.Printf("Battery:")282 log.Printf(" Last update: %s", carBattery.LastUpdatedDateAndTime)283 log.Printf(" Plugged: %s", carBattery.PluginState)284 log.Printf(" Charging: %s", carBattery.BatteryStatus.BatteryChargingStatus)285 log.Printf(" Capacity: %d%%", carBattery.BatteryStatus.BatteryRemainingAmount)286 log.Printf("Range:")287 log.Printf(" With HVAC: %.0fkm", carBattery.CruisingRangeACOn/1000)288 log.Printf(" Without HVAC: %.0fkm", carBattery.CruisingRangeACOff/1000)289 log.Printf("Climate:")290 log.Printf(" HVAC: %s", hvacState)291 log.Printf(" Inside temperature: %sC", carTemperature.Temperature)292 }293 if carLocation != nil {294 log.Printf("Location:")295 log.Printf(" Last update: %s", carLocation.ReceivedDate)296 log.Printf(" Latitude: %s", carLocation.Latitude)297 log.Printf(" Longitude: %s", carLocation.Longitude)298 }299 log.Printf("[info] End of status update")300 }301 for {302 printCurrent()303 time.Sleep(2*time.Minute)304 }305}306func updateBattery(session *leaf.Session, mqttSession mqtt.Client, chError chan error) {307 for {308 sessionLock.Lock()309 retBattery, retTemperature, err := session.ChargingStatus()310 sessionLock.Unlock()311 if err != nil {312 log.Printf("[battery] Failed, retrying in 5 minutes: %v", err)313 sessionLock.Lock()314 for {315 _, _, _, err := session.Login()316 if err != nil {317 log.Printf("[battery] Failed login, retrying in 15s: %v", err)318 time.Sleep(15*time.Second)319 continue320 }321 break322 }323 sessionLock.Unlock()324 time.Sleep(5*time.Minute)325 continue326 }327 carBatteryLock.Lock()328 carBattery = retBattery329 carBatteryLock.Unlock()330 carTemperatureLock.Lock()331 carTemperature = retTemperature332 carTemperatureLock.Unlock()333 // MQTT charge update.334 if token := mqttSession.Publish(fmt.Sprintf("%s/charge/state", mqttSensorTopic), 0, true, fmt.Sprintf("%d", retBattery.BatteryStatus.BatteryRemainingAmount)); token.Wait() && token.Error() != nil {335 log.Printf("[battery] Failed to update MQTT: %v", token.Error())336 }337 // MQTT range update.338 if token := mqttSession.Publish(fmt.Sprintf("%s/range_ac/state", mqttSensorTopic), 0, true, fmt.Sprintf("%.0f", retBattery.CruisingRangeACOn/1000)); token.Wait() && token.Error() != nil {339 log.Printf("[battery] Failed to update MQTT: %v", token.Error())340 }341 if token := mqttSession.Publish(fmt.Sprintf("%s/range_noac/state", mqttSensorTopic), 0, true, fmt.Sprintf("%.0f", retBattery.CruisingRangeACOff/1000)); token.Wait() && token.Error() != nil {342 log.Printf("[battery] Failed to update MQTT: %v", token.Error())343 }344 // MQTT plugged in update.345 state := "off"346 if retBattery.PluginState == "CONNECTED" {347 state = "on"348 }349 if token := mqttSession.Publish(fmt.Sprintf("%s/plugged/state", mqttBinaryTopic), 0, true, state); token.Wait() && token.Error() != nil {350 log.Printf("[battery] Failed to update MQTT: %v", token.Error())351 }352 // MQTT charging update.353 state = "off"354 if retBattery.BatteryStatus.BatteryChargingStatus == "YES" {355 state = "on"356 }357 if token := mqttSession.Publish(fmt.Sprintf("%s/charging/state", mqttBinaryTopic), 0, true, state); token.Wait() && token.Error() != nil {358 log.Printf("[battery] Failed to update MQTT: %v", token.Error())359 }360 // MQTT temperature update.361 if token := mqttSession.Publish(fmt.Sprintf("%s/temperature/state", mqttSensorTopic), 0, true, retTemperature.Temperature); token.Wait() && token.Error() != nil {362 log.Printf("[battery] Failed to update MQTT: %v", token.Error())363 }364 if retBattery.BatteryStatus.BatteryChargingStatus.IsCharging() {365 log.Printf("[battery] Waiting for 5min (charging)")366 time.Sleep(5*time.Minute)367 } else if carHVAC {368 log.Printf("[battery] Waiting for 5min (heating up)")369 time.Sleep(5*time.Minute)370 } else {371 log.Printf("[battery] Waiting for 15min")372 time.Sleep(15*time.Minute)373 }374 }375}376func updateLocation(session *leaf.Session, mqttSession mqtt.Client, chError chan error) {377 for {378 sessionLock.Lock()379 ret, err := session.LocateVehicle()380 sessionLock.Unlock()381 if err != nil {382 log.Printf("[locate] Failed, retrying in 5 minutes: %v", err)383 sessionLock.Lock()384 for {385 _, _, _, err := session.Login()386 if err != nil {387 log.Printf("[battery] Failed login, retrying in 15s: %v", err)388 time.Sleep(15*time.Second)389 continue390 }391 break392 }393 sessionLock.Unlock()394 time.Sleep(5*time.Minute)395 continue396 }397 if ret.Latitude == "" || ret.Longitude == "" {398 log.Printf("[locate] Failed, retrying in 5 minutes: bad data received")399 sessionLock.Lock()400 for {401 _, _, _, err := session.Login()402 if err != nil {403 log.Printf("[battery] Failed login, retrying in 15s: %v", err)404 time.Sleep(15*time.Second)405 continue406 }407 break408 }409 sessionLock.Unlock()410 time.Sleep(5*time.Minute)411 continue412 }413 // MQTT location update.414 data := fmt.Sprintf(`{415 "latitude": %s,416 "longitude": %s,417 "gps_accuracy": 1.2418}`, ret.Latitude, ret.Longitude)419 if token := mqttSession.Publish(fmt.Sprintf("%s/location/attributes", mqttTrackerTopic), 0, true, data); token.Wait() && token.Error() != nil {420 log.Printf("[locate] Failed to update MQTT: %v", token.Error())421 }422 carLocationLock.Lock()423 moved := carLocation == nil || ret.Longitude != carLocation.Longitude || ret.Latitude != carLocation.Latitude...
pushChannel.go
Source:pushChannel.go
...22 err = p.engineMqtt.Where("channel_type = ?", channelType+p.lastfix).And("service = ?", "mqtt").And("production = ?", "storybox").Find(&pushChannels)23 return24}25func (p *PushChannel) Delete(channelType string) (err error) {26 sessionMqtt := p.engineMqtt.NewSession()27 defer sessionMqtt.Close()28 lists, err := p.List(channelType)29 if err != nil {30 sessionMqtt.Rollback()31 return32 }33 for _, one := range lists {34 _, err = sessionMqtt.Id(one.General.Id).Delete(&one)35 if err != nil {36 sessionMqtt.Rollback()37 return38 }39 }40 err = sessionMqtt.Commit()41 return42}43func (p *PushChannel) Add(autobuildId int, paramsJson string) (err error) {44 sessionTK := p.engineTK.NewSession()45 sessionMqtt := p.engineMqtt.NewSession()46 defer sessionTK.Close()47 defer sessionMqtt.Close()48 err = sessionTK.Begin()49 if err != nil {50 return51 }52 err = sessionMqtt.Begin()53 if err != nil {54 return55 }56 autobuild := models.AutoBuild{}57 _, err = sessionTK.Where("id = ?", autobuildId).Get(&autobuild)58 if err != nil {59 sessionTK.Rollback()60 sessionMqtt.Rollback()61 return err62 }63 channelType := autobuild.AppId + p.lastfix64 // å¦æmqttç¶æ为trueï¼åæ£æ¥æ¯å¦å·²æå
¥65 if autobuild.Mqtt != 0 {66 tmpMqtt := models.PushChannel{}67 _, err = sessionMqtt.Where("channel_type = ?", channelType).And("service = ?", "mqtt").And("production = ?", "storybox").Get(&tmpMqtt)68 if err != nil {69 sessionTK.Rollback()70 sessionMqtt.Rollback()71 return err72 }73 if tmpMqtt.Id > 0 {74 return errors.New("mqtt already set")75 }76 }77 mqtt := models.PushChannel{78 ChannelType: channelType,79 Production: "storybox",80 Service: "mqtt",81 Params: paramsJson,82 Status: 1,83 }84 mqtt.General.UpdatedAt = int(time.Now().Unix())85 mqtt.General.CreatedAt = int(time.Now().Unix())86 _, err = sessionMqtt.Insert(&mqtt)87 if err != nil {88 sessionTK.Rollback()89 sessionMqtt.Rollback()90 return err91 }92 autobuild.Mqtt = 193 updateAutobuild := models.AutoBuild{94 Mqtt: 1,95 }96 _, err = sessionTK.Where("id = ?", autobuild.Id).Update(&updateAutobuild)97 if err != nil {98 sessionTK.Rollback()99 sessionMqtt.Rollback()100 return err101 }102 err = sessionTK.Commit()103 if err != nil {104 sessionTK.Rollback()105 sessionMqtt.Rollback()106 return err107 }108 err = sessionMqtt.Commit()109 if err != nil {110 sessionTK.Rollback()111 sessionMqtt.Rollback()112 return err113 }114 return nil115}...
config.go
Source:config.go
...14 // if this flag set true, sever will store the message and can be delivered to future subscribers15 // default false16 // Note: Can not use "omitempty" option, It will affect the output of the default configuration file17 MqttRetain bool `json:"mqttRetain"`18 // MqttSessionQueueSize indicates the size of how many sessions will be handled.19 // default 10020 MqttSessionQueueSize int `json:"mqttSessionQueueSize,omitempty"`21 // MqttServerInternal indicates internal mqtt broker url22 // default tcp://127.0.0.1:188423 MqttServerInternal string `json:"mqttServerInternal,omitempty"`24 // MqttServerExternal indicates external mqtt broker url25 // default tcp://127.0.0.1:188326 MqttServerExternal string `json:"mqttServerExternal,omitempty"`27 // MqttMode indicates which broker type will be choose28 // 0: internal mqtt broker enable only. 1: internal and external mqtt broker enable. 2: external mqtt broker enable only29 // +Required30 // default: 031 MqttMode int `json:"mqttMode"`32}33func GetEventBusConfig() *EventBusConfig {34 eBConfig := &EventBusConfig{}35 qos, err := config.CONFIG.GetValue("eventbus.mqtt.qos").ToInt()36 if err != nil {37 klog.Infof("eventbus.mqtt.qos is empty")38 qos = 239 }40 eBConfig.MqttQOS = qos41 retain, err := config.CONFIG.GetValue("eventbus.mqtt.retain").ToBool()42 if err != nil {43 klog.Infof("eventbus.mqtt.retain is empty")44 retain = false45 }46 eBConfig.MqttRetain = retain47 sessionQueueSize, err := config.CONFIG.GetValue("eventbus.mqtt.session-queue-size").ToInt()48 if err != nil {49 klog.Infof("eventbus.mqtt.session-queue-size is empty")50 sessionQueueSize = 10051 }52 eBConfig.MqttSessionQueueSize = sessionQueueSize53 MqttServerInternal, err := config.CONFIG.GetValue("eventbus.mqtt.mqttServerInternal").ToString()54 if err != nil {55 klog.Infof("eventbus.mqtt.mqttServerInternal is empty")56 MqttServerInternal = "tcp://127.0.0.1:1884"57 }58 eBConfig.MqttServerInternal = MqttServerInternal59 MqttServerExternal, err := config.CONFIG.GetValue("eventbus.mqtt.mqttServerExternal").ToString()60 if err != nil {61 klog.Infof("eventbus.mqtt.mqttServerExternal is empty")62 MqttServerExternal = "tcp://127.0.0.1:1883"63 }64 eBConfig.MqttServerExternal = MqttServerExternal65 mode, err := config.CONFIG.GetValue("eventbus.mqtt.mode").ToInt()66 if err != nil {...
session
Using AI Code Generation
1import (2func main() {3 opts.SetKeepAlive(2 * time.Second)4 opts.SetPingTimeout(1 * time.Second)5 c := mqtt.NewClient(opts)6 if token := c.Connect(); token.Wait() && token.Error() != nil {7 panic(token.Error())8 }9 fmt.Println("Connected to broker")10 if token := c.Subscribe("hello/world/1", 0, func(client mqtt.Client, msg mqtt.Message) {11 fmt.Printf("Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())12 }); token.Wait() && token.Error() != nil {13 log.Println(token.Error())14 os.Exit(1)15 }16 fmt.Println("Subscribed to topic")17 time.Sleep(3 * time.Second)18 if token := c.Publish("hello/world/1", 0, false, "Hello World"); token.Wait() && token.Error() != nil {19 log.Println(token.Error())20 os.Exit(1)21 }22 fmt.Println("Published message to topic")23 time.Sleep(3 * time.Second)24}25import (26func main() {27 opts.SetKeepAlive(2 * time.Second)28 opts.SetPingTimeout(1 * time.Second)29 c := mqtt.NewClient(opts)30 if token := c.Connect(); token.Wait() && token.Error() != nil {31 panic(token.Error())32 }33 fmt.Println("Connected to broker")34 if token := c.Subscribe("hello/world/2", 0, func(client mqtt.Client, msg mqtt.Message) {35 fmt.Printf("Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())36 }); token.Wait() && token.Error() != nil {37 log.Println(token.Error())38 os.Exit(1)39 }40 fmt.Println("Subscribed to topic")41 time.Sleep(3 * time.Second)42 if token := c.Publish("hello/world/2", 0, false,
session
Using AI Code Generation
1import (2func main() {3 opts := mqtt.NewClientOptions()4 opts.SetClientID("go-simple")5 opts.SetCleanSession(false)6 opts.SetUsername("test")7 opts.SetPassword("test")8 opts.SetKeepAlive(2 * time.Second)9 opts.SetPingTimeout(1 * time.Second)10 opts.SetWriteTimeout(1 * time.Second)11 opts.SetMaxReconnectInterval(1 * time.Second)12 opts.SetAutoReconnect(false)13 client := mqtt.NewClient(opts)14 if token := client.Connect(); token.Wait() && token.Error() != nil {15 panic(token.Error())16 }17 fmt.Println("Connected")18 if token := client.Subscribe("test", 0, func(client mqtt.Client, msg mqtt.Message) {19 fmt.Printf("Received message on topic: %s Message: %s\n", msg.Topic(), msg.Payload())20 }); token.Wait() && token.Error() != nil {21 fmt.Println(token.Error())22 os.Exit(1)23 }24 for i := 0; i < 5; i++ {25 text := fmt.Sprintf("this is msg #%d!", i)26 token := client.Publish("test", 0, false, text)27 token.Wait()28 }29 time.Sleep(6 * time.Second)30 if token := client.Unsubscribe("test"); token.Wait() && token.Error() != nil {31 fmt.Println(token.Error())32 os.Exit(1)33 }34 client.Disconnect(250)35}36import (37func main() {38 opts := mqtt.NewClientOptions()39 opts.SetClientID("go-simple")40 opts.SetCleanSession(false)41 opts.SetUsername("test")42 opts.SetPassword("test")43 opts.SetKeepAlive(2 * time.Second)44 opts.SetPingTimeout(1 * time.Second)45 opts.SetWriteTimeout(1 * time.Second)46 opts.SetMaxReconnectInterval(1 *
session
Using AI Code Generation
1import (2func main() {3 opts.SetClientID("go-simple")4 opts.SetCleanSession(false)5 opts.SetKeepAlive(2 * time.Second)6 opts.SetPingTimeout(1 * time.Second)7 c := mqtt.NewClient(opts)8 if token := c.Connect(); token.Wait() && token.Error() != nil {9 panic(token.Error())10 }11 if token := c.Subscribe("go-simple", 0, nil); token.Wait() && token.Error() != nil {12 fmt.Println(token.Error())13 os.Exit(1)14 }15 time.Sleep(3 * time.Second)16 if token := c.Unsubscribe("go-simple"); token.Wait() && token.Error() != nil {17 fmt.Println(token.Error())18 os.Exit(1)19 }20 c.Disconnect(250)21}22import (23func main() {24 opts.SetClientID("go-simple")25 opts.SetCleanSession(true)26 opts.SetKeepAlive(2 * time.Second)27 opts.SetPingTimeout(1 * time.Second)28 c := mqtt.NewClient(opts)29 if token := c.Connect(); token.Wait() && token.Error() != nil {30 panic(token.Error())31 }32 if token := c.Subscribe("go-simple", 0, nil); token.Wait() && token.Error() != nil {33 fmt.Println(token.Error())34 os.Exit(1)35 }36 time.Sleep(3 * time.Second)37 if token := c.Unsubscribe("go-simple"); token.Wait() && token.Error() != nil {38 fmt.Println(token.Error())39 os.Exit(1)40 }41 c.Disconnect(250)42}
session
Using AI Code Generation
1import (2func main() {3 opts.SetClientID("go-simple")4 opts.SetUsername("admin")5 opts.SetPassword("password")6 c := mqtt.NewClient(opts)7 if token := c.Connect(); token.Wait() && token.Error() != nil {8 panic(token.Error())9 }10 if token := c.Subscribe("go-mqtt/sample", 0, nil); token.Wait() && token.Error() != nil {11 fmt.Println(token.Error())12 os.Exit(1)13 }14 for i := 0; i < 5; i++ {15 text := fmt.Sprintf("this is msg #%d!", i)16 token := c.Publish("go-mqtt/sample", 0, false, text)17 token.Wait()18 }19 if token := c.Unsubscribe("go-mqtt/sample"); token.Wait() && token.Error() != nil {20 fmt.Println(token.Error())21 os.Exit(1)22 }23 c.Disconnect(250)24}
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run Venom automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Most used method in
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
