bldr/resource/routed-client.go

Rama master · réplica de lectura del repositorio original.

package resource

import (
	"context"
	"strconv"
	"strings"
	"sync"

	"github.com/aperturerobotics/starpc/srpc"
)

// routedClient wraps an SRPC client and prefixes service IDs with a resource
// ID for routing. Used by the server to target specific attached resources
// over a shared yamux session.
type routedClient struct {
	inner  srpc.Client
	prefix string
	// done closes when the routed resource is detached.
	done <-chan struct{}
}

// NewRoutedClient wraps an SRPC client so all calls are prefixed with the
// resource ID for routing to the correct attached resource mux.
func NewRoutedClient(inner srpc.Client, resourceID uint32) srpc.Client {
	return NewRoutedClientWithDone(inner, resourceID, nil)
}

// NewRoutedClientWithDone wraps a client with a resource route and its lifetime.
func NewRoutedClientWithDone(inner srpc.Client, resourceID uint32, done <-chan struct{}) srpc.Client {
	return &routedClient{
		inner:  inner,
		prefix: strconv.FormatUint(uint64(resourceID), 10) + "/",
		done:   done,
	}
}

// Done closes when the routed resource is detached.
func (c *routedClient) Done() <-chan struct{} {
	return c.done
}

// ExecCall executes a request/reply RPC with the resource ID prefix.
func (c *routedClient) ExecCall(ctx context.Context, service, method string, in, out srpc.Message) error {
	return c.inner.ExecCall(ctx, c.prefix+service, method, in, out)
}

// NewStream opens a streaming RPC with the resource ID prefix.
func (c *routedClient) NewStream(ctx context.Context, service, method string, firstMsg srpc.Message) (srpc.Stream, error) {
	return c.inner.NewStream(ctx, c.prefix+service, method, firstMsg)
}

// RoutedInvoker dispatches SRPC calls to the correct mux by parsing a
// resource ID prefix from the service ID. Used by the client to route
// incoming yamux sub-stream RPCs to the right attached resource.
type RoutedInvoker struct {
	mu        sync.Mutex
	muxes     map[uint32]srpc.Invoker
	contextFn func(context.Context, uint32) context.Context
}

// NewRoutedInvoker creates a new RoutedInvoker.
func NewRoutedInvoker() *RoutedInvoker {
	return &RoutedInvoker{muxes: make(map[uint32]srpc.Invoker)}
}

// NewRoutedInvokerWithContext creates a RoutedInvoker that can replace the
// stream context before invoking a routed mux.
func NewRoutedInvokerWithContext(contextFn func(context.Context, uint32) context.Context) *RoutedInvoker {
	return &RoutedInvoker{
		muxes:     make(map[uint32]srpc.Invoker),
		contextFn: contextFn,
	}
}

// SetMux registers or replaces a mux for a resource ID.
func (r *RoutedInvoker) SetMux(resourceID uint32, mux srpc.Invoker) {
	r.mu.Lock()
	r.muxes[resourceID] = mux
	r.mu.Unlock()
}

// RemoveMux removes a mux for a resource ID.
func (r *RoutedInvoker) RemoveMux(resourceID uint32) {
	r.mu.Lock()
	delete(r.muxes, resourceID)
	r.mu.Unlock()
}

// InvokeMethod parses the resource ID prefix from serviceID and dispatches.
func (r *RoutedInvoker) InvokeMethod(serviceID, methodID string, strm srpc.Stream) (bool, error) {
	// Parse the resource ID prefix from the service ID.
	prefix, rest, ok := strings.Cut(serviceID, "/")
	if !ok {
		return false, nil
	}
	id, err := strconv.ParseUint(prefix, 10, 32)
	if err != nil {
		return false, nil
	}

	// Look up the mux and context override for the resource ID.
	r.mu.Lock()
	mux := r.muxes[uint32(id)]
	contextFn := r.contextFn
	r.mu.Unlock()
	if mux == nil {
		return false, ErrResourceNotFound
	}

	// Wrap the stream context and dispatch to the mux.
	if contextFn != nil {
		strm = srpc.NewStreamWithContext(strm, contextFn(strm.Context(), uint32(id)))
	}
	return mux.InvokeMethod(rest, methodID, strm)
}

// _ is a type assertion
var _ srpc.Client = (*routedClient)(nil)

// _ is a type assertion
var _ srpc.Invoker = (*RoutedInvoker)(nil)