@author liuqing

@commit context cancel 泄露问题修改

cloud/ipolesvr/mqttclient.go

@@ -52,7 +52,9 @@ func NewMqttClient(server, clientId, user, password string, timeout uint, mqttOn
 		return nil
 	}
 	o.mqtt = client
-	err = client.Connect(o.Ctx())
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	err = client.Connect(ctx)
 	return &o
 }
 
@@ -83,7 +85,9 @@ func (o *MqttClient) GetWill() (topic string, payload string) {
 }
 
 func (o *MqttClient) Connect() error {
-	return o.mqtt.Connect(o.Ctx())
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Connect(ctx)
 }
 
 func (o *MqttClient) IsConnected() bool {
@@ -91,13 +95,19 @@ func (o *MqttClient) IsConnected() bool {
 }
 
 func (o *MqttClient) Publish(topic string, payload []byte, qos mqtt.QOS) error {
-	return o.mqtt.Publish(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Publish(ctx, topic, payload, qos)
 }
 func (o *MqttClient) PublishString(topic string, payload string, qos mqtt.QOS) error {
-	return o.mqtt.PublishString(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishString(ctx, topic, payload, qos)
 }
 func (o *MqttClient) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
-	return o.mqtt.PublishJSON(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
 }
 
 func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
@@ -106,7 +116,9 @@ func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
 	if _, ok := o.mapTopics[topic]; !ok {
 		o.mapTopics[topic] = qos
 	}
-	return o.mqtt.Subscribe(o.Ctx(), topic, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Subscribe(ctx, topic, qos)
 }
 
 func (o *MqttClient) Unsubscribe(topic string) error {
@@ -115,14 +127,15 @@ func (o *MqttClient) Unsubscribe(topic string) error {
 	if _, ok := o.mapTopics[topic]; ok {
 		delete(o.mapTopics, topic)
 	}
-	return o.mqtt.Unsubscribe(o.Ctx(), topic)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Unsubscribe(ctx, topic)
 }
 
 func (o *MqttClient) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
 	return o.mqtt.Handle(topic, handler)
 }
 
-func (o *MqttClient) Ctx() context.Context {
-	ctx, _ := context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
-	return ctx
+func (o *MqttClient) Ctx() (context.Context, context.CancelFunc) {
+	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
 }

cloud/mqttforward/mqttclient.go

@@ -52,7 +52,9 @@ func NewMqttClient(server, clientid, user, password string, timeout uint, mqttOn
 		return nil
 	}
 	o.mqtt = client
-	err = client.Connect(o.Ctx())
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	err = client.Connect(ctx)
 	return &o
 }
 
@@ -83,7 +85,9 @@ func (o *MqttClient) GetWill() (topic string, payload string) {
 }
 
 func (o *MqttClient) Connect() error {
-	return o.mqtt.Connect(o.Ctx())
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Connect(ctx)
 }
 
 func (o *MqttClient) IsConnected() bool {
@@ -91,13 +95,19 @@ func (o *MqttClient) IsConnected() bool {
 }
 
 func (o *MqttClient) Publish(topic string, payload []byte, qos mqtt.QOS) error {
-	return o.mqtt.Publish(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Publish(ctx, topic, payload, qos)
 }
 func (o *MqttClient) PublishString(topic string, payload string, qos mqtt.QOS) error {
-	return o.mqtt.PublishString(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishString(ctx, topic, payload, qos)
 }
 func (o *MqttClient) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
-	return o.mqtt.PublishJSON(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
 }
 
 func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
@@ -106,7 +116,9 @@ func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
 	if _, ok := o.mapTopics[topic]; !ok {
 		o.mapTopics[topic] = qos
 	}
-	return o.mqtt.Subscribe(o.Ctx(), topic, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Subscribe(ctx, topic, qos)
 }
 
 func (o *MqttClient) Unsubscribe(topic string) error {
@@ -115,14 +127,15 @@ func (o *MqttClient) Unsubscribe(topic string) error {
 	if _, ok := o.mapTopics[topic]; ok {
 		delete(o.mapTopics, topic)
 	}
-	return o.mqtt.Unsubscribe(o.Ctx(), topic)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Unsubscribe(ctx, topic)
 }
 
 func (o *MqttClient) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
 	return o.mqtt.Handle(topic, handler)
 }
 
-func (o *MqttClient) Ctx() context.Context {
-	ctx, _ := context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
-	return ctx
+func (o *MqttClient) Ctx() (context.Context, context.CancelFunc) {
+	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
 }

cloud/websvr/controllers/mqtthandler.go

@@ -46,7 +46,9 @@ func NewMqttHandler(server, user, password string, timeout uint) *MqttHandler {
 		return nil
 	}
 	o.mqtt = client
-	err = client.Connect(o.Ctx())
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	err = client.Connect(ctx)
 	return &o
 }
 
@@ -68,7 +70,9 @@ func (o *MqttHandler) GetWill() (string, string) {
 
 func (o *MqttHandler) Connect() error {
 	if !o.mqtt.IsConnected() {
-		return o.mqtt.Connect(o.Ctx())
+		ctx, cancel := o.Ctx()
+		defer cancel()
+		return o.mqtt.Connect(ctx)
 	}
 	return nil
 }
@@ -78,15 +82,21 @@ func (o *MqttHandler) IsConnected() bool {
 }
 
 func (o *MqttHandler) Publish(topic string, payload []byte, qos mqtt.QOS) error {
-	return o.mqtt.Publish(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Publish(ctx, topic, payload, qos)
 }
 
 func (o *MqttHandler) PublishString(topic string, payload string, qos mqtt.QOS) error {
-	return o.mqtt.PublishString(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishString(ctx, topic, payload, qos)
 }
 
 func (o *MqttHandler) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
-	return o.mqtt.PublishJSON(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
 }
 
 func (o *MqttHandler) Subscribe(topic string, qos mqtt.QOS) error {
@@ -95,7 +105,9 @@ func (o *MqttHandler) Subscribe(topic string, qos mqtt.QOS) error {
 	if _, ok := o.mapTopics[topic]; !ok {
 		o.mapTopics[topic] = qos
 	}
-	return o.mqtt.Subscribe(o.Ctx(), topic, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Subscribe(ctx, topic, qos)
 }
 
 func (o *MqttHandler) Unsubscribe(ctx context.Context, topic string) error {
@@ -104,14 +116,15 @@ func (o *MqttHandler) Unsubscribe(ctx context.Context, topic string) error {
 	if _, ok := o.mapTopics[topic]; ok {
 		delete(o.mapTopics, topic)
 	}
-	return o.mqtt.Unsubscribe(o.Ctx(), topic)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Unsubscribe(ctx, topic)
 }
 
 func (o *MqttHandler) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
 	return o.mqtt.Handle(topic, handler)
 }
 
-func (o *MqttHandler) Ctx() context.Context {
-	ctx, _ := context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.Timeout))
-	return ctx
+func (o *MqttHandler) Ctx() (context.Context, context.CancelFunc) {
+	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.Timeout))
 }

cloud/websvr/controllers/mqtthandlerHL.go

@@ -47,7 +47,9 @@ func NewMqttHandlerhl(server, user, password string, timeout uint) *MqttHandlerh
 		return nil
 	}
 	o.mqtt = client
-	err = client.Connect(o.Ctx())
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	err = client.Connect(ctx)
 	return &o
 }
 
@@ -69,7 +71,9 @@ func (o *MqttHandlerhl) GetWill() (string, string) {
 
 func (o *MqttHandlerhl) Connect() error {
 	if !o.mqtt.IsConnected() {
-		return o.mqtt.Connect(o.Ctx())
+		ctx, cancel := o.Ctx()
+		defer cancel()
+		return o.mqtt.Connect(ctx)
 	}
 	return nil
 }
@@ -79,15 +83,21 @@ func (o *MqttHandlerhl) IsConnected() bool {
 }
 
 func (o *MqttHandlerhl) Publish(topic string, payload []byte, qos mqtt.QOS) error {
-	return o.mqtt.Publish(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Publish(ctx, topic, payload, qos)
 }
 
 func (o *MqttHandlerhl) PublishString(topic string, payload string, qos mqtt.QOS) error {
-	return o.mqtt.PublishString(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishString(ctx, topic, payload, qos)
 }
 
 func (o *MqttHandlerhl) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
-	return o.mqtt.PublishJSON(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
 }
 
 func (o *MqttHandlerhl) Subscribe(topic string, qos mqtt.QOS) error {
@@ -96,7 +106,9 @@ func (o *MqttHandlerhl) Subscribe(topic string, qos mqtt.QOS) error {
 	if _, ok := o.mapTopics[topic]; !ok {
 		o.mapTopics[topic] = qos
 	}
-	return o.mqtt.Subscribe(o.Ctx(), topic, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Subscribe(ctx, topic, qos)
 }
 
 func (o *MqttHandlerhl) Unsubscribe(ctx context.Context, topic string) error {
@@ -105,14 +117,15 @@ func (o *MqttHandlerhl) Unsubscribe(ctx context.Context, topic string) error {
 	if _, ok := o.mapTopics[topic]; ok {
 		delete(o.mapTopics, topic)
 	}
-	return o.mqtt.Unsubscribe(o.Ctx(), topic)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Unsubscribe(ctx, topic)
 }
 
 func (o *MqttHandlerhl) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
 	return o.mqtt.Handle(topic, handler)
 }
 
-func (o *MqttHandlerhl) Ctx() context.Context {
-	ctx, _ := context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.Timeout))
-	return ctx
+func (o *MqttHandlerhl) Ctx() (context.Context, context.CancelFunc) {
+	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.Timeout))
 }

edge/camera/mqttclient.go

@@ -52,7 +52,9 @@ func NewMqttClient(server, clientid, user, password string, timeout uint, mqttOn
 		return nil
 	}
 	o.mqtt = client
-	err = client.Connect(o.Ctx())
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	err = client.Connect(ctx)
 	return &o
 }
 
@@ -78,7 +80,9 @@ func (o *MqttClient) GetWill() (topic string, payload string) {
 
 func (o *MqttClient) Connect() error {
 	if !o.mqtt.IsConnected() {
-		return o.mqtt.Connect(o.Ctx())
+		ctx, cancel := o.Ctx()
+		defer cancel()
+		return o.mqtt.Connect(ctx)
 	}
 	return nil
 }
@@ -88,13 +92,19 @@ func (o *MqttClient) IsConnected() bool {
 }
 
 func (o *MqttClient) Publish(topic string, payload []byte, qos mqtt.QOS) error {
-	return o.mqtt.Publish(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Publish(ctx, topic, payload, qos)
 }
 func (o *MqttClient) PublishString(topic string, payload string, qos mqtt.QOS) error {
-	return o.mqtt.PublishString(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishString(ctx, topic, payload, qos)
 }
 func (o *MqttClient) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
-	return o.mqtt.PublishJSON(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
 }
 
 func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
@@ -103,7 +113,9 @@ func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
 	if _, ok := o.mapTopics[topic]; !ok {
 		o.mapTopics[topic] = qos
 	}
-	return o.mqtt.Subscribe(o.Ctx(), topic, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Subscribe(ctx, topic, qos)
 }
 
 func (o *MqttClient) Unsubscribe(topic string) error {
@@ -112,14 +124,15 @@ func (o *MqttClient) Unsubscribe(topic string) error {
 	if _, ok := o.mapTopics[topic]; ok {
 		delete(o.mapTopics, topic)
 	}
-	return o.mqtt.Unsubscribe(o.Ctx(), topic)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Unsubscribe(ctx, topic)
 }
 
 func (o *MqttClient) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
 	return o.mqtt.Handle(topic, handler)
 }
 
-func (o *MqttClient) Ctx() context.Context {
-	ctx, _ := context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
-	return ctx
+func (o *MqttClient) Ctx() (context.Context, context.CancelFunc) {
+	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
 }

edge/ipole/mqttclient.go

@@ -52,7 +52,9 @@ func NewMqttClient(server, clientid, user, password string, timeout uint, mqttOn
 		return nil
 	}
 	o.mqtt = client
-	err = client.Connect(o.Ctx())
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	err = client.Connect(ctx)
 	return &o
 }
 
@@ -78,7 +80,9 @@ func (o *MqttClient) GetWill() (topic string, payload string) {
 
 func (o *MqttClient) Connect() error {
 	if !o.mqtt.IsConnected() {
-		return o.mqtt.Connect(o.Ctx())
+		ctx, cancel := o.Ctx()
+		defer cancel()
+		return o.mqtt.Connect(ctx)
 	}
 	return nil
 }
@@ -88,13 +92,19 @@ func (o *MqttClient) IsConnected() bool {
 }
 
 func (o *MqttClient) Publish(topic string, payload []byte, qos mqtt.QOS) error {
-	return o.mqtt.Publish(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Publish(ctx, topic, payload, qos)
 }
 func (o *MqttClient) PublishString(topic string, payload string, qos mqtt.QOS) error {
-	return o.mqtt.PublishString(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishString(ctx, topic, payload, qos)
 }
 func (o *MqttClient) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
-	return o.mqtt.PublishJSON(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
 }
 
 func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
@@ -103,7 +113,9 @@ func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
 	if _, ok := o.mapTopics[topic]; !ok {
 		o.mapTopics[topic] = qos
 	}
-	return o.mqtt.Subscribe(o.Ctx(), topic, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Subscribe(ctx, topic, qos)
 }
 
 func (o *MqttClient) Unsubscribe(topic string) error {
@@ -112,14 +124,15 @@ func (o *MqttClient) Unsubscribe(topic string) error {
 	if _, ok := o.mapTopics[topic]; ok {
 		delete(o.mapTopics, topic)
 	}
-	return o.mqtt.Unsubscribe(o.Ctx(), topic)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Unsubscribe(ctx, topic)
 }
 
 func (o *MqttClient) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
 	return o.mqtt.Handle(topic, handler)
 }
 
-func (o *MqttClient) Ctx() context.Context {
-	ctx, _ := context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
-	return ctx
+func (o *MqttClient) Ctx() (context.Context, context.CancelFunc) {
+	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
 }

edge/its/mqttclient.go

@@ -52,7 +52,9 @@ func NewMqttClient(server, clientid, user, password string, timeout uint, mqttOn
 		return nil
 	}
 	o.mqtt = client
-	err = client.Connect(o.Ctx())
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	err = client.Connect(ctx)
 	return &o
 }
 
@@ -78,7 +80,9 @@ func (o *MqttClient) GetWill() (topic string, payload string) {
 
 func (o *MqttClient) Connect() error {
 	if !o.mqtt.IsConnected() {
-		return o.mqtt.Connect(o.Ctx())
+		ctx, cancel := o.Ctx()
+		defer cancel()
+		return o.mqtt.Connect(ctx)
 	}
 	return nil
 }
@@ -88,13 +92,19 @@ func (o *MqttClient) IsConnected() bool {
 }
 
 func (o *MqttClient) Publish(topic string, payload []byte, qos mqtt.QOS) error {
-	return o.mqtt.Publish(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Publish(ctx, topic, payload, qos)
 }
 func (o *MqttClient) PublishString(topic string, payload string, qos mqtt.QOS) error {
-	return o.mqtt.PublishString(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishString(ctx, topic, payload, qos)
 }
 func (o *MqttClient) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
-	return o.mqtt.PublishJSON(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
 }
 
 func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
@@ -103,7 +113,9 @@ func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
 	if _, ok := o.mapTopics[topic]; !ok {
 		o.mapTopics[topic] = qos
 	}
-	return o.mqtt.Subscribe(o.Ctx(), topic, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Subscribe(ctx, topic, qos)
 }
 
 func (o *MqttClient) Unsubscribe(topic string) error {
@@ -112,14 +124,15 @@ func (o *MqttClient) Unsubscribe(topic string) error {
 	if _, ok := o.mapTopics[topic]; ok {
 		delete(o.mapTopics, topic)
 	}
-	return o.mqtt.Unsubscribe(o.Ctx(), topic)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Unsubscribe(ctx, topic)
 }
 
 func (o *MqttClient) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
 	return o.mqtt.Handle(topic, handler)
 }
 
-func (o *MqttClient) Ctx() context.Context {
-	ctx, _ := context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
-	return ctx
+func (o *MqttClient) Ctx() (context.Context, context.CancelFunc) {
+	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
 }

edge/its_win/mqttclient.go

@@ -52,7 +52,9 @@ func NewMqttClient(server, clientid, user, password string, timeout uint, mqttOn
 		return nil
 	}
 	o.mqtt = client
-	err = client.Connect(o.Ctx())
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	err = client.Connect(ctx)
 	return &o
 }
 
@@ -78,7 +80,9 @@ func (o *MqttClient) GetWill() (topic string, payload string) {
 
 func (o *MqttClient) Connect() error {
 	if !o.mqtt.IsConnected() {
-		return o.mqtt.Connect(o.Ctx())
+		ctx, cancel := o.Ctx()
+		defer cancel()
+		return o.mqtt.Connect(ctx)
 	}
 	return nil
 }
@@ -88,13 +92,19 @@ func (o *MqttClient) IsConnected() bool {
 }
 
 func (o *MqttClient) Publish(topic string, payload []byte, qos mqtt.QOS) error {
-	return o.mqtt.Publish(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Publish(ctx, topic, payload, qos)
 }
 func (o *MqttClient) PublishString(topic string, payload string, qos mqtt.QOS) error {
-	return o.mqtt.PublishString(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishString(ctx, topic, payload, qos)
 }
 func (o *MqttClient) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
-	return o.mqtt.PublishJSON(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
 }
 
 func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
@@ -103,7 +113,9 @@ func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
 	if _, ok := o.mapTopics[topic]; !ok {
 		o.mapTopics[topic] = qos
 	}
-	return o.mqtt.Subscribe(o.Ctx(), topic, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Subscribe(ctx, topic, qos)
 }
 
 func (o *MqttClient) Unsubscribe(topic string) error {
@@ -112,14 +124,15 @@ func (o *MqttClient) Unsubscribe(topic string) error {
 	if _, ok := o.mapTopics[topic]; ok {
 		delete(o.mapTopics, topic)
 	}
-	return o.mqtt.Unsubscribe(o.Ctx(), topic)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Unsubscribe(ctx, topic)
 }
 
 func (o *MqttClient) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
 	return o.mqtt.Handle(topic, handler)
 }
 
-func (o *MqttClient) Ctx() context.Context {
-	ctx, _ := context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
-	return ctx
+func (o *MqttClient) Ctx() (context.Context, context.CancelFunc) {
+	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
 }

edge/led_screen/mqttclient.go

@@ -52,7 +52,9 @@ func NewMqttClient(server, clientid, user, password string, timeout uint, mqttOn
 		return nil
 	}
 	o.mqtt = client
-	err = client.Connect(o.Ctx())
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	err = client.Connect(ctx)
 	return &o
 }
 
@@ -78,7 +80,9 @@ func (o *MqttClient) GetWill() (topic string, payload string) {
 
 func (o *MqttClient) Connect() error {
 	if !o.mqtt.IsConnected() {
-		return o.mqtt.Connect(o.Ctx())
+		ctx, cancel := o.Ctx()
+		defer cancel()
+		return o.mqtt.Connect(ctx)
 	}
 	return nil
 }
@@ -88,13 +92,19 @@ func (o *MqttClient) IsConnected() bool {
 }
 
 func (o *MqttClient) Publish(topic string, payload []byte, qos mqtt.QOS) error {
-	return o.mqtt.Publish(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Publish(ctx, topic, payload, qos)
 }
 func (o *MqttClient) PublishString(topic string, payload string, qos mqtt.QOS) error {
-	return o.mqtt.PublishString(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishString(ctx, topic, payload, qos)
 }
 func (o *MqttClient) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
-	return o.mqtt.PublishJSON(o.Ctx(), topic, payload, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
 }
 
 func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
@@ -103,7 +113,9 @@ func (o *MqttClient) Subscribe(topic string, qos mqtt.QOS) error {
 	if _, ok := o.mapTopics[topic]; !ok {
 		o.mapTopics[topic] = qos
 	}
-	return o.mqtt.Subscribe(o.Ctx(), topic, qos)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Subscribe(ctx, topic, qos)
 }
 
 func (o *MqttClient) Unsubscribe(topic string) error {
@@ -112,14 +124,15 @@ func (o *MqttClient) Unsubscribe(topic string) error {
 	if _, ok := o.mapTopics[topic]; ok {
 		delete(o.mapTopics, topic)
 	}
-	return o.mqtt.Unsubscribe(o.Ctx(), topic)
+	ctx, cancel := o.Ctx()
+	defer cancel()
+	return o.mqtt.Unsubscribe(ctx, topic)
 }
 
 func (o *MqttClient) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
 	return o.mqtt.Handle(topic, handler)
 }
 
-func (o *MqttClient) Ctx() context.Context {
-	ctx, _ := context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
-	return ctx
+func (o *MqttClient) Ctx() (context.Context, context.CancelFunc) {
+	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.timeout))
 }