Split out a new message package

app.go

@@ -15,7 +15,8 @@ import (
 
 	"saml.dev/gome-assistant/internal"
 	"saml.dev/gome-assistant/internal/http"
-	"saml.dev/gome-assistant/internal/websocket"
+	"saml.dev/gome-assistant/message"
+	"saml.dev/gome-assistant/websocket"
 )
 
 var ErrInvalidArgs = errors.New("invalid arguments provided")
@@ -261,7 +262,14 @@ func (app *App) registerEventListener(evl EventListener) {
 			eventType := eventType
 			app.conn.SubscribeToEventType(
 				eventType,
-				func(msg websocket.ChanMsg) {
+				func(msg message.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 +336,7 @@ func (app *App) Start() {
 
 	// subscribe to state_changed events
 	app.entitySubscription = app.conn.SubscribeToStateChangedEvents(
-		func(msg websocket.ChanMsg) {
+		func(msg message.Message) {
 			go app.callEntityListeners(msg.Raw)
 		},
 	)
@@ -345,13 +353,18 @@ func (app *App) Start() {
 				}
 
 				etl.runOnStartupCompleted = true
-				go etl.callback(app.service, app.state, EntityData{
-					TriggerEntityID: eid,
-					FromState:       entityState.State,
-					FromAttributes:  entityState.Attributes,
-					ToState:         entityState.State,
-					ToAttributes:    entityState.Attributes,
-					LastChanged:     entityState.LastChanged,
+				stateChangedState := message.StateChangedState{
+					EntityID:    eid,
+					LastChanged: entityState.LastChanged,
+					State:       entityState.State,
+					Attributes:  entityState.Attributes,
+					LastUpdated: entityState.LastChanged,
+				}
+
+				go etl.callback(app.service, app.state, message.StateChangedData{
+					EntityID: eid,
+					NewState: stateChangedState,
+					OldState: stateChangedState,
 				})
 			}
 		}

call.go

@@ -1,17 +1,108 @@
 package gomeassistant
 
 import (
-	"saml.dev/gome-assistant/internal/services"
-	"saml.dev/gome-assistant/internal/websocket"
+	"context"
+	"sync"
+
+	"saml.dev/gome-assistant/message"
+	"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 message.CallServiceData) error {
+	reqMsg := message.CallServiceRequest{
+		BaseMessage: message.BaseMessage{
+			Type: "call_service",
+		},
+		CallServiceData: 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 message.CallServiceData, result any,
+) error {
+	// Call works as follows:
+	//  1. Generate a message ID.
+	//  2. Subscribe to that ID.
+	//  3. Send a `CallServiceRequest` 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 := message.CallServiceRequest{
+		BaseMessage: message.BaseMessage{
+			Type: "call_service",
+		},
+		CallServiceData: 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 message.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
 }

checkers_test.go

@@ -6,12 +6,13 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"saml.dev/gome-assistant/internal"
+	"saml.dev/gome-assistant/message"
 )
 
 type MockState struct {
 	EqualsReturn bool
 	EqualsError  bool
-	GetReturn    EntityState
+	GetReturn    message.EntityState
 	GetError     bool
 }
 
@@ -27,14 +28,14 @@ func (s MockState) AfterSunset(_ ...DurationString) bool {
 func (s MockState) BeforeSunset(_ ...DurationString) bool {
 	return true
 }
-func (s MockState) Get(eid string) (EntityState, error) {
+func (s MockState) Get(eid string) (message.EntityState, error) {
 	if s.GetError {
-		return EntityState{}, errors.New("some error")
+		return message.EntityState{}, errors.New("some error")
 	}
 	return s.GetReturn, nil
 }
-func (s MockState) ListEntities() ([]EntityState, error) {
-	return []EntityState{}, nil
+func (s MockState) ListEntities() ([]message.EntityState, error) {
+	return []message.EntityState{}, nil
 }
 func (s MockState) Equals(eid, state string) (bool, error) {
 	if s.EqualsError {

cmd/example/example.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"saml.dev/gome-assistant/cmd/example/entities" // Optional import generated entities
+	"saml.dev/gome-assistant/message"
 
 	ga "saml.dev/gome-assistant"
 )
@@ -38,18 +39,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 message.StateChangedData) {
+			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,43 +73,60 @@ 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 message.StateChangedData,
+) {
 	l := "light.pantry"
 	// l := entities.Light.Pantry // Or use generated entity constant
-	if sensor.ToState == "on" {
-		service.HomeAssistant.TurnOn(l)
+	if sensor.NewState.State == "on" {
+		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")
+		}
 	}
 }
 
-func onEvent(service *ga.Service, state ga.State, data ga.EventData) {
+func onEvent(service *ga.Service, state ga.State, msg message.Message) {
 	// Since the structure of the event changes depending
 	// on the event type, you can Unmarshal the raw json
 	// into a Go type. If a type for your event doesn't
 	// exist, you can write it yourself! PR's welcome to
 	// the eventTypes.go file :)
-	ev := ga.EventZWaveJSValueNotification{}
-	json.Unmarshal(data.RawEventJSON, &ev)
+	ev := message.ZWaveJSValueNotificationEventMessage{}
+	json.Unmarshal(msg.Raw, &ev)
 	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 s.State == "off" && time.Since(s.LastChanged.Time()).Minutes() > 30 {
+		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

@@ -14,6 +14,7 @@ import (
 	"gopkg.in/yaml.v3"
 
 	ga "saml.dev/gome-assistant"
+	"saml.dev/gome-assistant/message"
 )
 
 type (
@@ -106,11 +107,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)
@@ -130,8 +132,13 @@ func (s *MySuite) TestSchedule() {
 }
 
 // Capture event after light entity state has changed
-func (s *MySuite) entityCallback(se *ga.Service, st ga.State, e ga.EntityData) {
-	slog.Info("Entity callback called.", "entity id", e.TriggerEntityID, "from state", e.FromState, "to state", e.ToState)
+func (s *MySuite) entityCallback(se *ga.Service, st ga.State, e message.StateChangedData) {
+	slog.Info(
+		"Entity callback called",
+		"entity id", e.EntityID,
+		"from state", e.OldState.State,
+		"to state", e.NewState.State,
+	)
 	s.suiteCtx.entityCallbackInvoked.Store(true)
 }