sql

package

Go to main page

Versions in this module

v4

v4.1.2

Oct 28, 2025

v4.1.1

Oct 23, 2025

Changes in this version

type DelayedRequeuer

+ func NewMySQLDelayedRequeuer(config DelayedRequeuerConfig) (*DelayedRequeuer, error)

v4.1.0

Oct 22, 2025

Changes in this version

+ func NewDelayedMySQLPublisher(db ContextExecutor, config DelayedMySQLPublisherConfig) (message.Publisher, error)

+ func NewDelayedMySQLSubscriber(db Beginner, config DelayedMySQLSubscriberConfig) (message.Subscriber, error)

+ type DelayedMySQLPublisherConfig struct

+ DelayPublisherConfig delay.PublisherConfig

+ Logger watermill.LoggerAdapter

+ OverridePublisherConfig func(config *PublisherConfig) error

+ type DelayedMySQLSubscriberConfig struct

+ AllowNoDelay bool

+ DeleteOnAck bool

+ Logger watermill.LoggerAdapter

+ OverrideSubscriberConfig func(config *SubscriberConfig) error

+ type MySQLQueueOffsetsAdapter struct

+ DeleteOnAck bool

+ GenerateMessagesTableName func(topic string) string

+ func (a MySQLQueueOffsetsAdapter) AckMessageQuery(params AckMessageQueryParams) (Query, error)

+ func (a MySQLQueueOffsetsAdapter) BeforeSubscribingQueries(params BeforeSubscribingQueriesParams) ([]Query, error)

+ func (a MySQLQueueOffsetsAdapter) ConsumedMessageQuery(params ConsumedMessageQueryParams) (Query, error)

+ func (a MySQLQueueOffsetsAdapter) MessagesTable(topic string) string

+ func (a MySQLQueueOffsetsAdapter) NextOffsetQuery(params NextOffsetQueryParams) (Query, error)

+ func (a MySQLQueueOffsetsAdapter) SchemaInitializingQueries(params OffsetsSchemaInitializingQueriesParams) ([]Query, error)

+ type MySQLQueueSchema struct

+ GenerateMessagesTableName func(topic string) string

+ GeneratePayloadType func(topic string) string

+ GenerateWhereClause func(params GenerateWhereClauseParams) (string, []any)

+ SubscribeBatchSize int

+ func (s MySQLQueueSchema) InsertQuery(params InsertQueryParams) (Query, error)

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

+ func (s MySQLQueueSchema) SchemaInitializingQueries(params SchemaInitializingQueriesParams) ([]Query, error)

+ func (s MySQLQueueSchema) SelectQuery(params SelectQueryParams) (Query, error)

+ func (s MySQLQueueSchema) SubscribeIsolationLevel() sql.IsolationLevel

+ func (s MySQLQueueSchema) UnmarshalMessage(params UnmarshalMessageParams) (Row, error)

v4.0.0

Aug 25, 2025

Changes in this version

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

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

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

+ func DefaultSchemaInitializationLock(appName string) int

+ func NewDelayedPostgreSQLPublisher(db ContextExecutor, config DelayedPostgreSQLPublisherConfig) (message.Publisher, error)

+ func NewDelayedPostgreSQLSubscriber(db Beginner, config DelayedPostgreSQLSubscriberConfig) (message.Subscriber, error)

+ type AckMessageQueryParams struct

+ ConsumerGroup string

+ LastRow Row

+ Rows []Row

+ Topic string

+ type BackoffManager interface

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

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

+ type BeforeSubscribingQueriesParams struct

+ ConsumerGroup string

+ Topic string

+ type Beginner interface

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

+ func BeginnerFromPgx(conn Conn) Beginner

+ func BeginnerFromStdSQL(sqlBeginner SQLBeginner) Beginner

+ type Conn interface

+ BeginTx func(ctx context.Context, options pgx.TxOptions) (pgx.Tx, error)

+ Exec func(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)

+ Query func(ctx context.Context, sql string, arguments ...any) (pgx.Rows, error)

+ type ConsumedMessageQueryParams struct

+ ConsumerGroup string

+ ConsumerULID []byte

+ Row Row

+ Topic string

+ type ContextExecutor interface

+ ExecContext func(ctx context.Context, query string, args ...any) (Result, error)

+ QueryContext func(ctx context.Context, query string, args ...any) (Rows, error)

+ type DefaultMySQLOffsetsAdapter struct

+ GenerateMessagesOffsetsTableName func(topic string) string

+ func (a DefaultMySQLOffsetsAdapter) AckMessageQuery(params AckMessageQueryParams) (Query, error)

+ func (a DefaultMySQLOffsetsAdapter) BeforeSubscribingQueries(params BeforeSubscribingQueriesParams) ([]Query, error)

+ func (a DefaultMySQLOffsetsAdapter) ConsumedMessageQuery(params ConsumedMessageQueryParams) (Query, error)

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

+ func (a DefaultMySQLOffsetsAdapter) NextOffsetQuery(params NextOffsetQueryParams) (Query, error)

+ func (a DefaultMySQLOffsetsAdapter) SchemaInitializingQueries(params OffsetsSchemaInitializingQueriesParams) ([]Query, error)

+ type DefaultMySQLSchema struct

+ GenerateMessagesTableName func(topic string) string

+ GeneratePayloadType func(topic string) string

+ SubscribeBatchSize int

+ func (s DefaultMySQLSchema) InsertQuery(params InsertQueryParams) (Query, error)

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

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

+ func (s DefaultMySQLSchema) SchemaInitializingQueries(params SchemaInitializingQueriesParams) ([]Query, error)

+ func (s DefaultMySQLSchema) SelectQuery(params SelectQueryParams) (Query, error)

+ func (s DefaultMySQLSchema) SubscribeIsolationLevel() sql.IsolationLevel

+ func (s DefaultMySQLSchema) UnmarshalMessage(params UnmarshalMessageParams) (Row, error)

+ type DefaultPostgreSQLOffsetsAdapter struct

+ GenerateMessagesOffsetsTableName func(topic string) string

+ func (a DefaultPostgreSQLOffsetsAdapter) AckMessageQuery(params AckMessageQueryParams) (Query, error)

+ func (a DefaultPostgreSQLOffsetsAdapter) BeforeSubscribingQueries(params BeforeSubscribingQueriesParams) ([]Query, error)

+ func (a DefaultPostgreSQLOffsetsAdapter) ConsumedMessageQuery(params ConsumedMessageQueryParams) (Query, error)

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

+ func (a DefaultPostgreSQLOffsetsAdapter) NextOffsetQuery(params NextOffsetQueryParams) (Query, error)

+ func (a DefaultPostgreSQLOffsetsAdapter) SchemaInitializingQueries(params OffsetsSchemaInitializingQueriesParams) ([]Query, error)

+ type DefaultPostgreSQLSchema struct

+ GenerateMessagesTableName func(topic string) string

+ GeneratePayloadType func(topic string) string

+ InitializeSchemaLock int

+ InitializeSchemaWithoutTransaction bool

+ SubscribeBatchSize int

+ func (s DefaultPostgreSQLSchema) InsertQuery(params InsertQueryParams) (Query, error)

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

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

+ func (s DefaultPostgreSQLSchema) RequiresTransaction() bool

+ func (s DefaultPostgreSQLSchema) SchemaInitializingQueries(params SchemaInitializingQueriesParams) ([]Query, error)

+ func (s DefaultPostgreSQLSchema) SelectQuery(params SelectQueryParams) (Query, error)

+ func (s DefaultPostgreSQLSchema) SubscribeIsolationLevel() sql.IsolationLevel

+ func (s DefaultPostgreSQLSchema) UnmarshalMessage(params UnmarshalMessageParams) (Row, error)

+ type DefaultSchema = DefaultMySQLSchema

+ type DelayedPostgreSQLPublisherConfig struct

+ DelayPublisherConfig delay.PublisherConfig

+ Logger watermill.LoggerAdapter

+ OverridePublisherConfig func(config *PublisherConfig) error

+ type DelayedPostgreSQLSubscriberConfig struct

+ AllowNoDelay bool

+ DeleteOnAck bool

+ Logger watermill.LoggerAdapter

+ OverrideSubscriberConfig func(config *SubscriberConfig) error

+ type DelayedRequeuer struct

+ func NewPostgreSQLDelayedRequeuer(config DelayedRequeuerConfig) (*DelayedRequeuer, error)

+ func (q DelayedRequeuer) Middleware() []message.HandlerMiddleware

+ func (q DelayedRequeuer) Run(ctx context.Context) error

+ type DelayedRequeuerConfig struct

+ DB Beginner

+ DelayOnError *middleware.DelayOnError

+ GeneratePublishTopic func(params requeuer.GeneratePublishTopicParams) (string, error)

+ Logger watermill.LoggerAdapter

+ Publisher message.Publisher

+ RequeueTopic string

+ func (c *DelayedRequeuerConfig) Validate() error

+ type GenerateWhereClauseParams struct

+ Topic string

+ type InsertQueryParams struct

+ Msgs message.Messages

+ Topic string

+ type NextOffsetQueryParams struct

+ ConsumerGroup string

+ Topic string

+ type OffsetsAdapter interface

+ AckMessageQuery func(params AckMessageQueryParams) (Query, error)

+ BeforeSubscribingQueries func(params BeforeSubscribingQueriesParams) ([]Query, error)

+ ConsumedMessageQuery func(params ConsumedMessageQueryParams) (Query, error)

+ NextOffsetQuery func(params NextOffsetQueryParams) (Query, error)

+ SchemaInitializingQueries func(params OffsetsSchemaInitializingQueriesParams) ([]Query, error)

+ type OffsetsSchemaInitializingQueriesParams struct

+ Topic string

+ type PgxBeginner struct

+ func (c PgxBeginner) BeginTx(ctx context.Context, options *stdSQL.TxOptions) (Tx, error)

+ func (c PgxBeginner) ExecContext(ctx context.Context, query string, args ...any) (Result, error)

+ func (c PgxBeginner) QueryContext(ctx context.Context, query string, args ...any) (Rows, error)

+ type PgxResult struct

+ func (p PgxResult) RowsAffected() (int64, error)

+ type PgxRows struct

+ func (p PgxRows) Close() error

+ type PgxTx struct

+ func (t PgxTx) Commit() error

+ func (t PgxTx) ExecContext(ctx context.Context, query string, args ...any) (Result, error)

+ func (t PgxTx) QueryContext(ctx context.Context, query string, args ...any) (Rows, error)

+ func (t PgxTx) Rollback() error

+ type PostgreSQLQueueOffsetsAdapter struct

+ DeleteOnAck bool

+ GenerateMessagesTableName func(topic string) string

+ func (a PostgreSQLQueueOffsetsAdapter) AckMessageQuery(params AckMessageQueryParams) (Query, error)

+ func (a PostgreSQLQueueOffsetsAdapter) BeforeSubscribingQueries(params BeforeSubscribingQueriesParams) ([]Query, error)

+ func (a PostgreSQLQueueOffsetsAdapter) ConsumedMessageQuery(params ConsumedMessageQueryParams) (Query, error)

+ func (a PostgreSQLQueueOffsetsAdapter) MessagesTable(topic string) string

+ func (a PostgreSQLQueueOffsetsAdapter) NextOffsetQuery(params NextOffsetQueryParams) (Query, error)

+ func (a PostgreSQLQueueOffsetsAdapter) SchemaInitializingQueries(params OffsetsSchemaInitializingQueriesParams) ([]Query, error)

+ type PostgreSQLQueueSchema struct

+ GenerateMessagesTableName func(topic string) string

+ GeneratePayloadType func(topic string) string

+ GenerateWhereClause func(params GenerateWhereClauseParams) (string, []any)

+ SubscribeBatchSize int

+ func (s PostgreSQLQueueSchema) InsertQuery(params InsertQueryParams) (Query, error)

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

+ func (s PostgreSQLQueueSchema) SchemaInitializingQueries(params SchemaInitializingQueriesParams) ([]Query, error)

+ func (s PostgreSQLQueueSchema) SelectQuery(params SelectQueryParams) (Query, error)

+ func (s PostgreSQLQueueSchema) SubscribeIsolationLevel() sql.IsolationLevel

+ func (s PostgreSQLQueueSchema) UnmarshalMessage(params UnmarshalMessageParams) (Row, error)

+ type Publisher struct

+ func NewPublisher(db ContextExecutor, 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 Query struct

+ Args []any

+ Query string

+ func (q Query) IsZero() bool

+ func (q Query) String() string

+ type RequiresTransaction interface

+ RequiresTransaction func() bool

+ type Result interface

+ RowsAffected func() (int64, error)

+ type Row struct

+ ExtraData map[string]any

+ Metadata []byte

+ Msg *message.Message

+ Offset int64

+ Payload []byte

+ UUID []byte

+ type Rows interface

+ Close func() error

+ Next func() bool

+ Scan func(dest ...any) error

+ type SQLBeginner interface

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

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

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

+ type Scanner interface

+ Scan func(dest ...any) error

+ type SchemaAdapter interface

+ InsertQuery func(params InsertQueryParams) (Query, error)

+ SchemaInitializingQueries func(params SchemaInitializingQueriesParams) ([]Query, error)

+ SelectQuery func(params SelectQueryParams) (Query, error)

+ SubscribeIsolationLevel func() sql.IsolationLevel

+ UnmarshalMessage func(params UnmarshalMessageParams) (Row, error)

+ type SchemaInitializingQueriesParams struct

+ Topic string

+ type SelectQueryParams struct

+ ConsumerGroup string

+ OffsetsAdapter OffsetsAdapter

+ Topic string

+ type StdSQLBeginner struct

+ func (c StdSQLBeginner) BeginTx(ctx context.Context, options *sql.TxOptions) (Tx, error)

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

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

+ type StdSQLTx struct

+ func (t StdSQLTx) ExecContext(ctx context.Context, query string, args ...any) (Result, error)

+ func (t StdSQLTx) QueryContext(ctx context.Context, query string, args ...any) (Rows, error)

+ type Subscriber struct

+ func NewSubscriber(db Beginner, 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

+ AckDeadline *time.Duration

+ BackoffManager BackoffManager

+ ConsumerGroup string

+ InitializeSchema bool

+ OffsetsAdapter OffsetsAdapter

+ PollInterval time.Duration

+ ResendInterval time.Duration

+ RetryInterval time.Duration

+ SchemaAdapter SchemaAdapter

+ type Tx interface

+ Commit func() error

+ Rollback func() error

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

+ func TxFromPgx(tx pgx.Tx) Tx

+ func TxFromStdSQL(tx *sql.Tx) Tx

+ type UnmarshalMessageParams struct

+ Row Scanner

+ type XID8 uint64

+ func (x *XID8) Scan(src interface{}) error

+ func (x *XID8) Value() (driver.Value, error)

v4.0.0-rc.6

Jun 12, 2025

v4.0.0-rc.5

Apr 30, 2025

v4.0.0-rc.4

Jan 31, 2025

v4.0.0-rc.3

Jan 30, 2025

v4.0.0-rc.2

Dec 18, 2024

v4.0.0-rc.1

Oct 25, 2024

Other modules containing this package

github.com/ThreeDotsLabs/watermill-sql

github.com/ThreeDotsLabs/watermill-sql/v2

github.com/ThreeDotsLabs/watermill-sql/v3

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