gopp

README

gopp

This library provides parallel processing functionality with configurable retry mechanism and error handling.

Features

• Ability to configure the retry strategy (e.g. based on constant interval or exponential backoff) and implement additional strategies easily
• Ability to configure the maximum amount of retries, no matter which retry strategy is used
• Ability to route failed submissions to the caller thread for further processing that the user may want to do

Usage

There are a couple of examples provided in the examples folder that can help understand how to use that library.

Default Behavior

By default, the parallel processor will retry the submissions immediately and up to 3 times. If there is not a successful execution, it will simply discard the failed submission.

Documentation

Index

• type ParallelProcessor
  • func NewParallelProcessor[T any](options ParallelProcessorOptions, processFunc func(item *T) error) *ParallelProcessor[T]
  • func (p *ParallelProcessor[T]) GetFailedSubmission() (T, error)
  • func (p *ParallelProcessor[T]) Start() error
  • func (p *ParallelProcessor[T]) Submit(item *T) error
  • func (p *ParallelProcessor[T]) Terminate() error
• type ParallelProcessorOptions
  • func NewParallelProcessorOptionsWithDefaults() ParallelProcessorOptions
  • func (p *ParallelProcessorOptions) SetBufferSize(bufferSize int) error
  • func (p *ParallelProcessorOptions) SetDiscardFailures(discardFailures bool)
  • func (p *ParallelProcessorOptions) SetMaxRetries(maxRetries int) error
  • func (p *ParallelProcessorOptions) SetNumOfWorkers(numOfWorkers int) error
  • func (p *ParallelProcessorOptions) SetRetryStrategy(rs RetryStrategy)
• type RetryExponentialBackoff
  • func NewRetryExponentialBackoff() *RetryExponentialBackoff
• type RetryOnInterval
  • func NewRetryOnInterval(interval time.Duration) *RetryOnInterval
• type RetryStrategy

Types

type ParallelProcessor

type ParallelProcessor[T any] struct {
	// contains filtered or unexported fields
}

ParallelProcessor enables parallel processing of items given a specific function.

func NewParallelProcessor

func NewParallelProcessor[T any](options ParallelProcessorOptions, processFunc func(item *T) error) *ParallelProcessor[T]

func (*ParallelProcessor[T]) GetFailedSubmission

func (p *ParallelProcessor[T]) GetFailedSubmission() (T, error)

func (*ParallelProcessor[T]) Start

func (p *ParallelProcessor[T]) Start() error

func (*ParallelProcessor[T]) Submit

func (p *ParallelProcessor[T]) Submit(item *T) error

func (*ParallelProcessor[T]) Terminate

func (p *ParallelProcessor[T]) Terminate() error

type ParallelProcessorOptions

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

func NewParallelProcessorOptionsWithDefaults

func NewParallelProcessorOptionsWithDefaults() ParallelProcessorOptions

func (*ParallelProcessorOptions) SetBufferSize

func (p *ParallelProcessorOptions) SetBufferSize(bufferSize int) error

func (*ParallelProcessorOptions) SetDiscardFailures

func (p *ParallelProcessorOptions) SetDiscardFailures(discardFailures bool)

func (*ParallelProcessorOptions) SetMaxRetries

func (p *ParallelProcessorOptions) SetMaxRetries(maxRetries int) error

func (*ParallelProcessorOptions) SetNumOfWorkers

func (p *ParallelProcessorOptions) SetNumOfWorkers(numOfWorkers int) error

func (*ParallelProcessorOptions) SetRetryStrategy

func (p *ParallelProcessorOptions) SetRetryStrategy(rs RetryStrategy)

type RetryExponentialBackoff

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

func NewRetryExponentialBackoff

func NewRetryExponentialBackoff() *RetryExponentialBackoff

type RetryOnInterval

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

func NewRetryOnInterval

func NewRetryOnInterval(interval time.Duration) *RetryOnInterval

type RetryStrategy

type RetryStrategy interface {
	// contains filtered or unexported methods
}