sql

package

Go to main page

Versions in this module

v3

v3.1.0

Sep 24, 2024

Changes in this version

type DefaultMySQLSchema

+ GeneratePayloadType func(topic string) string

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

type DefaultPostgreSQLSchema

+ GeneratePayloadType func(topic string) string

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

v3.0.3

Aug 28, 2024

v3.0.2

Aug 23, 2024

v3.0.1

Mar 28, 2024

v3.0.0

Dec 19, 2023

Changes in this version

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

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

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

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

+ type BackoffManager interface

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

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

+ 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 DefaultMySQLOffsetsAdapter struct

+ GenerateMessagesOffsetsTableName func(topic string) string

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

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

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

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

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

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

+ type DefaultMySQLSchema struct

+ GenerateMessagesTableName func(topic string) string

+ SubscribeBatchSize int

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

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

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

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

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

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

+ type DefaultPostgreSQLOffsetsAdapter struct

+ GenerateMessagesOffsetsTableName func(topic string) string

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

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

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

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

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

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

+ type DefaultPostgreSQLSchema struct

+ GenerateMessagesTableName func(topic string) string

+ SubscribeBatchSize int

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

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

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

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

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

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

+ type DefaultSchema = DefaultMySQLSchema

+ 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

+ type OffsetsAdapter interface

+ AckMessageQuery func(topic string, row Row, consumerGroup string) Query

+ BeforeSubscribingQueries func(topic string, consumerGroup string) []Query

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

+ NextOffsetQuery func(topic, consumerGroup string) Query

+ SchemaInitializingQueries func(topic string) []Query

+ 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 Row struct

+ ExtraData map[string]any

+ Metadata []byte

+ Msg *message.Message

+ Offset int64

+ Payload []byte

+ UUID []byte

+ type Scanner interface

+ Scan func(dest ...any) error

+ type SchemaAdapter interface

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

+ SchemaInitializingQueries func(topic string) []Query

+ SelectQuery func(topic string, consumerGroup string, offsetsAdapter OffsetsAdapter) Query

+ SubscribeIsolationLevel func() sql.IsolationLevel

+ UnmarshalMessage func(row Scanner) (Row, 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

Other modules containing this package

github.com/ThreeDotsLabs/watermill-sql

github.com/ThreeDotsLabs/watermill-sql/v2

github.com/ThreeDotsLabs/watermill-sql/v4

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