nexus

Documentation

Index

• Constants
• Variables
• func APIFailureToNexusFailure(failure *failurepb.Failure) *nexus.Failure
• func AdaptAuthorizeError(err error) error
• func ConvertGRPCError(err error, exposeDetails bool) error
• func FormatDuration(d time.Duration) string
• func HandlerErrorTypeFromHTTPStatus(statusCode int) nexus.HandlerErrorType
• func NexusFailureToProtoFailure(failure *nexus.Failure) *nexuspb.Failure
• func ProtoFailureToNexusFailure(failure *nexuspb.Failure) *nexus.Failure
• func UnsuccessfulOperationErrorToTemporalFailure(err *nexus.UnsuccessfulOperationError) *failurepb.Failure
• type CallbackToken
  • func DecodeCallbackToken(encoded string) (token *CallbackToken, err error)
• type CallbackTokenGenerator
  • func NewCallbackTokenGenerator() *CallbackTokenGenerator
  • func (g *CallbackTokenGenerator) DecodeCompletion(token *CallbackToken) (*tokenspb.NexusOperationCompletion, error)
  • func (g *CallbackTokenGenerator) Tokenize(completion *tokenspb.NexusOperationCompletion) (string, error)
• type EndpointRegistry
• type EndpointRegistryConfig
  • func NewEndpointRegistryConfig(dc *dynamicconfig.Collection) *EndpointRegistryConfig
• type EndpointRegistryImpl
  • func NewEndpointRegistry(config *EndpointRegistryConfig, ...) *EndpointRegistryImpl
  • func (r *EndpointRegistryImpl) GetByID(ctx context.Context, id string) (*persistencespb.NexusEndpointEntry, error)
  • func (r *EndpointRegistryImpl) GetByName(ctx context.Context, _ namespace.ID, endpointName string) (*persistencespb.NexusEndpointEntry, error)
  • func (r *EndpointRegistryImpl) StartLifecycle()
  • func (r *EndpointRegistryImpl) StopLifecycle()
• type NamespaceAndTaskQueue

Constants

const (
	// Currently supported token version.
	TokenVersion = 1
	// Header key for the callback token in StartOperation requests.
	CallbackTokenHeader = "Temporal-Callback-Token"
)

Variables

var ErrNexusDisabled = serviceerror.NewFailedPrecondition("nexus is disabled")

var PayloadSerializer nexus.Serializer = payloadSerializer{}

var RouteCompletionCallback = routing.NewBuilder[string]().
	Constant("namespaces").
	StringVariable("namespace", func(namespace *string) *string { return namespace }).
	Constant("nexus", "callback").
	Build()

RouteCompletionCallback is an HTTP route for completing a Nexus operation via callback.

var RouteDispatchNexusTaskByEndpoint = routing.NewBuilder[string]().
	Constant("nexus", "endpoints").
	StringVariable("endpoint", func(endpoint *string) *string { return endpoint }).
	Constant("services").
	Build()

var RouteDispatchNexusTaskByNamespaceAndTaskQueue = routing.NewBuilder[NamespaceAndTaskQueue]().
	Constant("namespaces").
	StringVariable("namespace", func(params *NamespaceAndTaskQueue) *string { return &params.Namespace }).
	Constant("task-queues").
	StringVariable("task_queue", func(params *NamespaceAndTaskQueue) *string { return &params.TaskQueue }).
	Constant("nexus-services").
	Build()

Functions

func APIFailureToNexusFailure

func APIFailureToNexusFailure(failure *failurepb.Failure) *nexus.Failure

APIFailureToNexusFailure converts an API proto Failure to a Nexus SDK Failure taking only the failure message to avoid leaking too many details to 3rd party callers. Always returns a non-nil value.

func AdaptAuthorizeError

func AdaptAuthorizeError(err error) error

func ConvertGRPCError

func ConvertGRPCError(err error, exposeDetails bool) error

ConvertGRPCError converts either a serviceerror or a gRPC status error into a Nexus HandlerError if possible. If exposeDetails is true, the error message from the given error is exposed in the converted HandlerError, otherwise, a default message with minimal information is attached to the returned error. Roughly taken from https://github.com/googleapis/googleapis/blob/master/google/rpc/code.proto and https://github.com/grpc-ecosystem/grpc-gateway/blob/a7cf811e6ffabeaddcfb4ff65602c12671ff326e/runtime/errors.go#L56.

func FormatDuration

func FormatDuration(d time.Duration) string

FormatDuration converts a duration into a string representation in millisecond resolution. TODO: replace this with the version exported from the Nexus SDK

func HandlerErrorTypeFromHTTPStatus

func HandlerErrorTypeFromHTTPStatus(statusCode int) nexus.HandlerErrorType

func NexusFailureToProtoFailure

func NexusFailureToProtoFailure(failure *nexus.Failure) *nexuspb.Failure

NexusFailureToProtoFailure converts a Nexus SDK Failure to a proto Nexus Failure. Always returns a non-nil value.

func ProtoFailureToNexusFailure

func ProtoFailureToNexusFailure(failure *nexuspb.Failure) *nexus.Failure

ProtoFailureToNexusFailure converts a proto Nexus Failure to a Nexus SDK Failure. Always returns a non-nil value.

func UnsuccessfulOperationErrorToTemporalFailure

func UnsuccessfulOperationErrorToTemporalFailure(err *nexus.UnsuccessfulOperationError) *failurepb.Failure

Types

type CallbackToken

type CallbackToken struct {
	// Token version - currently only [TokenVersion] is supported.
	Version int `json:"v"`
	// Encoded [tokenspb.NexusOperationCompletion].
	Data string `json:"d"`
}

CallbackToken contains an encoded NexusOperationCompletion message.

func DecodeCallbackToken

func DecodeCallbackToken(encoded string) (token *CallbackToken, err error)

DecodeCallbackToken unmarshals the given token applying minimal data verification.

type CallbackTokenGenerator

type CallbackTokenGenerator struct {
}

func NewCallbackTokenGenerator

func NewCallbackTokenGenerator() *CallbackTokenGenerator

func (*CallbackTokenGenerator) DecodeCompletion

func (g *CallbackTokenGenerator) DecodeCompletion(token *CallbackToken) (*tokenspb.NexusOperationCompletion, error)

DecodeCompletion decodes a callback token unwrapping the contained NexusOperationCompletion proto struct.

func (*CallbackTokenGenerator) Tokenize

func (g *CallbackTokenGenerator) Tokenize(completion *tokenspb.NexusOperationCompletion) (string, error)

type EndpointRegistry

type EndpointRegistry interface {
	// GetByName returns an endpoint entry for the endpoint name for a caller from the given namespace ID.
	// Note that the default implementation is global to the cluster and can ignore the namespace ID param.
	GetByName(ctx context.Context, namespaceID namespace.ID, endpointName string) (*persistencespb.NexusEndpointEntry, error)
	GetByID(ctx context.Context, endpointID string) (*persistencespb.NexusEndpointEntry, error)
	StartLifecycle()
	StopLifecycle()
}

type EndpointRegistryConfig

type EndpointRegistryConfig struct {
	// contains filtered or unexported fields
}

func NewEndpointRegistryConfig

func NewEndpointRegistryConfig(dc *dynamicconfig.Collection) *EndpointRegistryConfig

type EndpointRegistryImpl

type EndpointRegistryImpl struct {
	// contains filtered or unexported fields
}

EndpointRegistryImpl manages a cached view of Nexus endpoints. Endpoints are lazily-loaded into memory on the first read. Thereafter, endpoint data is kept up to date by background long polling on matching service ListNexusEndpoints.

func NewEndpointRegistry

func NewEndpointRegistry(
	config *EndpointRegistryConfig,
	matchingClient matchingservice.MatchingServiceClient,
	persistence p.NexusEndpointManager,
	logger log.Logger,
	metricsHandler metrics.Handler,
) *EndpointRegistryImpl

func (*EndpointRegistryImpl) GetByID

func (r *EndpointRegistryImpl) GetByID(ctx context.Context, id string) (*persistencespb.NexusEndpointEntry, error)

func (*EndpointRegistryImpl) GetByName

func (r *EndpointRegistryImpl) GetByName(ctx context.Context, _ namespace.ID, endpointName string) (*persistencespb.NexusEndpointEntry, error)

func (*EndpointRegistryImpl) StartLifecycle

func (r *EndpointRegistryImpl) StartLifecycle()

StartLifecycle starts this component. It should only be invoked by an fx lifecycle hook. Should not be called multiple times or concurrently with StopLifecycle()

func (*EndpointRegistryImpl) StopLifecycle

func (r *EndpointRegistryImpl) StopLifecycle()

StopLifecycle stops this component. It should only be invoked by an fx lifecycle hook. Should not be called multiple times or concurrently with StartLifecycle()

type NamespaceAndTaskQueue

type NamespaceAndTaskQueue struct {
	Namespace string
	TaskQueue string
}