operationsbus

package
v0.0.39 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Feb 8, 2025 License: MIT Imports: 13 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	Unknown     string = "Unknown"
	Pending            = "Pending"
	In_Progress        = "In_Progress"
	Completed          = "Completed"
	Failed             = "Failed"
	Cancelled          = "Cancelled"
)

All the status types that should be supported.

Variables

This section is empty.

Functions

func CreateProcessor 

func CreateProcessor(
	serviceBusReceiver sb.ReceiverInterface,
	matcher *Matcher,
	operationContainer oc.OperationContainerClient,
	entityController EntityController,
	logger *slog.Logger,
	customHandler shuttle.HandlerFunc,
	processorOptions *shuttle.ProcessorOptions,
	hooks []BaseOperationHooksInterface,
) (*shuttle.Processor, error)

The processor will be utilized to "process" all the operations by receiving the message, guarding against concurrency, running the operation, and updating the right database status.

func DefaultHandlers added in v0.0.31 

func DefaultHandlers(
	serviceBusReceiver sb.ReceiverInterface,
	matcher *Matcher,
	operationContainer oc.OperationContainerClient,
	entityController EntityController,
	logger *slog.Logger,
	hooks []BaseOperationHooksInterface,
) shuttle.HandlerFunc

func NewErrorHandler added in v0.0.30 

func NewErrorHandler(errHandler ErrorHandlerFunc, receiver sb.ReceiverInterface, next shuttle.HandlerFunc) shuttle.HandlerFunc

An error handler that continues the normal shuttle.HandlerFunc handler chain.

func NewLogHandler added in v0.0.30 

func NewLogHandler(logger *slog.Logger, next shuttle.HandlerFunc) shuttle.HandlerFunc

NewLogHandler creates a new log handler with the provided logger.

func NewQoSHandler added in v0.0.30 

func NewQoSHandler(logger *slog.Logger, next shuttle.HandlerFunc) shuttle.HandlerFunc

NewQoSHandler creates a new QoS handler with the provided logger.

func NewQosErrorHandler added in v0.0.30 

func NewQosErrorHandler(errHandler ErrorHandlerFunc) shuttle.HandlerFunc

A QoS handler that is able to log the errors as well.

Types

type ApiOperation added in v0.0.22 

type ApiOperation interface {
	InitOperation(context.Context, OperationRequest) (ApiOperation, error)
	GuardConcurrency(context.Context, Entity) *CategorizedError
	Run(context.Context) error
	GetOperationRequest() *OperationRequest
}

ApiOperation is the interface all operations will need to implement.

type BaseOperationHooksInterface added in v0.0.22 

type BaseOperationHooksInterface interface {
	BeforeInitOperation(ctx context.Context, req OperationRequest) error
	AfterInitOperation(ctx context.Context, op ApiOperation, req OperationRequest, err error) error

	BeforeGuardConcurrency(ctx context.Context, op ApiOperation, entity Entity) error
	AfterGuardConcurrency(ctx context.Context, op ApiOperation, ce *CategorizedError) error

	BeforeRun(ctx context.Context, op ApiOperation) error
	AfterRun(ctx context.Context, op ApiOperation, err error) error
}

type CategorizedError 

type CategorizedError struct {
	Message      string
	InnerMessage string
	ErrorCode    int
	Err          error
}

func NewCategorizedError 

func NewCategorizedError(message string, innerMessage string, errorCode int, err error) *CategorizedError

func (*CategorizedError) Error 

func (ce *CategorizedError) Error() string

type Entity 

type Entity interface {
	GetLatestOperationID() string
}

type EntityController added in v0.0.32 

type EntityController interface {
	GetEntity(context.Context, OperationRequest) (Entity, error)
}

type EntityFactoryFunc added in v0.0.37 

type EntityFactoryFunc func(string) Entity

type ErrorHandler added in v0.0.30 

type ErrorHandler interface {
	Handle(ctx context.Context, settler shuttle.MessageSettler, message *azservicebus.ReceivedMessage) error
}

ErrorHandler interface that returns an error. Required for any error handling and not depending on panics.

type ErrorHandlerFunc added in v0.0.30 

type ErrorHandlerFunc func(ctx context.Context, settler shuttle.MessageSettler, message *azservicebus.ReceivedMessage) error

func NewErrorReturnHandler added in v0.0.30 

func NewErrorReturnHandler(errHandler ErrorHandlerFunc, receiver sb.ReceiverInterface, next shuttle.HandlerFunc) ErrorHandlerFunc

An error handler that provides the error to the parent handler for logging.

func NewOperationContainerHandler added in v0.0.32 

func NewOperationContainerHandler(errHandler ErrorHandlerFunc, operationContainer oc.OperationContainerClient) ErrorHandlerFunc

Handler for when the user uses the OperationContainer

func OperationHandler added in v0.0.31 

func OperationHandler(matcher *Matcher, hooks []BaseOperationHooksInterface, entityController EntityController) ErrorHandlerFunc

func (ErrorHandlerFunc) Handle added in v0.0.30 

func (f ErrorHandlerFunc) Handle(ctx context.Context, settler shuttle.MessageSettler, message *azservicebus.ReceivedMessage) error

type HookedApiOperation added in v0.0.22 

type HookedApiOperation struct {
	Operation      ApiOperation
	OperationHooks []BaseOperationHooksInterface
}

func (*HookedApiOperation) AfterGuardConcurrency added in v0.0.22 

func (h *HookedApiOperation) AfterGuardConcurrency(ctx context.Context, op ApiOperation, ce *CategorizedError) error

func (*HookedApiOperation) AfterInitOperation added in v0.0.25 

func (h *HookedApiOperation) AfterInitOperation(ctx context.Context, op ApiOperation, req OperationRequest, err error) error

func (*HookedApiOperation) AfterRun added in v0.0.22 

func (h *HookedApiOperation) AfterRun(ctx context.Context, op ApiOperation, err error) error

func (*HookedApiOperation) BeforeGuardConcurrency added in v0.0.22 

func (h *HookedApiOperation) BeforeGuardConcurrency(ctx context.Context, op ApiOperation, entity Entity) error

func (*HookedApiOperation) BeforeInitOperation added in v0.0.25 

func (h *HookedApiOperation) BeforeInitOperation(ctx context.Context, req OperationRequest) error

func (*HookedApiOperation) BeforeRun added in v0.0.22 

func (h *HookedApiOperation) BeforeRun(ctx context.Context, op ApiOperation) error

func (*HookedApiOperation) GuardConcurrency added in v0.0.22 

func (h *HookedApiOperation) GuardConcurrency(ctx context.Context, entity Entity) *CategorizedError

func (*HookedApiOperation) InitOperation added in v0.0.25 

func (h *HookedApiOperation) InitOperation(ctx context.Context, opReq OperationRequest) (ApiOperation, error)

func (*HookedApiOperation) Run added in v0.0.22 

func (h *HookedApiOperation) Run(ctx context.Context) error

type Matcher 

type Matcher struct {
	Types          map[string]reflect.Type
	EntityCreators map[string]EntityFactoryFunc
}

The matcher is utilized in order to keep track of the name and type of each operation. This is required because we only send the OperationRequest through the service bus, but we utilize the name shown in that struct in order to create an instance of the right operation type (e.g. LongRunning) and Run with the correct logic.

func NewMatcher 

func NewMatcher() *Matcher

func (*Matcher) CreateEntityInstance added in v0.0.36 

func (m *Matcher) CreateEntityInstance(key string, lastOperationId string) (Entity, error)

func (*Matcher) CreateHookedInstace added in v0.0.22 

func (m *Matcher) CreateHookedInstace(key string, hooks []BaseOperationHooksInterface) (*HookedApiOperation, error)

func (*Matcher) CreateOperationInstance added in v0.0.36 

func (m *Matcher) CreateOperationInstance(key string) (ApiOperation, error)

This will create an empty instance of the type, with which you can then call op.Init() and initialize any info you need.

func (*Matcher) Get 

func (m *Matcher) Get(key string) (reflect.Type, bool)

Get retrieves a value from the map by its key

func (*Matcher) Register 

func (m *Matcher) Register(key string, value ApiOperation)

Set adds a key-value pair to the map Ex: matcher.Register("LongRunning", &LongRunning{})

func (*Matcher) RegisterEntity added in v0.0.36 

func (m *Matcher) RegisterEntity(key string, value EntityFactoryFunc)

Set adds a key-value pair to the map Ex: matcher.Register("LongRunning", &LongRunning{})

type NonRetryError added in v0.0.30 

type NonRetryError struct {
	Message string
}

func (*NonRetryError) Error added in v0.0.30 

func (e *NonRetryError) Error() string

type OperationRequest 

type OperationRequest struct {
	OperationName       string                 // Name of the operation being processed. Used to match the ApiOperation with the right implementation.
	APIVersion          string                 // Specifies the version of the API the operation is associated with, ensuring compatibility.
	RetryCount          int                    // Tracks the number of retries of the operation to prevent infinite looping or special logic around retries.
	OperationId         string                 // A unique identifier for the operation.
	EntityId            string                 // A unique identifier for the entity (resource) the operation is acting on, used with EntityType to ensure we have selected the right entity.
	EntityType          string                 // Specified the type of entity the operation is acting on, used with EntityId to ensure we have selected the right Entity.
	ExpirationTimestamp *timestamppb.Timestamp // Defines when the operation should expire and prevent execution, should it have passed this date.

	// HTTP
	Body       []byte // Contains request payload or data needed for the operation in HTTP operations.
	HttpMethod string // Indicated the GGPT method if the operation requires HTTP-based communication.

	Extension interface{} // An optional and flexible field to add any data the user may require.
}

All the fields that the operations might need. This struct will be part of every operation.

func NewOperationRequest added in v0.0.21 

func NewOperationRequest(
	operationName string,
	apiVersion string,
	operationId string,
	entityId string,
	entityType string,
	retryCount int,
	expirationTimestamp *timestamppb.Timestamp,
	body []byte,
	httpMethod string,
	extension interface{},
) *OperationRequest

func (*OperationRequest) Retry added in v0.0.8 

func (opRequest *OperationRequest) Retry(ctx context.Context, sender sb.SenderInterface) error

Generalized method to retry every operation. If the operation failed or hit an error at any stage, this method will be called after the panic is handled.

func (*OperationRequest) SetExtension added in v0.0.21 

func (opRequest *OperationRequest) SetExtension(newValue interface{}) error

SetExtension sets the Extension field to a new type and value, copying data if possible

type RetryError added in v0.0.30 

type RetryError struct {
	Message string
}

Default errors for the error handler.

func (*RetryError) Error added in v0.0.30 

func (e *RetryError) Error() string

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.