runtime

Documentation

Index

• Constants
• func NewFunctionPipeline(id string, topics []string, transforms []interfaces.AppFunction) interfaces.FunctionPipeline
• type FunctionsPipelineRuntime
  • func NewFunctionPipelineRuntime(serviceKey string, targetType interface{}, dic *di.Container) *FunctionsPipelineRuntime
  • func (fpr *FunctionsPipelineRuntime) AddFunctionsPipeline(id string, topics []string, transforms []interfaces.AppFunction) error
  • func (fpr *FunctionsPipelineRuntime) ClearAllFunctionsPipelineTransforms()
  • func (fpr *FunctionsPipelineRuntime) DecodeMessage(appContext *appfunction.Context, envelope types.MessageEnvelope) (interface{}, *MessageError, bool)
  • func (fpr *FunctionsPipelineRuntime) ExecutePipeline(target interface{}, appContext *appfunction.Context, ...) *MessageError
  • func (fpr *FunctionsPipelineRuntime) GetDefaultPipeline() *interfaces.FunctionPipeline
  • func (fpr *FunctionsPipelineRuntime) GetMatchingPipelines(incomingTopic string) []*interfaces.FunctionPipeline
  • func (fpr *FunctionsPipelineRuntime) GetPipelineById(id string) *interfaces.FunctionPipeline
  • func (fpr *FunctionsPipelineRuntime) ProcessMessage(appContext *appfunction.Context, target interface{}, ...) *MessageError
  • func (fpr *FunctionsPipelineRuntime) RemoveAllFunctionPipelines()
  • func (fpr *FunctionsPipelineRuntime) SetDefaultFunctionsPipeline(transforms []interfaces.AppFunction)
  • func (fpr *FunctionsPipelineRuntime) SetFunctionsPipelineTopics(id string, topics []string)
  • func (fpr *FunctionsPipelineRuntime) SetFunctionsPipelineTransforms(id string, transforms []interfaces.AppFunction)
  • func (fpr *FunctionsPipelineRuntime) StartStoreAndForward(appWg *sync.WaitGroup, appCtx context.Context, enabledWg *sync.WaitGroup, ...)
• type MessageError
  • func (merr *MessageError) Error() string

Constants

const (
	TopicWildCard            = "#"
	TopicSingleLevelWildcard = "+"
	TopicLevelSeparator      = "/"
)

Functions

func NewFunctionPipeline

func NewFunctionPipeline(id string, topics []string, transforms []interfaces.AppFunction) interfaces.FunctionPipeline

Types

type FunctionsPipelineRuntime

type FunctionsPipelineRuntime struct {
	TargetType interface{}
	ServiceKey string
	// contains filtered or unexported fields
}

FunctionsPipelineRuntime represents the golang runtime environment for App Services' Functions Pipelines

func NewFunctionPipelineRuntime

func NewFunctionPipelineRuntime(serviceKey string, targetType interface{}, dic *di.Container) *FunctionsPipelineRuntime

NewFunctionPipelineRuntime creates and initializes the AppServiceRuntime instance

func (*FunctionsPipelineRuntime) AddFunctionsPipeline

func (fpr *FunctionsPipelineRuntime) AddFunctionsPipeline(id string, topics []string, transforms []interfaces.AppFunction) error

AddFunctionsPipeline is thread safe to set transforms

func (*FunctionsPipelineRuntime) ClearAllFunctionsPipelineTransforms

func (fpr *FunctionsPipelineRuntime) ClearAllFunctionsPipelineTransforms()

ClearAllFunctionsPipelineTransforms clears the transforms for all existing function pipelines.

func (*FunctionsPipelineRuntime) DecodeMessage

func (fpr *FunctionsPipelineRuntime) DecodeMessage(appContext *appfunction.Context, envelope types.MessageEnvelope) (interface{}, *MessageError, bool)

DecodeMessage decode the message wrapped in the MessageEnvelope and return the data to be processed.

func (*FunctionsPipelineRuntime) ExecutePipeline

func (fpr *FunctionsPipelineRuntime) ExecutePipeline(
	target interface{},
	appContext *appfunction.Context,
	pipeline *interfaces.FunctionPipeline,
	startPosition int,
	isRetry bool) *MessageError

func (*FunctionsPipelineRuntime) GetDefaultPipeline

func (fpr *FunctionsPipelineRuntime) GetDefaultPipeline() *interfaces.FunctionPipeline

func (*FunctionsPipelineRuntime) GetMatchingPipelines

func (fpr *FunctionsPipelineRuntime) GetMatchingPipelines(incomingTopic string) []*interfaces.FunctionPipeline

func (*FunctionsPipelineRuntime) GetPipelineById

func (fpr *FunctionsPipelineRuntime) GetPipelineById(id string) *interfaces.FunctionPipeline

func (*FunctionsPipelineRuntime) ProcessMessage

func (fpr *FunctionsPipelineRuntime) ProcessMessage(appContext *appfunction.Context, target interface{}, pipeline *interfaces.FunctionPipeline) *MessageError

ProcessMessage sends the contents of the message through the functions pipeline

func (*FunctionsPipelineRuntime) RemoveAllFunctionPipelines

func (fpr *FunctionsPipelineRuntime) RemoveAllFunctionPipelines()

RemoveAllFunctionPipelines removes all existing function pipelines

func (*FunctionsPipelineRuntime) SetDefaultFunctionsPipeline

func (fpr *FunctionsPipelineRuntime) SetDefaultFunctionsPipeline(transforms []interfaces.AppFunction)

SetDefaultFunctionsPipeline sets the default function pipeline

func (*FunctionsPipelineRuntime) SetFunctionsPipelineTopics

func (fpr *FunctionsPipelineRuntime) SetFunctionsPipelineTopics(id string, topics []string)

SetFunctionsPipelineTopics sets the topics for an existing function pipeline. Non-existent pipelines are ignored

func (*FunctionsPipelineRuntime) SetFunctionsPipelineTransforms

func (fpr *FunctionsPipelineRuntime) SetFunctionsPipelineTransforms(id string, transforms []interfaces.AppFunction)

SetFunctionsPipelineTransforms sets the transforms for an existing function pipeline. Non-existent pipelines are ignored

func (*FunctionsPipelineRuntime) StartStoreAndForward

func (fpr *FunctionsPipelineRuntime) StartStoreAndForward(
	appWg *sync.WaitGroup,
	appCtx context.Context,
	enabledWg *sync.WaitGroup,
	enabledCtx context.Context,
	serviceKey string)

type MessageError

type MessageError struct {
	Err       error
	ErrorCode int
}

func (*MessageError) Error

func (merr *MessageError) Error() string