bldr/resource/resource-rpc-client.go

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

package resource

import (
	"context"

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

// resourceRpcCallKey carries the unary call's route into its stream opener.
type resourceRpcCallKey struct{}

// resourceRpcCall retains the transport until ExecCall settles its result.
type resourceRpcCall struct {
	// stream is set by the opener before it negotiates the route.
	stream ResourceRpcStream
	// cancel ends the transport context after the result is settled.
	cancel context.CancelFunc
}

// resourceRpcClient abandons pending children when a unary call fails.
type resourceRpcClient struct {
	// inner opens and executes the nested SRPC call.
	inner srpc.Client
}

// ExecCall retains the route until its result decides whether to abandon it.
func (c *resourceRpcClient) ExecCall(ctx context.Context, service, method string, in, out srpc.Message) error {
	// Give this call one route record without sharing state between calls.
	call := &resourceRpcCall{}
	err := c.inner.ExecCall(context.WithValue(ctx, resourceRpcCallKey{}, call), service, method, in, out)
	if call.cancel != nil {
		defer call.cancel()
	}

	// A failed result leaves every returned child ID unavailable to the caller.
	if call.stream != nil {
		if err != nil {
			// A broken transport can reject abandonment; retain the original error.
			_ = call.stream.Send(&ResourceRpcPacket{Body: &ResourceRpcPacket_Abandon{Abandon: &ResourceRpcAbandon{}}})
			_ = call.stream.Close()
			return err
		}

		// Complete the successful route before releasing its context and hooks.
		_ = call.stream.CloseSend()
		_ = call.stream.Close()
	}
	return err
}

// NewStream preserves the streaming caller's control over delivered messages.
func (c *resourceRpcClient) NewStream(ctx context.Context, service, method string, firstMsg srpc.Message) (srpc.Stream, error) {
	return c.inner.NewStream(ctx, service, method, firstMsg)
}

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