dispatch

Documentation

Index

• Variables
• func CheckDepth(ctx context.Context, req HasMetadata) error
• func CheckRequestToKey(req *v1.DispatchCheckRequest) string
• func CheckRequestToKeyWithCanonical(req *v1.DispatchCheckRequest, canonicalKey string) string
• func ExpandRequestToKey(req *v1.DispatchExpandRequest) string
• func LookupRequestToKey(req *v1.DispatchLookupRequest) string
• func ReachableResourcesRequestToKey(req *v1.DispatchReachableResourcesRequest) string
• type Check
• type CollectingDispatchStream
  • func NewCollectingDispatchStream[T any](ctx context.Context) *CollectingDispatchStream[T]
  • func (s *CollectingDispatchStream[T]) Context() context.Context
  • func (s *CollectingDispatchStream[T]) Publish(result T) error
  • func (s *CollectingDispatchStream[T]) Results() []T
• type Dispatcher
• type Expand
• type HasMetadata
• type Lookup
• type ReachableResources
• type ReachableResourcesStream
• type Stream
  • func StreamWithContext[T any](context context.Context, stream Stream[T]) Stream[T]
  • func WrapGRPCStream[R any, S grpcStream[R]](grpcStream S) Stream[R]
• type WrappedDispatchStream
  • func (s *WrappedDispatchStream[T]) Context() context.Context
  • func (s *WrappedDispatchStream[T]) Publish(result T) error

Variables

var ErrMaxDepth = errors.New("max depth exceeded: this usually indicates a recursive or too deep data dependency")

ErrMaxDepth is returned from CheckDepth when the max depth is exceeded.

Functions

func CheckDepth

func CheckDepth(ctx context.Context, req HasMetadata) error

CheckDepth returns ErrMaxDepth if there is insufficient depth remaining to dispatch.

func CheckRequestToKey

func CheckRequestToKey(req *v1.DispatchCheckRequest) string

CheckRequestToKey converts a check request into a cache key based on the relation

func CheckRequestToKeyWithCanonical

func CheckRequestToKeyWithCanonical(req *v1.DispatchCheckRequest, canonicalKey string) string

CheckRequestToKeyWithCanonical converts a check request into a cache key based on the canonical key.

func ExpandRequestToKey

func ExpandRequestToKey(req *v1.DispatchExpandRequest) string

ExpandRequestToKey converts an expand request into a cache key

func LookupRequestToKey

func LookupRequestToKey(req *v1.DispatchLookupRequest) string

LookupRequestToKey converts a lookup request into a cache key

func ReachableResourcesRequestToKey

func ReachableResourcesRequestToKey(req *v1.DispatchReachableResourcesRequest) string

ReachableResourcesRequestToKey converts a reachable resources request into a cache key

Types

type Check

type Check interface {
	// DispatchCheck submits a single check request and returns its result.
	DispatchCheck(ctx context.Context, req *v1.DispatchCheckRequest) (*v1.DispatchCheckResponse, error)
}

Check interface describes just the methods required to dispatch check requests.

type CollectingDispatchStream

type CollectingDispatchStream[T any] struct {
	// contains filtered or unexported fields
}

CollectingDispatchStream is a dispatch stream that collects results in memory.

func NewCollectingDispatchStream

func NewCollectingDispatchStream[T any](ctx context.Context) *CollectingDispatchStream[T]

NewCollectingDispatchStream creates a new CollectingDispatchStream.

func (*CollectingDispatchStream[T]) Context

func (s *CollectingDispatchStream[T]) Context() context.Context

func (*CollectingDispatchStream[T]) Publish

func (s *CollectingDispatchStream[T]) Publish(result T) error

func (*CollectingDispatchStream[T]) Results

func (s *CollectingDispatchStream[T]) Results() []T

type Dispatcher

type Dispatcher interface {
	Check
	Expand
	Lookup
	ReachableResources

	// Close closes the dispatcher.
	Close() error

	// IsReady returns true when dispatcher is able to respond to requests
	IsReady() bool
}

Dispatcher interface describes a method for passing subchecks off to additional machines.

type Expand

type Expand interface {
	// DispatchExpand submits a single expand request and returns its result.
	DispatchExpand(ctx context.Context, req *v1.DispatchExpandRequest) (*v1.DispatchExpandResponse, error)
}

Expand interface describes just the methods required to dispatch expand requests.

type HasMetadata

type HasMetadata interface {
	zerolog.LogObjectMarshaler

	GetMetadata() *v1.ResolverMeta
}

HasMetadata is an interface for requests containing resolver metadata.

type Lookup

type Lookup interface {
	// DispatchLookup submits a single lookup request and returns its result.
	DispatchLookup(ctx context.Context, req *v1.DispatchLookupRequest) (*v1.DispatchLookupResponse, error)
}

Lookup interface describes just the methods required to dispatch lookup requests.

type ReachableResources

type ReachableResources interface {
	// DispatchReachableResources submits a single reachable resources request, writing its results to the specified stream.
	DispatchReachableResources(
		req *v1.DispatchReachableResourcesRequest,
		stream ReachableResourcesStream,
	) error
}

ReachableResources interface describes just the methods required to dispatch reachable resources requests.

type ReachableResourcesStream

type ReachableResourcesStream = Stream[*v1.DispatchReachableResourcesResponse]

ReachableResourcesStream is an alias for the stream to which reachable resources will be written.

type Stream

type Stream[T any] interface {
	// Publish publishes the result to the stream.
	Publish(T) error

	// Context returns the context for the stream.
	Context() context.Context
}

Stream defines the interface generically matching a streaming dispatch response.

func StreamWithContext

func StreamWithContext[T any](context context.Context, stream Stream[T]) Stream[T]

StreamWithContext returns the given dispatch stream, wrapped to return the given context.

func WrapGRPCStream

func WrapGRPCStream[R any, S grpcStream[R]](grpcStream S) Stream[R]

WrapGRPCStream wraps a gRPC result stream with a concurrent-safe dispatch stream. This is necessary because gRPC response streams are *not concurrent safe*. See: https://groups.google.com/g/grpc-io/c/aI6L6M4fzQ0?pli=1

type WrappedDispatchStream

type WrappedDispatchStream[T any] struct {
	Stream    Stream[T]
	Ctx       context.Context
	Processor func(result T) (T, error)
}

WrappedDispatchStream is a dispatch stream that wraps another dispatch stream, and performs an operation on each result before puppeting back up to the parent stream.

func (*WrappedDispatchStream[T]) Context

func (s *WrappedDispatchStream[T]) Context() context.Context

func (*WrappedDispatchStream[T]) Publish

func (s *WrappedDispatchStream[T]) Publish(result T) error