sql

package
v1.4.0 Latest Latest
Warning

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

 Go to latest Published: May 23, 2023 License: MIT Imports: 12 Imported by: 9

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 added in v1.3.5 

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 added in v1.3.1 

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 added in v1.3.1 

func NewDefaultBackoffManager(pollInterval, retryInterval time.Duration) BackoffManager

type Beginner added in v1.3.7 

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

Beginner begins transactions.

type ContextExecutor added in v1.3.7 

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, offset int, consumerGroup string) (string, []interface{})

func (DefaultMySQLOffsetsAdapter) ConsumedMessageQuery 

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

func (DefaultMySQLOffsetsAdapter) MessagesOffsetsTable 

func (a DefaultMySQLOffsetsAdapter) MessagesOffsetsTable(topic string) string

func (DefaultMySQLOffsetsAdapter) NextOffsetQuery 

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

func (DefaultMySQLOffsetsAdapter) SchemaInitializingQueries 

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

type DefaultMySQLSchema added in v1.1.0 

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

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 added in v1.1.0 

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

func (DefaultMySQLSchema) MessagesTable added in v1.1.0 

func (s DefaultMySQLSchema) MessagesTable(topic string) string

func (DefaultMySQLSchema) SchemaInitializingQueries added in v1.1.0 

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

func (DefaultMySQLSchema) SelectQuery added in v1.1.0 

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

func (DefaultMySQLSchema) UnmarshalMessage added in v1.1.0 

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

type DefaultPostgreSQLOffsetsAdapter added in v1.1.0 

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 added in v1.1.0 

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

func (DefaultPostgreSQLOffsetsAdapter) ConsumedMessageQuery added in v1.1.0 

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

func (DefaultPostgreSQLOffsetsAdapter) MessagesOffsetsTable added in v1.1.0 

func (a DefaultPostgreSQLOffsetsAdapter) MessagesOffsetsTable(topic string) string

func (DefaultPostgreSQLOffsetsAdapter) NextOffsetQuery added in v1.1.0 

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

func (DefaultPostgreSQLOffsetsAdapter) SchemaInitializingQueries added in v1.1.0 

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

type DefaultPostgreSQLSchema added in v1.1.0 

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

DefaultPostgreSQLSchema is a default implementation of SchemaAdapter based on PostgreSQL.

func (DefaultPostgreSQLSchema) InsertQuery added in v1.1.0 

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

func (DefaultPostgreSQLSchema) MessagesTable added in v1.1.0 

func (s DefaultPostgreSQLSchema) MessagesTable(topic string) string

func (DefaultPostgreSQLSchema) SchemaInitializingQueries added in v1.1.0 

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

func (DefaultPostgreSQLSchema) SelectQuery added in v1.1.0 

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

func (DefaultPostgreSQLSchema) UnmarshalMessage added in v1.1.0 

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

type DefaultSchema deprecated 

type DefaultSchema = DefaultMySQLSchema

Deprecated: Use DefaultMySQLSchema instead.

type Executor added in v1.3.7 

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, offset int, consumerGroup string) (string, []interface{})

	// 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, offset int, consumerGroup string, consumerULID []byte) (string, []interface{})

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

	// 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) []string
}

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 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) (string, []interface{}, error)

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

	// 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 *sql.Row) (offset int, msg *message.Message, err 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) []string
}

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

	// 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 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.