diff --git a/plugin/mqtt.go b/plugin/mqtt.go index 62b8cf37d..a6582dbe5 100644 --- a/plugin/mqtt.go +++ b/plugin/mqtt.go @@ -12,6 +12,7 @@ import ( // Mqtt provider type Mqtt struct { *getter + ctx context.Context log *util.Logger client *mqtt.Client topic string @@ -49,7 +50,7 @@ func NewMqttPluginFromConfig(ctx context.Context, other map[string]any) (Plugin, return nil, err } - m := NewMqtt(log, client, cc.Topic, cc.Timeout).WithScale(cc.Scale).WithPayload(cc.Payload) + m := NewMqtt(log, client, cc.Topic, cc.Timeout).WithContext(ctx).WithScale(cc.Scale).WithPayload(cc.Payload) if cc.Retained { m = m.WithRetained() } @@ -65,6 +66,7 @@ func NewMqttPluginFromConfig(ctx context.Context, other map[string]any) (Plugin, // NewMqtt creates mqtt provider for given topic func NewMqtt(log *util.Logger, client *mqtt.Client, topic string, timeout time.Duration) *Mqtt { m := &Mqtt{ + ctx: context.Background(), log: log, client: client, topic: topic, @@ -76,6 +78,12 @@ func NewMqtt(log *util.Logger, client *mqtt.Client, topic string, timeout time.D return m } +// WithContext sets a context that can cancel the blocking first-read wait +func (m *Mqtt) WithContext(ctx context.Context) *Mqtt { + m.ctx = ctx + return m +} + // WithPayload adds payload for setters func (m *Mqtt) WithPayload(payload string) *Mqtt { m.payload = payload @@ -103,6 +111,7 @@ func (p *Mqtt) WithPipeline(pipeline *pipeline.Pipeline) *Mqtt { // newReceiver creates a msgHandler and subscribes it to the topic. func (m *Mqtt) newReceiver() (*msgHandler, error) { h := &msgHandler{ + ctx: m.ctx, topic: m.topic, pipeline: m.pipeline, val: util.NewMonitor[string](m.timeout), diff --git a/plugin/mqtt_handler.go b/plugin/mqtt_handler.go index f993a5ce2..7e02ce3a3 100644 --- a/plugin/mqtt_handler.go +++ b/plugin/mqtt_handler.go @@ -1,11 +1,14 @@ package plugin import ( + "context" + "github.com/evcc-io/evcc/plugin/pipeline" "github.com/evcc-io/evcc/util" ) type msgHandler struct { + ctx context.Context topic string pipeline *pipeline.Pipeline val *util.Monitor[string] @@ -17,7 +20,7 @@ func (h *msgHandler) receive(payload string) { // hasValue returned the received and processed payload as string func (h *msgHandler) hasValue() (string, error) { - payload, err := h.val.Get() + payload, err := h.val.GetContext(h.ctx) if err != nil { return "", err } diff --git a/util/monitor.go b/util/monitor.go index ab5497c7f..1ee8151cf 100644 --- a/util/monitor.go +++ b/util/monitor.go @@ -1,6 +1,7 @@ package util import ( + "context" "sync" "time" @@ -54,8 +55,14 @@ func (m *Monitor[T]) SetFunc(set func(T) T) { // Get returns the current value or ErrOutdated if timeout exceeded func (m *Monitor[T]) Get() (T, error) { + return m.GetContext(context.Background()) +} + +// GetContext returns the current value or ErrOutdated if timeout exceeded. +// The context can cancel the blocking first-call wait. +func (m *Monitor[T]) GetContext(ctx context.Context) (T, error) { var res T - err := m.GetFunc(func(v T) { + err := m.GetFuncContext(ctx, func(v T) { res = v }) return res, err @@ -63,6 +70,12 @@ func (m *Monitor[T]) Get() (T, error) { // GetFunc returns the current value or ErrOutdated if timeout exceeded while holding the lock func (m *Monitor[T]) GetFunc(get func(T)) error { + return m.GetFuncContext(context.Background(), get) +} + +// GetFuncContext returns the current value or ErrOutdated if timeout exceeded while holding the lock. +// The context can cancel the blocking first-call wait. +func (m *Monitor[T]) GetFuncContext(ctx context.Context, get func(T)) error { m.mu.RLock() defer m.mu.RUnlock() @@ -94,6 +107,8 @@ func (m *Monitor[T]) GetFunc(get func(T)) error { case <-m.done: // got value and updated timestamp err = nil + case <-ctx.Done(): + err = ctx.Err() case <-m.clock.After(m.timeout): } diff --git a/util/monitor_test.go b/util/monitor_test.go index b1e8bd8f6..60be90d40 100644 --- a/util/monitor_test.go +++ b/util/monitor_test.go @@ -1,6 +1,7 @@ package util import ( + "context" "math/rand" "testing" "time" @@ -69,3 +70,16 @@ func TestMonitorWithoutTimeout(t *testing.T) { _, err = m.Get() assert.NoError(t, err) } + +func TestMonitorGetContext(t *testing.T) { + // a long timeout would block the first-read wait; a cancelled context cuts it short + m := NewMonitor[int](time.Minute) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + start := time.Now() + _, err := m.GetContext(ctx) + assert.ErrorIs(t, err, context.DeadlineExceeded) + assert.Less(t, time.Since(start), time.Second, "context must cancel the first-read wait") +}