backendutils

Documentation

Index

• Constants
• func RetryWithBackoff(...) error
• func StartSinkWorker(input <-chan []envelope.Envelope, shutdown <-chan int, sink Sink) error
• type Sink
• type SinkMetadata
  • func NewSinkMetadataFromConfig(conf config.Sink) SinkMetadata

Constants

const (
	DEFAULT_SINK_TIMEOUT_SECONDS int           = 15
	DEFAULT_RETRY_COUNT          int           = 15              // A sane default
	DEFAULT_RETRY_DELAY          time.Duration = 5 * time.Second // A sane default
)

Functions

func RetryWithBackoff

func RetryWithBackoff(f func(ctx context.Context, envelopes []envelope.Envelope, output string) error, ctx context.Context, envelopes []envelope.Envelope, output string) error

RetryWithBackoff retries the provided function up to N times on an exponential backoff NOTE - this function considers all errors as retriable. They are not so.

func StartSinkWorker

func StartSinkWorker(input <-chan []envelope.Envelope, shutdown <-chan int, sink Sink) error

Each sink runs an associated worker goroutine, which is responsible for dequeuing envelopes.

Types

type Sink

type Sink interface {
	Metadata() SinkMetadata
	Initialize(conf config.Sink) error
	StartWorker() error
	Enqueue(envelopes []envelope.Envelope) error
	Dequeue(ctx context.Context, envelopes []envelope.Envelope, output string) error
	Shutdown() error
}

type SinkMetadata

type SinkMetadata struct {
	Id               uuid.UUID `json:"id"`
	SinkType         string    `json:"sinkType"`
	Name             string    `json:"name"`
	DeliveryRequired bool      `json:"deliveryRequired"`
	DefaultOutput    string    `json:"defaultOutput"`
	DeadletterOutput string    `json:"deadletterOutput"`
}

func NewSinkMetadataFromConfig

func NewSinkMetadataFromConfig(conf config.Sink) SinkMetadata