messageprocessor

README

messageprocessor - An abstraction for consuming aws Kinesis messages

• messageprocessor - An abstraction for consuming aws Kinesis messages
  • Usage
  • Contributing
    • License
    • Contributing Agreement

Contributing

License

This project is licensed under Apache 2.0. See LICENSE.txt for details.

Contributing Agreement

Atlassian requires signing a contributor's agreement before we can accept a patch. If you are an individual you can fill out the individual CLA. If you are contributing on behalf of your company then please fill out the corporate CLA.

Documentation

Overview

Package messageprocessor is a generated GoMock package.

Index

• type Chain
  • func (c Chain) Apply(base MessageProcessor) MessageProcessor
• type Decorator
• type MaxRetriesExceededError
  • func (t MaxRetriesExceededError) Error() string
  • func (t MaxRetriesExceededError) IsRetryable() bool
  • func (t MaxRetriesExceededError) RetryAfter() int
• type MessageProcessor
• type MessageProcessorError
• type MockMessageProcessor
  • func NewMockMessageProcessor(ctrl *gomock.Controller) *MockMessageProcessor
  • func (m *MockMessageProcessor) EXPECT() *MockMessageProcessorMockRecorder
  • func (m *MockMessageProcessor) ProcessMessage(ctx context.Context, record *kinesis.Record) MessageProcessorError
• type MockMessageProcessorError
  • func NewMockMessageProcessorError(ctrl *gomock.Controller) *MockMessageProcessorError
  • func (m *MockMessageProcessorError) EXPECT() *MockMessageProcessorErrorMockRecorder
  • func (m *MockMessageProcessorError) Error() string
  • func (m *MockMessageProcessorError) IsRetryable() bool
  • func (m *MockMessageProcessorError) RetryAfter() int
• type MockMessageProcessorErrorMockRecorder
  • func (mr *MockMessageProcessorErrorMockRecorder) Error() *gomock.Call
  • func (mr *MockMessageProcessorErrorMockRecorder) IsRetryable() *gomock.Call
  • func (mr *MockMessageProcessorErrorMockRecorder) RetryAfter() *gomock.Call
• type MockMessageProcessorMockRecorder
  • func (mr *MockMessageProcessorMockRecorder) ProcessMessage(ctx, record interface{}) *gomock.Call
• type RetryableMessageProcessor
  • func (t *RetryableMessageProcessor) ProcessMessage(ctx context.Context, record *kinesis.Record) MessageProcessorError
• type RetryableMessageProcessorComponent
  • func NewComponent() *RetryableMessageProcessorComponent
  • func (c *RetryableMessageProcessorComponent) New(_ context.Context, conf *RetryableMessageProcessorConfig) (func(MessageProcessor) MessageProcessor, error)
  • func (*RetryableMessageProcessorComponent) Settings() *RetryableMessageProcessorConfig
• type RetryableMessageProcessorConfig
  • func (*RetryableMessageProcessorConfig) Name() string

Types

type Chain

type Chain []Decorator

Chain is an ordered collection of Decorators.

func (Chain) Apply

func (c Chain) Apply(base MessageProcessor) MessageProcessor

Apply wraps the given RoundTripper with the Decorator chain.

type Decorator

type Decorator func(MessageProcessor) MessageProcessor

Decorator is a named type for any function that takes a RoundTripper and returns a RoundTripper.

type MaxRetriesExceededError

type MaxRetriesExceededError struct {
	Retryable bool
	OrigErr   error
	Wait      int
}

MaxRetriesExceededError implements MessageProcessorError and is used to indicate to upstream application/ decorators that retries have been attempted and exhausted

func (MaxRetriesExceededError) Error

func (t MaxRetriesExceededError) Error() string

func (MaxRetriesExceededError) IsRetryable

func (t MaxRetriesExceededError) IsRetryable() bool

IsRetryable indicates whether or not the JiraClient Error that was returned should be retried

func (MaxRetriesExceededError) RetryAfter

func (t MaxRetriesExceededError) RetryAfter() int

RetryAfter is not relevant in context of MaxRetriesExceededError as IsRetryable is set to false

type MessageProcessor

type MessageProcessor interface {
	ProcessMessage(ctx context.Context, record *kinesis.Record) MessageProcessorError
}

MessageProcessor processes a consumed message. Implementors are responsible for unmarshalling the data

type MessageProcessorError

type MessageProcessorError interface {
	IsRetryable() bool
	Error() string
	RetryAfter() int
}

MessageProcessorError represents an error that can be used to indicate to the consumer that an error should be retried

type MockMessageProcessor

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

MockMessageProcessor is a mock of MessageProcessor interface.

func NewMockMessageProcessor

func NewMockMessageProcessor(ctrl *gomock.Controller) *MockMessageProcessor

NewMockMessageProcessor creates a new mock instance.

func (*MockMessageProcessor) EXPECT

func (m *MockMessageProcessor) EXPECT() *MockMessageProcessorMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

func (*MockMessageProcessor) ProcessMessage

func (m *MockMessageProcessor) ProcessMessage(ctx context.Context, record *kinesis.Record) MessageProcessorError

ProcessMessage mocks base method.

type MockMessageProcessorError

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

MockMessageProcessorError is a mock of MessageProcessorError interface.

func NewMockMessageProcessorError

func NewMockMessageProcessorError(ctrl *gomock.Controller) *MockMessageProcessorError

NewMockMessageProcessorError creates a new mock instance.

func (*MockMessageProcessorError) EXPECT

func (m *MockMessageProcessorError) EXPECT() *MockMessageProcessorErrorMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

func (*MockMessageProcessorError) Error

func (m *MockMessageProcessorError) Error() string

Error mocks base method.

func (*MockMessageProcessorError) IsRetryable

func (m *MockMessageProcessorError) IsRetryable() bool

IsRetryable mocks base method.

func (*MockMessageProcessorError) RetryAfter

func (m *MockMessageProcessorError) RetryAfter() int

RetryAfter mocks base method.

type MockMessageProcessorErrorMockRecorder

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

MockMessageProcessorErrorMockRecorder is the mock recorder for MockMessageProcessorError.

func (*MockMessageProcessorErrorMockRecorder) Error

func (mr *MockMessageProcessorErrorMockRecorder) Error() *gomock.Call

Error indicates an expected call of Error.

func (*MockMessageProcessorErrorMockRecorder) IsRetryable

func (mr *MockMessageProcessorErrorMockRecorder) IsRetryable() *gomock.Call

IsRetryable indicates an expected call of IsRetryable.

func (*MockMessageProcessorErrorMockRecorder) RetryAfter

func (mr *MockMessageProcessorErrorMockRecorder) RetryAfter() *gomock.Call

RetryAfter indicates an expected call of RetryAfter.

type MockMessageProcessorMockRecorder

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

MockMessageProcessorMockRecorder is the mock recorder for MockMessageProcessor.

func (*MockMessageProcessorMockRecorder) ProcessMessage

func (mr *MockMessageProcessorMockRecorder) ProcessMessage(ctx, record interface{}) *gomock.Call

ProcessMessage indicates an expected call of ProcessMessage.

type RetryableMessageProcessor

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

RetryableMessageProcessor is a `MessageProcessor` decorator that re-attempts processing of messages 'maxAttempts' number of times in case of failures 'maxAttempts' is a configurable parameter which can be set by consumer of this lib Exponential backoff has been implemented as a retry mechanism

func (*RetryableMessageProcessor) ProcessMessage

func (t *RetryableMessageProcessor) ProcessMessage(ctx context.Context, record *kinesis.Record) MessageProcessorError

ProcessMessage invokes the wrapped `MessageProcessor`. Attempts retries using exponential backoff if underlying 'MessageProcessor' returns an error. If 'maxAttempts' are exceeded without successful processing, it emits a stat indicating the same

type RetryableMessageProcessorComponent

type RetryableMessageProcessorComponent struct{}

RetryableMessageProcessorComponent implements the settings.Component interface.

func NewComponent

func NewComponent() *RetryableMessageProcessorComponent

NewComponent populates default values.

func (*RetryableMessageProcessorComponent) New

func (c *RetryableMessageProcessorComponent) New(_ context.Context, conf *RetryableMessageProcessorConfig) (func(MessageProcessor) MessageProcessor, error)

func (*RetryableMessageProcessorComponent) Settings

func (*RetryableMessageProcessorComponent) Settings() *RetryableMessageProcessorConfig

Settings generates a config populated with defaults.

type RetryableMessageProcessorConfig

type RetryableMessageProcessorConfig struct {
	MaxAttempts int `description:"Maximum number of attempts to process kinesis message"`
}

RetryableMessageProcessorConfig is the config for creating a RetryableMessageProcessor

func (*RetryableMessageProcessorConfig) Name

func (*RetryableMessageProcessorConfig) Name() string

Name of the config root.