gomdb

package module

Go to main page

Versions in this module

v0

v0.4.0

Dec 26, 2021

Changes in this version

+ func DynamicPolling(target float64, step, min, max time.Duration) func() PollingStrategy

type ClientOption

+ func WithDefaultPollingStrategy(strat func() PollingStrategy) ClientOption

type GetCategoryOption

+ func WithCategoryPollingStrategy(strat PollingStrategy) GetCategoryOption

type GetStreamOption

+ func WithStreamPollingStrategy(strat PollingStrategy) GetStreamOption

v0.3.0

May 22, 2021

v0.2.0

May 16, 2021

Changes in this version

+ const DefaultPollingInterval

+ var ErrInvalidConsumerGroupMember = errors.New("consumer group member must be >= 0 < group size")

+ var ErrInvalidConsumerGroupSize = errors.New("consumer group size must be 0 or greater (0 to disbale consumer groups)")

+ var ErrInvalidReadBatchSize = errors.New("batch size must be greater than 0")

+ var ErrInvalidReadPosition = errors.New("stream position cannot be less than 0")

+ var ErrInvalidReadStreamVersion = errors.New("stream version cannot be less than 0")

type Client

+ func (c *Client) SubscribeToCategory(ctx context.Context, category string, handleMessage MessageHandler, ...) error

+ func (c *Client) SubscribeToStream(ctx context.Context, stream StreamIdentifier, handleMessage MessageHandler, ...) error

+ type ClientOption func(*Client)

+ func WithSubPollingStrategy(strat PollingStrategy) ClientOption

+ type LivenessHandler func(bool)

+ type MessageHandler func(*Message)

+ type PollingStrategy func(retrieved, expected int64) time.Duration

+ func ConstantPolling(interval time.Duration) PollingStrategy

+ func ExpBackoffPolling(min, max time.Duration, multiplier float64) PollingStrategy

+ type SubDroppedHandler func(error)

v0.1.1

May 15, 2021

Changes in this version

+ const AnyVersion

+ const CorrelationKey

+ const GetCategoryMessagesSQL

+ const GetLastStreamMessageSQL

+ const GetStreamMessagesSQL

+ const GetStreamVersionSQL

+ const NoStreamVersion

+ const StreamNameSeparator

+ const WriteMessageSQL

+ var ErrInvalidCategory = fmt.Errorf("category cannot contain separator (%s)", StreamNameSeparator)

+ var ErrInvalidMessageID = errors.New("proposed message ID must be a valid UUID")

+ var ErrInvalidStreamID = fmt.Errorf("ID cannot contain separator (%s)", StreamNameSeparator)

+ var ErrMissingCategory = errors.New("category cannot be blank")

+ var ErrMissingData = errors.New("proposed message must include Data")

+ var ErrMissingStreamID = errors.New("ID cannot be blank")

+ var ErrMissingType = errors.New("proposed message must include Type")

+ var ErrUnexpectedStreamVersion = errors.New("unexpected stream version when writing message")

+ type Client struct

+ func NewClient(db *sql.DB) *Client

+ func (c *Client) GetCategoryMessages(ctx context.Context, category string, opts ...GetCategoryOption) ([]*Message, error)

+ func (c *Client) GetLastStreamMessage(ctx context.Context, stream StreamIdentifier) (*Message, error)

+ func (c *Client) GetStreamMessages(ctx context.Context, stream StreamIdentifier, opts ...GetStreamOption) ([]*Message, error)

+ func (c *Client) GetStreamVersion(ctx context.Context, stream StreamIdentifier) (int64, error)

+ func (c *Client) WriteMessage(ctx context.Context, stream StreamIdentifier, message ProposedMessage, ...) (int64, error)

+ type GetCategoryOption func(*categoryConfig)

+ func AsConsumerGroup(member, size int64) GetCategoryOption

+ func FromPosition(position int64) GetCategoryOption

+ func WithCategoryBatchSize(batchSize int64) GetCategoryOption

+ func WithCategoryCondition(condition string) GetCategoryOption

+ func WithCorrelation(correlation string) GetCategoryOption

+ type GetStreamOption func(*streamConfig)

+ func FromVersion(version int64) GetStreamOption

+ func WithStreamBatchSize(batchSize int64) GetStreamOption

+ func WithStreamCondition(condition string) GetStreamOption

+ type Message struct

+ GlobalPosition int64

+ ID uuid.UUID

+ Stream StreamIdentifier

+ Timestamp time.Time

+ Type string

+ Version int64

+ func (m *Message) UnmarshalData(i interface{}) error

+ func (m *Message) UnmarshalMetadata(i interface{}) error

+ type ProposedMessage struct

+ Data interface{}

+ ID string

+ Metadata interface{}

+ Type string

+ type StreamIdentifier struct

+ Category string

+ ID string

+ func (si StreamIdentifier) String() string

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL