fix: refactor for changes to starpc

Signed-off-by: Christian Stewart <christian@aperture.us>

Author: paralin

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/aperturerobotics/controllerbus v0.50.1 // latest
 	github.com/aperturerobotics/entitygraph v0.11.0 // latest
 	github.com/aperturerobotics/protobuf-go-lite v0.9.1 // latest
-	github.com/aperturerobotics/starpc v0.38.1 // latest
+	github.com/aperturerobotics/starpc v0.39.0 // latest
 	github.com/aperturerobotics/util v1.30.0 // latest
 )
 

go.sum

@@ -28,8 +28,8 @@ github.com/aperturerobotics/protobuf-go-lite v0.9.1 h1:P1knXKnwLJpVE8fmeXYGckKu7
 github.com/aperturerobotics/protobuf-go-lite v0.9.1/go.mod h1:fULrxQxEBWKQm7vvju9AfjTp9yfHoLgwMQWTiZQ2tg0=
 github.com/aperturerobotics/quic-go v0.48.2-0.20241029082227-fa76c393ee89 h1:MUXLrLDjCw36dx8OeBwF/B7YP5akyoU6Ps+v50qR5Lo=
 github.com/aperturerobotics/quic-go v0.48.2-0.20241029082227-fa76c393ee89/go.mod h1:vhhkHYq2i9lb0bOllqKhOdtg/3i7r4o/7zc8XI/T4lo=
-github.com/aperturerobotics/starpc v0.38.1 h1:pdDfhrTn6nsEtn+/qYbFNgXRHnlJNb/MdbF9CMqgMoA=
-github.com/aperturerobotics/starpc v0.38.1/go.mod h1:IT2VZ+tDYFleXhcOSvxcUe1fyVk8rD6JCISwjFeA6OA=
+github.com/aperturerobotics/starpc v0.39.0 h1:n/NR16KdzulIZFTf6PYo53mFHti4vg+O65kgaBGEXh4=
+github.com/aperturerobotics/starpc v0.39.0/go.mod h1:GVNyIvtGf/VO/MpWLCPiGMT9GVBFcdruSivsgD4oKcE=
 github.com/aperturerobotics/util v1.30.0 h1:OKhFVPnAfR8/dfVNV27EtMr27C0kzwPiStoCwKiint0=
 github.com/aperturerobotics/util v1.30.0/go.mod h1:T97YTP+FVLegYo5rylOVaPuTLyZyiDqYxD5zVLI9YAc=
 github.com/blang/semver/v4 v4.0.0 h1:1PFHFE6yCCTv8C1TeyNNarDzntLi7wMI5i/pzqYIsAM=

rpc/access/server.go

@@ -158,7 +158,7 @@ func (s *AccessRpcServiceServer) LookupRpcService(
 
 // CallRpcService looks up the rpc service with the request & invokes the RPC.
 func (s *AccessRpcServiceServer) CallRpcService(strm SRPCAccessRpcService_CallRpcServiceStream) error {
-	return rpcstream.HandleRpcStream(strm, func(ctx context.Context, componentID string) (srpc.Invoker, func(), error) {
+	return rpcstream.HandleRpcStream(strm, func(ctx context.Context, componentID string, released func()) (srpc.Invoker, func(), error) {
 		// parse component id json
 		req := &LookupRpcServiceRequest{}
 		if err := req.UnmarshalComponentID(componentID); err != nil {
@@ -182,6 +182,7 @@ func (s *AccessRpcServiceServer) CallRpcService(strm SRPCAccessRpcService_CallRp
 			req.GetServiceId(),
 			serverID,
 			s.waitOne,
+			released,
 		)
 		if err != nil || invokerRef == nil {
 			return nil, nil, err

rpc/invoker.go

@@ -34,7 +34,7 @@ func NewInvoker(b bus.Bus, serverID string, wait bool) *Invoker {
 func (i *Invoker) InvokeMethod(serviceID, methodID string, strm srpc.Stream) (bool, error) {
 	ctx := strm.Context()
 
-	invokers, _, invokerRef, err := ExLookupRpcService(ctx, i.b, serviceID, i.serverID, i.wait)
+	invokers, _, invokerRef, err := ExLookupRpcService(ctx, i.b, serviceID, i.serverID, i.wait, nil)
 	if err != nil || invokerRef == nil {
 		return false, err
 	}

rpc/lookup-rpc-service.go

@@ -52,18 +52,21 @@ func NewLookupRpcService(serviceID, serverID string) LookupRpcService {
 // If values are returned, returns vals, valsRef, nil
 // Otherwise returns nil, nil, err
 // If waitOne is set, waits for at least one value before returning.
+// valDisposeCb is called if any of the values are no longer valid.
+// valDisposeCb might be called multiple times.
 func ExLookupRpcService(
 	ctx context.Context,
 	b bus.Bus,
 	serviceID, serverID string,
 	waitOne bool,
+	valDisposeCb func(),
 ) ([]LookupRpcServiceValue, directive.Instance, directive.Reference, error) {
 	out, di, valsRef, err := bus.ExecCollectValues[LookupRpcServiceValue](
 		ctx,
 		b,
 		NewLookupRpcService(serviceID, serverID),
 		waitOne,
-		nil,
+		valDisposeCb,
 	)
 	if err != nil {
 		return nil, nil, nil, err