pipeline

Documentation

Index

• type Config
• type Message
  • func NewPipelineMessage(messageId string, payloadBody []byte, payloadAttrs map[string]string, ...) Message
  • func (m Message) AddAttribute(key string, value interface{}) (Message, error)
  • func (m Message) AddPayloadAttribute(key, value string) (Message, error)
  • func (m Message) Copy() Message
  • func (m Message) GetMessageAttributes() map[string]interface{}
  • func (m Message) GetMsgRefId() string
  • func (m Message) GetPayload() []byte
  • func (m Message) GetPayloadAttributes() map[string]string
  • func (m Message) UpsertPayloadAttribute(key, value string) (Message, error)
• type NamedStage
  • func (s NamedStage) Process(ctx context.Context, msg Message) (Message, error)
• type Orchestrator
  • func NewOrchestrator(pipelines map[string]Config) *Orchestrator
  • func (o *Orchestrator) Execute(cancelCtx context.Context, wg *sync.WaitGroup)
• type Pipeline
  • func NewPipeline(name string, stages []Stage) *Pipeline
  • func (p Pipeline) Process(ctx context.Context, msg Message) (Message, error)
• type Stage

Types

type Config

type Config struct {
	Pipeline   *Pipeline
	InputChan  chan Message
	OutputChan chan Message
	NumWorkers int
}

Config holds the configuration for each pipeline, including its input and output channels, and the number of workers.

type Message

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

A Message is a wrapper for a payload (consisting of a body and attributes) and internal message attributes. An instance of PipelineMessage can be routed through a pipeline.

func NewPipelineMessage

func NewPipelineMessage(messageId string, payloadBody []byte, payloadAttrs map[string]string, msgAttrs map[string]interface{}) Message

NewPipelineMessage creates a new PipelineMessage.

func (Message) AddAttribute

func (m Message) AddAttribute(key string, value interface{}) (Message, error)

AddAttribute adds an internal message attribute to a copy of the original message to keep the original instance unchanged. It will throw an error if the attribute key already exists.

func (Message) AddPayloadAttribute

func (m Message) AddPayloadAttribute(key, value string) (Message, error)

AddPayloadAttribute adds a message payload attribute to a copy of the original message to keep the original instance unchanged. It will throw an error if the attribute key already exists.

func (Message) Copy

func (m Message) Copy() Message

Copy copies a message and its contents.

func (Message) GetMessageAttributes

func (m Message) GetMessageAttributes() map[string]interface{}

GetMessageAttributes returns a copy of the internal message attributes.

func (Message) GetMsgRefId

func (m Message) GetMsgRefId() string

func (Message) GetPayload

func (m Message) GetPayload() []byte

GetPayload returns a copy of the message payload.

func (Message) GetPayloadAttributes

func (m Message) GetPayloadAttributes() map[string]string

GetPayloadAttributes returns a copy of the payload attributes.

func (Message) UpsertPayloadAttribute

func (m Message) UpsertPayloadAttribute(key, value string) (Message, error)

UpsertPayloadAttribute adds or updates a message payload attribute in a copy of the original message to keep the original instance unchanged.

type NamedStage

type NamedStage struct {
	Name string
	Stage
}

func (NamedStage) Process

func (s NamedStage) Process(ctx context.Context, msg Message) (Message, error)

type Orchestrator

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

Orchestrator orchestrates multiple pipelines, each with its own input and output channels and a configurable number of workers.

func NewOrchestrator

func NewOrchestrator(pipelines map[string]Config) *Orchestrator

NewOrchestrator creates a new Orchestrator.

Parameters: - pipelines: A map where each key is a pipeline identifier and the value is the Config.

Returns: - A new instance of Orchestrator.

func (*Orchestrator) Execute

func (o *Orchestrator) Execute(cancelCtx context.Context, wg *sync.WaitGroup)

Execute runs all pipelines, each with its configured number of workers.

This method starts multiple goroutines for each pipeline based on its configuration. Each goroutine listens for messages on its input channel, processes them through the pipeline, and sends the results to the output channel.

Parameters: - cancelCtx: A context that can be used to cancel the processing. - wg: A wait group to wait for all processing goroutines to complete.

type Pipeline

type Pipeline struct {
	Name string
	// contains filtered or unexported fields
}

A Pipeline helps orchestrate multiple pipeline steps.

func NewPipeline

func NewPipeline(name string, stages []Stage) *Pipeline

NewPipeline creates a new pipeline including the stages as configured.

func (Pipeline) Process

func (p Pipeline) Process(ctx context.Context, msg Message) (Message, error)

Process pipes an incoming message through the pipeline. Parameters:

• ctx (context.Context): Processing context. Used for tracing.
• msg (messaging.Message): Message to process.

Returns:

• messaging.Message: Processed message. Since incoming messages are immutable, this is an updated copy.
• error: If any error occurs during processing, this will not be nil.

type Stage

type Stage interface {
	// Process processes an incoming message and returns a modified copy and/or error once done.
	Process(context.Context, Message) (Message, error)
}

Stage describes a pipeline stage a message can be routed through. Depending on the implementation, it may return the same message or a modified copy.