dispatch

Documentation

Index

• func AddResponseMetadata(existing *v1.ResponseMeta, incoming *v1.ResponseMeta)
• func CheckDepth(ctx context.Context, req DispatchableRequest) error
• func NewMaxDepthExceededError(req DispatchableRequest) error
• 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 CountingDispatchStream
  • func NewCountingDispatchStream[T any](wrapped Stream[T]) *CountingDispatchStream[T]
  • func (s *CountingDispatchStream[T]) Context() context.Context
  • func (s *CountingDispatchStream[T]) Publish(result T) error
  • func (s *CountingDispatchStream[T]) PublishedCount() uint64
• type DispatchableRequest
• type Dispatcher
• type Expand
• type HandlingDispatchStream
  • func (s *HandlingDispatchStream[T]) Context() context.Context
  • func (s *HandlingDispatchStream[T]) Publish(result T) error
• type LookupResources
• type LookupResources2
• type LookupResources2Stream
• type LookupResourcesStream
• type LookupSubjects
• type LookupSubjectsStream
• type MaxDepthExceededError
  • func (err MaxDepthExceededError) GRPCStatus() *status.Status
• type ReachableResources
• type ReachableResourcesStream
• type ReadyState
• type Stream
  • func NewHandlingDispatchStream[T any](ctx context.Context, processor func(result T) error) Stream[T]
  • 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

Functions

func AddResponseMetadata

func AddResponseMetadata(existing *v1.ResponseMeta, incoming *v1.ResponseMeta)

AddResponseMetadata adds the metadata found in the incoming metadata to the existing metadata, *modifying it in place*.

func CheckDepth

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

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

func NewMaxDepthExceededError

func NewMaxDepthExceededError(req DispatchableRequest) error

NewMaxDepthExceededError creates a new MaxDepthExceededError.

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 CountingDispatchStream

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

CountingDispatchStream is a dispatch stream that counts the number of items published. It uses an internal atomic int to ensure it is thread safe.

func NewCountingDispatchStream

func NewCountingDispatchStream[T any](wrapped Stream[T]) *CountingDispatchStream[T]

func (*CountingDispatchStream[T]) Context

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

func (*CountingDispatchStream[T]) Publish

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

func (*CountingDispatchStream[T]) PublishedCount

func (s *CountingDispatchStream[T]) PublishedCount() uint64

type DispatchableRequest

type DispatchableRequest interface {
	zerolog.LogObjectMarshaler

	GetMetadata() *v1.ResolverMeta
}

DispatchableRequest is an interface for requests.

type Dispatcher

type Dispatcher interface {
	Check
	Expand
	ReachableResources
	LookupResources
	LookupSubjects
	LookupResources2

	// Close closes the dispatcher.
	Close() error

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

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 HandlingDispatchStream

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

HandlingDispatchStream is a dispatch stream that executes a handler for each item published. It uses an internal mutex to ensure it is thread safe.

func (*HandlingDispatchStream[T]) Context

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

func (*HandlingDispatchStream[T]) Publish

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

type LookupResources

type LookupResources interface {
	// DispatchLookupResources submits a single lookup request and returns its result.
	DispatchLookupResources(
		req *v1.DispatchLookupResourcesRequest,
		stream LookupResourcesStream,
	) error
}

LookupResources interface describes just the methods required to dispatch LookupResources requests.

type LookupResources2

type LookupResources2 interface {
	DispatchLookupResources2(
		req *v1.DispatchLookupResources2Request,
		stream LookupResources2Stream,
	) error
}

type LookupResources2Stream

type LookupResources2Stream = Stream[*v1.DispatchLookupResources2Response]

type LookupResourcesStream

type LookupResourcesStream = Stream[*v1.DispatchLookupResourcesResponse]

LookupResourcesStream is an alias for the stream to which found resources will be written.

type LookupSubjects

type LookupSubjects interface {
	// DispatchLookupSubjects submits a single lookup subjects request, writing its results to the specified stream.
	DispatchLookupSubjects(
		req *v1.DispatchLookupSubjectsRequest,
		stream LookupSubjectsStream,
	) error
}

LookupSubjects interface describes just the methods required to dispatch lookup subjects requests.

type LookupSubjectsStream

type LookupSubjectsStream = Stream[*v1.DispatchLookupSubjectsResponse]

LookupSubjectsStream is an alias for the stream to which found subjects will be written.

type MaxDepthExceededError

type MaxDepthExceededError struct {

	// Request is the request that exceeded the maximum depth.
	Request DispatchableRequest
	// contains filtered or unexported fields
}

MaxDepthExceededError is an error returned when the maximum depth for dispatching has been exceeded.

func (MaxDepthExceededError) GRPCStatus

func (err MaxDepthExceededError) GRPCStatus() *status.Status

GRPCStatus implements retrieving the gRPC status for the error.

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 ReadyState

type ReadyState struct {
	// Message is a human-readable status message for the current state.
	Message string

	// IsReady indicates whether the datastore is ready.
	IsReady bool
}

ReadyState represents the ready state of the dispatcher.

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 NewHandlingDispatchStream

func NewHandlingDispatchStream[T any](ctx context.Context, processor func(result T) error) Stream[T]

NewHandlingDispatchStream returns a new handling dispatch stream.

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, bool, 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