processor

package
v4.0.0 Latest Latest
Warning

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

 Go to latest Published: Aug 1, 2026 License: MIT Imports: 14 Imported by: 0

README

Processor

Overview

The processor package is a key component of the ETL project, providing a versatile and efficient mechanism for defining and executing data processing operations within a pipeline stage. Processors are responsible for transforming and enhancing the data flowing through the pipeline, allowing for complex data manipulations and business logic to be applied.

What's a Processor?

A processor is a self-contained unit of data transformation within a pipeline stage. It encapsulates a specific data processing operation, defined as a function that takes a context and an input slice of data, performs the necessary transformations, and returns the processed data along with any errors that occurred during processing.

Processors are highly modular and reusable, allowing them to be easily combined and composed within stages to create sophisticated data processing workflows.

How It Works

The core of the processor package revolves around the IProcessor interface, which defines the contract for a processor. The Processor struct implements this interface, providing the necessary functionality for creating and running processors.

To create a processor, you instantiate a new Processor using the New factory function, specifying the processor name, description, and the transform function. The transform function is defined using the Transform type, which takes a context and an input slice of data of type ProcessingData, and returns the processed data along with any errors.

When the Run method is called on a processor, it executes the following steps:

  1. It checks if the pipeline is paused. If paused, the processor waits until it is resumed or the context is done. This allows for graceful handling of pipeline pauses during processor execution.

  2. Once the pipeline is resumed or if it was not paused, the processor invokes the transform function, passing the input data and the context. The transform function performs the necessary data transformations and returns the processed data.

  3. If the transform function returns an error, the processor handles it gracefully, updating the relevant metrics and logging the error details.

  4. After successful execution of the transform function, the processor updates its metrics, such as incrementing the done counter and setting the duration.

  5. If an OnFinished callback function is provided, the processor invokes it, passing the original input data, the processed data, and the context. This allows for custom post-processing or logging after the processor finishes its execution.

  6. Finally, the processor returns the processed data to the caller.

Throughout the execution, the processor maintains comprehensive observability, including metrics, logging, and tracing, to monitor and debug the processor's performance and behavior.

Features

  1. Modularity and Reusability: Processors are designed to be modular and reusable, enabling easy composition and combination within pipeline stages to create complex data processing workflows.

  2. Flexibility: Processors can encapsulate any data transformation logic, from simple arithmetic operations to complex business rules and data enrichment.

  3. Pause and Resume: Processors support pipeline pausing and resuming. If the pipeline is paused during a processor's execution, the processor gracefully waits until it is resumed or the context is done, ensuring proper handling of pauses.

  4. Observability: The processor package provides comprehensive observability features, including metrics, logging, and tracing, to monitor and debug the processor's execution.

  5. Metrics: Processor metrics are exposed using the expvar package, allowing for easy integration with monitoring systems. Metrics include counters for created, running, failed, done, and interrupted processors, as well as duration.

  6. Logging: The package utilizes the sypl library for structured logging, providing rich context and consistent log levels throughout the codebase. Log messages include relevant information such as processor status, counters, and duration.

  7. Tracing: Tracing is implemented using the customapm package, which integrates with Elastic APM (Application Performance Monitoring) under the hood. This enables distributed tracing of the processor's execution, allowing developers to gain insights into the performance and behavior of their processors.

  8. Error Handling: The processor package includes robust error handling mechanisms, with detailed error messages and proper propagation of errors during processor execution.

  9. OnFinished Callback: Processors support an optional OnFinished callback function, which is invoked after the processor finishes its execution. This callback receives the original input data, the processed data, and the context, enabling custom post-processing or logging.

  10. Flexible Configuration: Processors can be configured with various options, such as the OnFinished callback, using a functional options pattern. This allows for easy customization of processor behavior without modifying the core processor struct.

  11. Thorough Testing: The codebase includes comprehensive unit tests, ensuring the reliability and correctness of the processor functionality. The tests cover various scenarios, including success cases, error handling, and pause/resume functionality.

  12. Well-Documented: The code is thoroughly documented, with clear comments explaining the purpose and functionality of each component. The package also includes usage examples and test cases.

  13. Idiomatic Go: The codebase follows idiomatic Go practices, leveraging the language's features and conventions for clean and efficient code.

  14. Typed Errors: The package utilizes typed errors, providing more context and facilitating error handling and debugging.

  15. Customizable: The processor package provides a high level of customization through the use of interfaces and generic types. Developers can easily create custom processors with specific transformation logic to meet their data processing requirements.

Documentation

Overview

Package processor wraps slice transforms with lifecycle metrics, tracing, logging, pause handling, and completion callbacks.

A processor honors a pipeline pause controller carried by its context before invoking its transform. A stage interprets the async flag, runs the processor in a goroutine, and does not forward its output. Transforms must not mutate their input slice or pointed-to values.

Index

Examples

Constants

View Source 
const Type = "processor"

Type identifies processor entities.

Variables

This section is empty.

Functions

This section is empty.

Types

type Func 

type Func[T any] func(p IProcessor[T]) IProcessor[T]

Func configures a processor and returns the implementation to retain. An option must return a usable implementation; returning one whose embedded interface is nil will panic when it is used, as it would anywhere in Go.

func WithAsync 

func WithAsync[T any](async bool) Func[T]

WithAsync returns a Func that controls goroutine execution by a stage.

WARN: The output of the processing will not be forwarded!

func WithOnFinished 

func WithOnFinished[T any](onFinished OnFinished[T]) Func[T]

WithOnFinished returns a Func that sets the completion callback.

type IProcessor 

type IProcessor[ProcessingData any] interface {
	shared.IMeta

	shared.IMetrics

	// GetOnFinished returns the `OnFinished` function.
	GetOnFinished() OnFinished[ProcessingData]

	// SetOnFinished sets the `OnFinished` function.
	SetOnFinished(onFinished OnFinished[ProcessingData])

	// GetCounterInterrupted returns the counter for paused runs interrupted by
	// context cancellation or deadline.
	GetCounterInterrupted() *expvar.Int

	// SetAsync controls whether a stage runs the processor in a goroutine. It
	// changes scheduling only; Run still records lifecycle status and counters.
	//
	// WARN: The output of the processing will not be forwarded!
	SetAsync(async bool)

	// GetAsync reports whether a stage should run the processor in a goroutine.
	GetAsync() bool

	// Run transforms processingData without mutating it or pointed-to values.
	// It records context termination while paused as an interruption and
	// transform errors as failures.
	Run(ctx context.Context, processingData []ProcessingData) (processedOut []ProcessingData, err error)
}

IProcessor defines processor lifecycle, metrics, configuration, and execution. Implementations must not mutate Run's input slice or the values it points to.

func New 

func New[ProcessingData any](
	name, description string,
	fn Transform[ProcessingData],
	opts ...Func[ProcessingData],
) (IProcessor[ProcessingData], error)

New creates a processor. It applies options in order, retains each returned implementation, and then validates the result. Nil options and results are rejected. An option must return a usable implementation; returning one whose embedded interface is nil will panic when it is used, as it would anywhere in Go. Structural validation runs before lifecycle metric getters; the final implementation must expose non-nil lifecycle metrics. The transform must not mutate its input slice or pointed-to values.

Example

Demonstrates a processor that transforms every item in a slice. 

double, err := New(
	"example-double",
	"doubles every number",
	func(_ context.Context, in []int) ([]int, error) {
		out := make([]int, len(in))
		for i, value := range in {
			out[i] = value * 2
		}

		return out, nil
	},
)
if err != nil {
	log.Fatalln(err)
}

out, err := double.Run(context.Background(), []int{1, 2, 3})
if err != nil {
	log.Fatalln(err)
}

fmt.Println(out)

Output:
[2 4 6]

type OnFinished 

type OnFinished[T any] func(ctx context.Context, p IProcessor[T], originalIn []T, processedOut []T)

OnFinished runs after a processor succeeds with its original input and processed output.

type Processor 

type Processor[ProcessingData any] struct {
	// Func transforms a slice without mutating its input or pointed-to values.
	Func Transform[ProcessingData] `json:"-" validate:"required"`

	// Logger records processor lifecycle events.
	Logger sypl.ISypl `json:"-" validate:"required"`

	// Name identifies the processor.
	Name string `json:"name" validate:"required"`

	// Description describes the processor.
	Description string `json:"description"`

	// Lifecycle metrics. CounterInterrupted counts paused runs whose wait ends
	// because their context is cancelled or reaches its deadline.
	CounterCreated     *expvar.Int `json:"counterCreated"`
	CounterDone        *expvar.Int `json:"counterDone"`
	CounterFailed      *expvar.Int `json:"counterFailed"`
	CounterInterrupted *expvar.Int `json:"counterInterrupted"`
	CounterRunning     *expvar.Int `json:"counterRunning"`

	CreatedAt time.Time      `json:"createdAt"`
	Duration  *expvar.Int    `json:"duration"`
	Status    *expvar.String `json:"status"`
	// contains filtered or unexported fields
}

Processor wraps a slice transform with lifecycle observability.

func (*Processor[ProcessingData]) GetAsync 

func (p *Processor[ProcessingData]) GetAsync() bool

GetAsync reports whether a stage should run the processor in a goroutine.

func (*Processor[ProcessingData]) GetCounterCreated 

func (p *Processor[ProcessingData]) GetCounterCreated() *expvar.Int

GetCounterCreated returns the `CounterCreated` metric.

func (*Processor[ProcessingData]) GetCounterDone 

func (p *Processor[ProcessingData]) GetCounterDone() *expvar.Int

GetCounterDone returns the `CounterDone` metric.

func (*Processor[ProcessingData]) GetCounterFailed 

func (p *Processor[ProcessingData]) GetCounterFailed() *expvar.Int

GetCounterFailed returns the `CounterFailed` metric.

func (*Processor[ProcessingData]) GetCounterInterrupted 

func (p *Processor[ProcessingData]) GetCounterInterrupted() *expvar.Int

GetCounterInterrupted returns the counter for paused runs interrupted by context cancellation or deadline.

func (*Processor[ProcessingData]) GetCounterRunning 

func (p *Processor[ProcessingData]) GetCounterRunning() *expvar.Int

GetCounterRunning returns the `CounterRunning` metric.

func (*Processor[ProcessingData]) GetCreatedAt 

func (p *Processor[ProcessingData]) GetCreatedAt() time.Time

GetCreatedAt returns when the processor was constructed.

func (*Processor[ProcessingData]) GetDescription 

func (p *Processor[ProcessingData]) GetDescription() string

GetDescription returns the `Description` of the processor.

func (*Processor[ProcessingData]) GetDuration 

func (p *Processor[ProcessingData]) GetDuration() *expvar.Int

GetDuration returns the most recent successful transform duration in milliseconds. A failed transform does not replace it.

func (*Processor[ProcessingData]) GetLogger 

func (p *Processor[ProcessingData]) GetLogger() sypl.ISypl

GetLogger returns the `Logger` of the processor.

func (*Processor[ProcessingData]) GetMetrics 

func (p *Processor[ProcessingData]) GetMetrics() map[string]string

GetMetrics returns selected processor metrics as strings.

func (*Processor[ProcessingData]) GetName 

func (p *Processor[ProcessingData]) GetName() string

GetName returns the `Name` of the processor.

func (*Processor[ProcessingData]) GetOnFinished 

func (p *Processor[ProcessingData]) GetOnFinished() OnFinished[ProcessingData]

GetOnFinished returns the `OnFinished` function.

func (*Processor[ProcessingData]) GetStatus 

func (p *Processor[ProcessingData]) GetStatus() *expvar.String

GetStatus returns the `Status` metric.

func (*Processor[ProcessingData]) GetType 

func (p *Processor[ProcessingData]) GetType() string

GetType returns the entity type.

func (*Processor[ProcessingData]) Run 

func (p *Processor[ProcessingData]) Run(ctx context.Context, processingData []ProcessingData) ([]ProcessingData, error)

Run waits if its context carries a paused pipeline controller, then invokes the transform and completion callback. A processor running without such a controller does not pause. Context termination while waiting is recorded as an interruption; transform errors are recorded as failures.

func (*Processor[ProcessingData]) SetAsync 

func (p *Processor[ProcessingData]) SetAsync(async bool)

SetAsync marks the processor for goroutine execution by a stage. Async scheduling does not change Run's lifecycle status or counters. Async processors receive a shallow copy, so transforms must not mutate input or pointed-to values.

WARN: The output of the processing will not be forwarded!

func (*Processor[ProcessingData]) SetOnFinished 

func (p *Processor[ProcessingData]) SetOnFinished(onFinished OnFinished[ProcessingData])

SetOnFinished sets the `OnFinished` function.

Example

Demonstrates setting a callback after constructing a processor. 

double, err := New(
	"example-double-callback",
	"doubles every number",
	func(_ context.Context, in []int) ([]int, error) {
		out := make([]int, len(in))
		for i, value := range in {
			out[i] = value * 2
		}

		return out, nil
	},
)
if err != nil {
	log.Fatalln(err)
}

double.SetOnFinished(func(_ context.Context, p IProcessor[int], in, out []int) {
	fmt.Printf("%s: %v -> %v\n", p.GetName(), in, out)
})

if _, err := double.Run(context.Background(), []int{2, 4}); err != nil {
	log.Fatalln(err)
}

Output:
example-double-callback: [2 4] -> [4 8]

type Transform 

type Transform[ProcessedData any] func(ctx context.Context, processingData []ProcessedData) (processedOut []ProcessedData, err error)

Transform transforms `processingData`, returning the processed data and any error. It must not mutate its input slice or pointed-to values; stages reuse results and async processors receive only a shallow copy.

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.
⌂ 换网址