sql

package
v3.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 24, 2024 License: MIT Imports: 12 Imported by: 6

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrInvalidTopicName = errors.New("topic name should not contain characters matched by " + disallowedTopicCharacters.String())
View Source 
var (
	ErrPublisherClosed = errors.New("publisher is closed")
)
View Source 
var (
	ErrSubscriberClosed = errors.New("subscriber is closed")
)

Functions

func TxFromContext 

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

TxFromContext returns the transaction used by the subscriber to consume the message. The transaction will be committed if ack of the message is successful. When a nack is sent, the transaction will be rolled back.

It is useful when you want to ensure that data is updated only when the message is processed. Example usage: https://github.com/ThreeDotsLabs/watermill/tree/master/_examples/real-world-examples/exactly-once-delivery-counter

Types

type BackoffManager 

type BackoffManager interface {
	// HandleError handles the error possibly logging it or returning a backoff time depending on the error or the absence of the message.
	HandleError(logger watermill.LoggerAdapter, noMsg bool, err error) time.Duration
}

BackoffManager handles errors or empty result sets and computes the backoff time. You could for example create a stateful version that computes a backoff depending on the error frequency or make errors more or less persistent.

func NewDefaultBackoffManager 

func NewDefaultBackoffManager(pollInterval, retryInterval time.Duration) BackoffManager

type Beginner 

type Beginner interface {
	BeginTx(context.Context, *sql.TxOptions) (*sql.Tx, error)
	ContextExecutor
}

Beginner begins transactions.

type ContextExecutor 

type ContextExecutor interface {
	Executor

	ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
	QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
	QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row
}

ContextExecutor can perform SQL queries with context

type DefaultMySQLOffsetsAdapter 

type DefaultMySQLOffsetsAdapter struct {
	// GenerateMessagesOffsetsTableName may be used to override how the messages/offsets table name is generated.
	GenerateMessagesOffsetsTableName func(topic string) string
}

DefaultMySQLOffsetsAdapter is adapter for storing offsets for MySQL (or MariaDB) databases.

DefaultMySQLOffsetsAdapter is designed to support multiple subscribers with exactly once delivery and guaranteed order.

We are using FOR UPDATE in NextOffsetQuery to lock consumer group in offsets table.

When another consumer is trying to consume the same message, deadlock should occur in ConsumedMessageQuery. After deadlock, consumer will consume next message.

func (DefaultMySQLOffsetsAdapter) AckMessageQuery 

func (a DefaultMySQLOffsetsAdapter) AckMessageQuery(topic string, row Row, consumerGroup string) Query

func (DefaultMySQLOffsetsAdapter) BeforeSubscribingQueries 

func (a DefaultMySQLOffsetsAdapter) BeforeSubscribingQueries(topic, consumerGroup string) []Query

func (DefaultMySQLOffsetsAdapter) ConsumedMessageQuery 

func (a DefaultMySQLOffsetsAdapter) ConsumedMessageQuery(topic string, row Row, consumerGroup string, consumerULID []byte) Query

func (DefaultMySQLOffsetsAdapter) MessagesOffsetsTable 

func (a DefaultMySQLOffsetsAdapter) MessagesOffsetsTable(topic string) string

func (DefaultMySQLOffsetsAdapter) NextOffsetQuery 

func (a DefaultMySQLOffsetsAdapter) NextOffsetQuery(topic, consumerGroup string) Query

func (DefaultMySQLOffsetsAdapter) SchemaInitializingQueries 

func (a DefaultMySQLOffsetsAdapter) SchemaInitializingQueries(topic string) []Query

type DefaultMySQLSchema 

type DefaultMySQLSchema struct {
	// GenerateMessagesTableName may be used to override how the messages table name is generated.
	GenerateMessagesTableName func(topic string) string

	// GeneratePayloadType is the type of the payload column in the messages table.
	// By default, it's JSON. If your payload is not JSON, you can use BYTEA.
	GeneratePayloadType func(topic string) string

	// SubscribeBatchSize is the number of messages to be queried at once.
	//
	// Higher value, increases a chance of message re-delivery in case of crash or networking issues.
	// 1 is the safest value, but it may have a negative impact on performance when consuming a lot of messages.
	//
	// Default value is 100.
	SubscribeBatchSize int
}

DefaultMySQLSchema is a default implementation of SchemaAdapter based on MySQL. If you need some customization, you can use composition to change schema and method of unmarshaling. 

type MyMessagesSchema struct {
	DefaultMySQLSchema
}

func (m MyMessagesSchema) SchemaInitializingQueries(topic string) []string {
	createMessagesTable := strings.Join([]string{
		"CREATE TABLE IF NOT EXISTS " + m.MessagesTable(topic) + " (",
		"`offset` BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,",
		"`uuid` BINARY(16) NOT NULL,",
		"`created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,",
		"`payload` JSON DEFAULT NULL,",
		"`metadata` JSON DEFAULT NULL",
		");",
	}, "\n")

	return []string{createMessagesTable}
}

func (m MyMessagesSchema) UnmarshalMessage(row *sql.Row) (offset int, msg *message.Message, err error) {
	// ...

For debugging your custom schema, we recommend to inject logger with trace logging level which will print all SQL queries.

func (DefaultMySQLSchema) InsertQuery 

func (s DefaultMySQLSchema) InsertQuery(topic string, msgs message.Messages) (Query, error)

func (DefaultMySQLSchema) MessagesTable 

func (s DefaultMySQLSchema) MessagesTable(topic string) string

func (DefaultMySQLSchema) PayloadColumnType added in v3.1.0 

func (s DefaultMySQLSchema) PayloadColumnType(topic string) string

func (DefaultMySQLSchema) SchemaInitializingQueries 

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

func (DefaultMySQLSchema) SelectQuery 

func (s DefaultMySQLSchema) SelectQuery(topic string, consumerGroup string, offsetsAdapter OffsetsAdapter) Query

func (DefaultMySQLSchema) SubscribeIsolationLevel 

func (s DefaultMySQLSchema) SubscribeIsolationLevel() sql.IsolationLevel

func (DefaultMySQLSchema) UnmarshalMessage 

func (s DefaultMySQLSchema) UnmarshalMessage(row Scanner) (Row, error)

type DefaultPostgreSQLOffsetsAdapter 

type DefaultPostgreSQLOffsetsAdapter struct {
	// GenerateMessagesOffsetsTableName may be used to override how the messages/offsets table name is generated.
	GenerateMessagesOffsetsTableName func(topic string) string
}

DefaultPostgreSQLOffsetsAdapter is adapter for storing offsets in PostgreSQL database.

DefaultPostgreSQLOffsetsAdapter is designed to support multiple subscribers with exactly once delivery and guaranteed order.

We are using FOR UPDATE in NextOffsetQuery to lock consumer group in offsets table.

When another consumer is trying to consume the same message, deadlock should occur in ConsumedMessageQuery. After deadlock, consumer will consume next message.

func (DefaultPostgreSQLOffsetsAdapter) AckMessageQuery 

func (a DefaultPostgreSQLOffsetsAdapter) AckMessageQuery(topic string, row Row, consumerGroup string) Query

func (DefaultPostgreSQLOffsetsAdapter) BeforeSubscribingQueries 

func (a DefaultPostgreSQLOffsetsAdapter) BeforeSubscribingQueries(topic string, consumerGroup string) []Query

func (DefaultPostgreSQLOffsetsAdapter) ConsumedMessageQuery 

func (a DefaultPostgreSQLOffsetsAdapter) ConsumedMessageQuery(topic string, row Row, consumerGroup string, consumerULID []byte) Query

func (DefaultPostgreSQLOffsetsAdapter) MessagesOffsetsTable 

func (a DefaultPostgreSQLOffsetsAdapter) MessagesOffsetsTable(topic string) string

func (DefaultPostgreSQLOffsetsAdapter) NextOffsetQuery 

func (a DefaultPostgreSQLOffsetsAdapter) NextOffsetQuery(topic, consumerGroup string) Query

func (DefaultPostgreSQLOffsetsAdapter) SchemaInitializingQueries 

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

type DefaultPostgreSQLSchema 

type DefaultPostgreSQLSchema struct {
	// GenerateMessagesTableName may be used to override how the messages table name is generated.
	GenerateMessagesTableName func(topic string) string

	// GeneratePayloadType is the type of the payload column in the messages table.
	// By default, it's JSON. If your payload is not JSON, you can use BYTEA.
	GeneratePayloadType func(topic string) string

	// SubscribeBatchSize is the number of messages to be queried at once.
	//
	// Higher value, increases a chance of message re-delivery in case of crash or networking issues.
	// 1 is the safest value, but it may have a negative impact on performance when consuming a lot of messages.
	//
	// Default value is 100.
	SubscribeBatchSize int
}

DefaultPostgreSQLSchema is a default implementation of SchemaAdapter based on PostgreSQL.

func (DefaultPostgreSQLSchema) InsertQuery 

func (s DefaultPostgreSQLSchema) InsertQuery(topic string, msgs message.Messages) (Query, error)

func (DefaultPostgreSQLSchema) MessagesTable 

func (s DefaultPostgreSQLSchema) MessagesTable(topic string) string

func (DefaultPostgreSQLSchema) PayloadColumnType added in v3.1.0 

func (s DefaultPostgreSQLSchema) PayloadColumnType(topic string) string

func (DefaultPostgreSQLSchema) SchemaInitializingQueries 

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

func (DefaultPostgreSQLSchema) SelectQuery 

func (s DefaultPostgreSQLSchema) SelectQuery(topic string, consumerGroup string, offsetsAdapter OffsetsAdapter) Query

func (DefaultPostgreSQLSchema) SubscribeIsolationLevel 

func (s DefaultPostgreSQLSchema) SubscribeIsolationLevel() sql.IsolationLevel

func (DefaultPostgreSQLSchema) UnmarshalMessage 

func (s DefaultPostgreSQLSchema) UnmarshalMessage(row Scanner) (Row, error)

type DefaultSchema deprecated 

type DefaultSchema = DefaultMySQLSchema

Deprecated: Use DefaultMySQLSchema instead.

type Executor 

type Executor interface {
	Exec(query string, args ...interface{}) (sql.Result, error)
	Query(query string, args ...interface{}) (*sql.Rows, error)
	QueryRow(query string, args ...interface{}) *sql.Row
}

Executor can perform SQL queries.

type OffsetsAdapter 

type OffsetsAdapter interface {
	// AckMessageQuery the SQL query and arguments that will mark a message as read for a given consumer group.
	AckMessageQuery(topic string, row Row, consumerGroup string) Query

	// ConsumedMessageQuery will return the SQL query and arguments which be executed after consuming message,
	// but before ack.
	//
	// ConsumedMessageQuery is optional, and will be not executed if query is empty.
	ConsumedMessageQuery(topic string, row Row, consumerGroup string, consumerULID []byte) Query

	// NextOffsetQuery returns the SQL query and arguments which should return offset of next message to consume.
	NextOffsetQuery(topic, consumerGroup string) Query

	// SchemaInitializingQueries returns SQL queries which will make sure (CREATE IF NOT EXISTS)
	// that the appropriate tables exist to write messages to the given topic.
	SchemaInitializingQueries(topic string) []Query

	// BeforeSubscribingQueries returns queries which will be executed before subscribing to a topic.
	// All queries will be executed in a single transaction.
	BeforeSubscribingQueries(topic string, consumerGroup string) []Query
}

type Publisher 

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

Publisher inserts the Messages as rows into a SQL table..

func NewPublisher 

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

func (*Publisher) Close 

func (p *Publisher) Close() error

Close closes the publisher, which means that all the Publish calls called before are finished and no more Publish calls are accepted. Close is blocking until all the ongoing Publish calls have returned.

func (*Publisher) Publish 

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

Publish inserts the messages as rows into the MessagesTable. Order is guaranteed for messages within one call. Publish is blocking until all rows have been added to the Publisher's transaction. Publisher doesn't guarantee publishing messages in a single transaction, but the constructor accepts both *sql.DB and *sql.Tx, so transactions may be handled upstream by the user.

type PublisherConfig 

type PublisherConfig struct {
	// SchemaAdapter provides the schema-dependent queries and arguments for them, based on topic/message etc.
	SchemaAdapter SchemaAdapter

	// AutoInitializeSchema enables initialization of schema database during publish.
	// Schema is initialized once per topic per publisher instance.
	// AutoInitializeSchema is forbidden if using an ongoing transaction as database handle;
	// That could result in an implicit commit of the transaction by a CREATE TABLE statement.
	AutoInitializeSchema bool
}

type Query 

type Query struct {
	Query string
	Args  []any
}

func (Query) IsZero 

func (q Query) IsZero() bool

func (Query) String 

func (q Query) String() string

type Row 

type Row struct {
	Offset   int64
	UUID     []byte
	Payload  []byte
	Metadata []byte

	Msg *message.Message

	ExtraData map[string]any
}

type Scanner 

type Scanner interface {
	Scan(dest ...any) error
}

type SchemaAdapter 

type SchemaAdapter interface {
	// InsertQuery returns the SQL query and arguments that will insert the Watermill message into the SQL storage.
	InsertQuery(topic string, msgs message.Messages) (Query, error)

	// SelectQuery returns the SQL query and arguments
	// that returns the next unread message for a given consumer group.
	SelectQuery(topic string, consumerGroup string, offsetsAdapter OffsetsAdapter) Query

	// UnmarshalMessage transforms the Row obtained SelectQuery a Watermill message.
	// It also returns the offset of the last read message, for the purpose of acking.
	UnmarshalMessage(row Scanner) (Row, error)

	// SchemaInitializingQueries returns SQL queries which will make sure (CREATE IF NOT EXISTS)
	// that the appropriate tables exist to write messages to the given topic.
	SchemaInitializingQueries(topic string) []Query

	// SubscribeIsolationLevel returns the isolation level that will be used when subscribing.
	SubscribeIsolationLevel() sql.IsolationLevel
}

SchemaAdapter produces the SQL queries and arguments appropriately for a specific schema and dialect It also transforms sql.Rows into Watermill messages.

type Subscriber 

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

Subscriber makes SELECT queries on the chosen table with the interval defined in the config. The rows are unmarshaled into Watermill messages.

func NewSubscriber 

func NewSubscriber(db Beginner, config SubscriberConfig, logger watermill.LoggerAdapter) (*Subscriber, error)

func (*Subscriber) Close 

func (s *Subscriber) Close() error

func (*Subscriber) Subscribe 

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

func (*Subscriber) SubscribeInitialize 

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

type SubscriberConfig 

type SubscriberConfig struct {
	ConsumerGroup string

	// AckDeadline is the time to wait for acking a message.
	// If message is not acked within this time, it will be nacked and re-delivered.
	//
	// When messages are read in bulk, this time is calculated for each message separately.
	//
	// If you want to disable ack deadline, set it to 0.
	// Warning: when ack deadline is disabled, messages which are not acked may block PostgreSQL subscriber from reading new messages
	// due to not increasing `pg_snapshot_xmin(pg_current_snapshot())` value.
	//
	// Must be non-negative. Nil value defaults to 30s.
	AckDeadline *time.Duration

	// PollInterval is the interval to wait between subsequent SELECT queries, if no more messages were found in the database (Prefer using the BackoffManager instead).
	// Must be non-negative. Defaults to 1s.
	PollInterval time.Duration

	// ResendInterval is the time to wait before resending a nacked message.
	// Must be non-negative. Defaults to 1s.
	ResendInterval time.Duration

	// RetryInterval is the time to wait before resuming querying for messages after an error (Prefer using the BackoffManager instead).
	// Must be non-negative. Defaults to 1s.
	RetryInterval time.Duration

	// BackoffManager defines how much to backoff when receiving errors.
	BackoffManager BackoffManager

	// SchemaAdapter provides the schema-dependent queries and arguments for them, based on topic/message etc.
	SchemaAdapter SchemaAdapter

	// OffsetsAdapter provides mechanism for saving acks and offsets of consumers.
	OffsetsAdapter OffsetsAdapter

	// InitializeSchema option enables initializing schema on making subscription.
	InitializeSchema bool
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.