Make App.Call() wait for and return the result of the Home Assistant API call

app.go

@@ -15,7 +15,7 @@ import (
 
 	"saml.dev/gome-assistant/internal"
 	"saml.dev/gome-assistant/internal/http"
-	"saml.dev/gome-assistant/internal/websocket"
+	"saml.dev/gome-assistant/websocket"
 )
 
 var ErrInvalidArgs = errors.New("invalid arguments provided")
@@ -261,7 +261,14 @@ func (app *App) registerEventListener(evl EventListener) {
 			eventType := eventType
 			app.conn.SubscribeToEventType(
 				eventType,
-				func(msg websocket.ChanMsg) {
+				func(msg websocket.Message) {
+					// Subscribing, itself, causes the server to send
+					// a "result" message. We don't want to forward
+					// that message to the listeners.
+					if msg.Type != eventType {
+						return
+					}
+
 					go app.callEventListeners(eventType, msg)
 				},
 			)
@@ -328,7 +335,7 @@ func (app *App) Start() {
 
 	// subscribe to state_changed events
 	app.entitySubscription = app.conn.SubscribeToStateChangedEvents(
-		func(msg websocket.ChanMsg) {
+		func(msg websocket.Message) {
 			go app.callEntityListeners(msg.Raw)
 		},
 	)

call.go

@@ -1,17 +1,108 @@
 package gomeassistant
 
 import (
+	"context"
+	"sync"
+
 	"saml.dev/gome-assistant/internal/services"
-	"saml.dev/gome-assistant/internal/websocket"
+	"saml.dev/gome-assistant/websocket"
 )
 
-func (app *App) Call(req services.BaseServiceRequest) error {
-	req.RequestType = "call_service"
+// CallAndForget implements [services.API.CallAndForget].
+func (app *App) CallAndForget(req services.BaseServiceRequest) error {
+	reqMsg := services.CallServiceMessage{
+		BaseMessage: websocket.BaseMessage{
+			Type: "call_service",
+		},
+		BaseServiceRequest: req,
+	}
 
 	return app.conn.Send(
 		func(lc websocket.LockedConn) error {
-			req.ID = lc.NextMessageID()
-			return lc.SendMessage(req)
+			reqMsg.ID = lc.NextMessageID()
+			return lc.SendMessage(reqMsg)
+		},
+	)
+}
+
+// Call implements [services.API.Call].
+func (app *App) Call(
+	ctx context.Context, req services.BaseServiceRequest, result any,
+) error {
+	// Call works as follows:
+	//  1. Generate a message ID.
+	//  2. Subscribe to that ID.
+	//  3. Send a `CallServiceMessage` containing `req` over the websocket.
+	//  4. Wait for a single "result" message.
+	//  5. Unsubscribe from ID.
+	//  6. Unmarshal the "result" part of the response into `result`.
+
+	reqMsg := services.CallServiceMessage{
+		BaseMessage: websocket.BaseMessage{
+			Type: "call_service",
+		},
+		BaseServiceRequest: req,
+	}
+
+	// once ensures that exactly one of the following occurs:
+	//	* a single response is handled and then the handler
+	//	 unsubscribes itself; or
+	//	* (if `ctx` expires) the handler is unsubscribed if and only
+	//	 if no response has been handled.
+	var once sync.Once
+
+	// responseErr is set either to the error in the response message,
+	// or to `ctx.Err()`.
+	var responseErr error
+
+	// done is closed once a response has been processed.
+	done := make(chan struct{})
+
+	var subscription websocket.Subscription
+
+	unsubscribe := func() {
+		_ = app.conn.Send(func(lc websocket.LockedConn) error {
+			lc.Unsubscribe(subscription)
+			return nil
+		})
+	}
+
+	handleResponse := func(msg websocket.Message) {
+		once.Do(
+			func() {
+				responseErr = msg.GetResult(result)
+				unsubscribe()
+				close(done)
+			},
+		)
+	}
+
+	err := app.conn.Send(
+		func(lc websocket.LockedConn) error {
+			subscription = lc.Subscribe(handleResponse)
+			reqMsg.ID = subscription.MessageID()
+			return lc.SendMessage(reqMsg)
 		},
 	)
+	if err != nil {
+		return err
+	}
+
+	select {
+	case <-done:
+		// `handleResponse` has processed a response and set
+		// `responseErr`.
+	case <-ctx.Done():
+		// The context has expired. Unsubscribe and return
+		// `ctx.Err()`, but only if `handleResponse` hasn't just
+		// racily processed a response.
+		once.Do(
+			func() {
+				unsubscribe()
+				responseErr = ctx.Err()
+			},
+		)
+	}
+
+	return responseErr
 }

cmd/example/example.go

@@ -38,18 +38,24 @@ func main() {
 	pantryDoor := ga.
 		NewEntityListener().
 		EntityIDs(entities.BinarySensor.PantryDoor). // Use generated entity constant
-		Call(pantryLights).
+		Call(func(service *ga.Service, state ga.State, sensor ga.EntityData) {
+			pantryLights(ctx, service, state, sensor)
+		}).
 		Build()
 
 	_11pmSched := ga.
 		NewDailySchedule().
-		Call(lightsOut).
+		Call(func(service *ga.Service, state ga.State) {
+			lightsOut(ctx, service, state)
+		}).
 		At("23:00").
 		Build()
 
 	_30minsBeforeSunrise := ga.
 		NewDailySchedule().
-		Call(sunriseSched).
+		Call(func(service *ga.Service, state ga.State) {
+			sunriseSched(ctx, service, state)
+		}).
 		Sunrise("-30m").
 		Build()
 
@@ -66,13 +72,19 @@ func main() {
 	app.Start()
 }
 
-func pantryLights(service *ga.Service, state ga.State, sensor ga.EntityData) {
+func pantryLights(
+	ctx context.Context, service *ga.Service, state ga.State, sensor ga.EntityData,
+) {
 	l := "light.pantry"
 	// l := entities.Light.Pantry // Or use generated entity constant
 	if sensor.ToState == "on" {
-		service.HomeAssistant.TurnOn(l)
+		if _, err := service.HomeAssistant.TurnOn(ctx, l); err != nil {
+			slog.Warn("couldn't turn on pantry light")
+		}
 	} else {
-		service.HomeAssistant.TurnOff(l)
+		if _, err := service.HomeAssistant.TurnOff(ctx, l); err != nil {
+			slog.Warn("couldn't turn off pantry light")
+		}
 	}
 }
 
@@ -87,22 +99,33 @@ func onEvent(service *ga.Service, state ga.State, data ga.EventData) {
 	slog.Info("On event invoked", "event", ev)
 }
 
-func lightsOut(service *ga.Service, state ga.State) {
+func lightsOut(ctx context.Context, service *ga.Service, state ga.State) {
 	// always turn off outside lights
-	service.Light.TurnOff(entities.Light.OutsideLights)
+	if _, err := service.Light.TurnOff(ctx, entities.Light.OutsideLights); err != nil {
+		slog.Warn("couldn't turn off living room light, doing nothing")
+		return
+	}
 	s, err := state.Get(entities.BinarySensor.LivingRoomMotion)
 	if err != nil {
-		slog.Warn("couldnt get living room motion state, doing nothing")
+		slog.Warn("couldn't get living room motion state, doing nothing")
 		return
 	}
 
 	// if no motion detected in living room for 30mins
 	if s.State == "off" && time.Since(s.LastChanged).Minutes() > 30 {
-		service.Light.TurnOff(entities.Light.MainLights)
+		if _, err := service.Light.TurnOff(ctx, entities.Light.MainLights); err != nil {
+			slog.Warn("couldn't turn off living light")
+			return
+		}
 	}
 }
 
-func sunriseSched(service *ga.Service, state ga.State) {
-	service.Light.TurnOn(entities.Light.LivingRoomLamps)
-	service.Light.TurnOff(entities.Light.ChristmasLights)
+func sunriseSched(ctx context.Context, service *ga.Service, state ga.State) {
+	if _, err := service.Light.TurnOn(ctx, entities.Light.LivingRoomLamps); err != nil {
+		slog.Warn("couldn't turn on living light")
+	}
+
+	if _, err := service.Light.TurnOff(ctx, entities.Light.ChristmasLights); err != nil {
+		slog.Warn("couldn't turn off Christmas lights")
+	}
 }

cmd/example/example_live_test.go

@@ -106,11 +106,12 @@ func (s *MySuite) TearDownSuite() {
 
 // Basic test of light toggle service and entity listener
 func (s *MySuite) TestLightService() {
+	ctx := context.TODO()
 	entityID := s.config.Entities.LightEntityID
 
 	if entityID != "" {
 		initState := getEntityState(s, entityID)
-		s.app.GetService().Light.Toggle(entityID)
+		s.app.GetService().Light.Toggle(ctx, entityID)
 
 		assert.EventuallyWithT(s.T(), func(c *assert.CollectT) {
 			newState := getEntityState(s, entityID)

entitylistener.go

@@ -8,6 +8,7 @@ import (
 	"github.com/golang-module/carbon"
 
 	"saml.dev/gome-assistant/internal"
+	"saml.dev/gome-assistant/websocket"
 )
 
 type EntityListener struct {
@@ -45,9 +46,8 @@ type EntityData struct {
 	LastChanged     time.Time
 }
 
-type stateChangedMsg struct {
-	ID    int    `json:"id"`
-	Type  string `json:"type"`
+type stateChangedMessage struct {
+	websocket.BaseMessage
 	Event struct {
 		Data      stateData `json:"data"`
 		EventType string    `json:"event_type"`
@@ -239,7 +239,7 @@ func (l *EntityListener) maybeCall(app *App, entityData EntityData, data stateDa
 
 /* Functions */
 func (app *App) callEntityListeners(msgBytes []byte) {
-	msg := stateChangedMsg{}
+	msg := stateChangedMessage{}
 	_ = json.Unmarshal(msgBytes, &msg)
 	data := msg.Event.Data
 	eid := data.EntityID

eventListener.go

@@ -7,7 +7,7 @@ import (
 	"github.com/golang-module/carbon"
 
 	"saml.dev/gome-assistant/internal"
-	"saml.dev/gome-assistant/internal/websocket"
+	"saml.dev/gome-assistant/websocket"
 )
 
 type EventListener struct {
@@ -158,7 +158,7 @@ func (l *EventListener) maybeCall(app *App, eventData EventData) {
 }
 
 /* Functions */
-func (app *App) callEventListeners(eventType string, msg websocket.ChanMsg) {
+func (app *App) callEventListeners(eventType string, msg websocket.Message) {
 	listeners, ok := app.eventListeners[eventType]
 	if !ok {
 		// no listeners registered for this event type

fire_event.go

@@ -1,7 +1,8 @@
 package gomeassistant
 
-import "saml.dev/gome-assistant/internal/websocket"
+import "saml.dev/gome-assistant/websocket"
 
+// FireEvent implements [services.API.FireEvent].
 func (app *App) FireEvent(eventType string, eventData map[string]any) error {
 	return app.conn.Send(
 		func(lc websocket.LockedConn) error {

internal/services/adaptive_lighting.go

@@ -1,5 +1,7 @@
 package services
 
+import "context"
+
 /* Structs */
 
 type AdaptiveLighting struct {
@@ -9,7 +11,9 @@ type AdaptiveLighting struct {
 /* Public API */
 
 // Set manual control for an adaptive lighting entity.
-func (al AdaptiveLighting) SetManualControl(entityID string, enabled bool) error {
+func (al AdaptiveLighting) SetManualControl(
+	ctx context.Context, entityID string, enabled bool,
+) (any, error) {
 	req := BaseServiceRequest{
 		Domain:  "adaptive_lighting",
 		Service: "set_manual_control",
@@ -20,5 +24,10 @@ func (al AdaptiveLighting) SetManualControl(entityID string, enabled bool) error
 		Target: Entity(entityID),
 	}
 
-	return al.api.Call(req)
+	var result any
+	if err := al.api.Call(ctx, req, &result); err != nil {
+		return nil, err
+	}
+
+	return result, nil
 }