sql

Nov 18, 2022

Changes in this version

+ type Beginner interface

+ BeginTx func(context.Context, *sql.TxOptions) (*sql.Tx, error)

+ type ContextExecutor interface

+ ExecContext func(ctx context.Context, query string, args ...interface{}) (sql.Result, error)

+ QueryContext func(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)

+ QueryRowContext func(ctx context.Context, query string, args ...interface{}) *sql.Row

+ type Executor interface

+ Exec func(query string, args ...interface{}) (sql.Result, error)

+ Query func(query string, args ...interface{}) (*sql.Rows, error)

+ QueryRow func(query string, args ...interface{}) *sql.Row

Apr 5, 2021

Changes in this version

+ func TxFromContext(ctx context.Context) (*sql.Tx, bool)

Nov 29, 2020

Changes in this version

+ type BackoffManager interface

+ HandleError func(logger watermill.LoggerAdapter, noMsg bool, err error) time.Duration

+ func NewDefaultBackoffManager(pollInterval, retryInterval time.Duration) BackoffManager

type SubscriberConfig

+ BackoffManager BackoffManager

Oct 3, 2019

Changes in this version

+ type DefaultMySQLSchema struct

+ GenerateMessagesTableName func(topic string) string

+ func (s DefaultMySQLSchema) InsertQuery(topic string, msgs message.Messages) (string, []interface{}, error)

+ func (s DefaultMySQLSchema) MessagesTable(topic string) string

+ func (s DefaultMySQLSchema) SchemaInitializingQueries(topic string) []string

+ func (s DefaultMySQLSchema) SelectQuery(topic string, consumerGroup string, offsetsAdapter OffsetsAdapter) (string, []interface{})

+ func (s DefaultMySQLSchema) UnmarshalMessage(row *sql.Row) (offset int, msg *message.Message, err error)

+ type DefaultPostgreSQLOffsetsAdapter struct

+ GenerateMessagesOffsetsTableName func(topic string) string

+ func (a DefaultPostgreSQLOffsetsAdapter) AckMessageQuery(topic string, offset int, consumerGroup string) (string, []interface{})

+ func (a DefaultPostgreSQLOffsetsAdapter) ConsumedMessageQuery(topic string, offset int, consumerGroup string, consumerULID []byte) (string, []interface{})

+ func (a DefaultPostgreSQLOffsetsAdapter) MessagesOffsetsTable(topic string) string

+ func (a DefaultPostgreSQLOffsetsAdapter) NextOffsetQuery(topic, consumerGroup string) (string, []interface{})

+ func (a DefaultPostgreSQLOffsetsAdapter) SchemaInitializingQueries(topic string) []string

+ type DefaultPostgreSQLSchema struct

+ GenerateMessagesTableName func(topic string) string

+ func (s DefaultPostgreSQLSchema) InsertQuery(topic string, msgs message.Messages) (string, []interface{}, error)

+ func (s DefaultPostgreSQLSchema) MessagesTable(topic string) string

+ func (s DefaultPostgreSQLSchema) SchemaInitializingQueries(topic string) []string

+ func (s DefaultPostgreSQLSchema) SelectQuery(topic string, consumerGroup string, offsetsAdapter OffsetsAdapter) (string, []interface{})

+ func (s DefaultPostgreSQLSchema) UnmarshalMessage(row *sql.Row) (offset int, msg *message.Message, err error)

Aug 7, 2019

Changes in this version

+ var ErrInvalidTopicName = errors.New(...)

+ var ErrPublisherClosed = errors.New("publisher is closed")

+ var ErrSubscriberClosed = errors.New("subscriber is closed")

+ type DefaultMySQLOffsetsAdapter struct

+ GenerateMessagesOffsetsTableName func(topic string) string

+ func (s DefaultMySQLOffsetsAdapter) AckMessageQuery(topic string, offset int, consumerGroup string) (string, []interface{})

+ func (s DefaultMySQLOffsetsAdapter) ConsumedMessageQuery(topic string, offset int, consumerGroup string, consumerULID []byte) (string, []interface{})

+ func (s DefaultMySQLOffsetsAdapter) MessagesOffsetsTable(topic string) string

+ func (s DefaultMySQLOffsetsAdapter) NextOffsetQuery(topic, consumerGroup string) (string, []interface{})

+ func (s DefaultMySQLOffsetsAdapter) SchemaInitializingQueries(topic string) []string

+ type DefaultSchema struct

+ GenerateMessagesTableName func(topic string) string

+ func (s DefaultSchema) InsertQuery(topic string, msgs message.Messages) (string, []interface{}, error)

+ func (s DefaultSchema) MessagesTable(topic string) string

+ func (s DefaultSchema) SchemaInitializingQueries(topic string) []string

+ func (s DefaultSchema) SelectQuery(topic string, consumerGroup string, offsetsAdapter OffsetsAdapter) (string, []interface{})

+ func (s DefaultSchema) UnmarshalMessage(row *sql.Row) (offset int, msg *message.Message, err error)

+ type OffsetsAdapter interface

+ AckMessageQuery func(topic string, offset int, consumerGroup string) (string, []interface{})

+ ConsumedMessageQuery func(topic string, offset int, consumerGroup string, consumerULID []byte) (string, []interface{})

+ NextOffsetQuery func(topic, consumerGroup string) (string, []interface{})

+ SchemaInitializingQueries func(topic string) []string

+ type Publisher struct

+ func NewPublisher(db db, config PublisherConfig, logger watermill.LoggerAdapter) (*Publisher, error)

+ func (p *Publisher) Close() error

+ func (p *Publisher) Publish(topic string, messages ...*message.Message) (err error)

+ type PublisherConfig struct

+ AutoInitializeSchema bool

+ SchemaAdapter SchemaAdapter

+ type SchemaAdapter interface

+ InsertQuery func(topic string, msgs message.Messages) (string, []interface{}, error)

+ SchemaInitializingQueries func(topic string) []string

+ SelectQuery func(topic string, consumerGroup string, offsetsAdapter OffsetsAdapter) (string, []interface{})

+ UnmarshalMessage func(row *sql.Row) (offset int, msg *message.Message, err error)

+ type Subscriber struct

+ func NewSubscriber(db *sql.DB, config SubscriberConfig, logger watermill.LoggerAdapter) (*Subscriber, error)

+ func (s *Subscriber) Close() error

+ func (s *Subscriber) Subscribe(ctx context.Context, topic string) (o <-chan *message.Message, err error)

+ func (s *Subscriber) SubscribeInitialize(topic string) error

+ type SubscriberConfig struct

+ ConsumerGroup string

+ InitializeSchema bool

+ OffsetsAdapter OffsetsAdapter

+ PollInterval time.Duration

+ ResendInterval time.Duration

+ RetryInterval time.Duration

+ SchemaAdapter SchemaAdapter