crdb

Documentation

Index

• Variables
• func ChangeFeedQuery(topic string, resolved time.Duration, cursor time.Time) string
• func InitializeClaimsSchema(ctx context.Context, db *sql.DB, topic string, consumerGroup string, ...) error
• func InitializeCursorsSchema(ctx context.Context, db *sql.DB, logger watermill.LoggerAdapter) error
• func InitializeMessageSchema(ctx context.Context, db *sql.DB, topic string, logger watermill.LoggerAdapter) error
• func InitializeSessionSchema(ctx context.Context, db *sql.DB) error
• func NewPublisher(conn DB, logger watermill.LoggerAdapter) *publisher
• func NewSubscriber(db *sql.DB, consumerGroup string, logger watermill.LoggerAdapter) *subscriber
• type CRDBEnvelope
• type DB
• type PublisherConfig
• type ResolvedTimestamp
• type Scanner
• type Session
  • func NewSession(db *sql.DB, logger watermill.LoggerAdapter) *Session
  • func (s *Session) Close() error
  • func (s *Session) Observe(ctx context.Context, tx *sql.Tx) error
  • func (s *Session) Run(ctx context.Context) error
  • func (s *Session) TableName() string

Variables

var ErrInvalidTopicName = errors.New("topic name should not contain characters matched by " + disallowedTopicCharacters.String())

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

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

Functions

func ChangeFeedQuery

func ChangeFeedQuery(topic string, resolved time.Duration, cursor time.Time) string

func InitializeClaimsSchema

func InitializeClaimsSchema(
	ctx context.Context,
	db *sql.DB,
	topic string,
	consumerGroup string,
	logger watermill.LoggerAdapter,
) error

func InitializeCursorsSchema

func InitializeCursorsSchema(
	ctx context.Context, db *sql.DB, logger watermill.LoggerAdapter,
) error

func InitializeMessageSchema

func InitializeMessageSchema(
	ctx context.Context, db *sql.DB, topic string, logger watermill.LoggerAdapter,
) error

func InitializeSessionSchema

func InitializeSessionSchema(ctx context.Context, db *sql.DB) error

func NewPublisher

func NewPublisher(conn DB, logger watermill.LoggerAdapter) *publisher

func NewSubscriber

func NewSubscriber(db *sql.DB, consumerGroup string, logger watermill.LoggerAdapter) *subscriber

Types

type CRDBEnvelope

type CRDBEnvelope struct {
	Value ResolvedTimestamp `json:"__crdb__"`
}

type DB

type DB interface {
	ExecContext(context.Context, string, ...interface{}) (sql.Result, error)
}

type PublisherConfig

type PublisherConfig struct {
	AutoInitializeSchema bool
}

type ResolvedTimestamp

type ResolvedTimestamp struct {
	Resolved string `json:"resolved"`
}

type Scanner

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

type Session

type Session struct {
	Start     chan struct{}
	SessionID string
	// contains filtered or unexported fields
}

func NewSession

func NewSession(db *sql.DB, logger watermill.LoggerAdapter) *Session

func (*Session) Close

func (s *Session) Close() error

func (*Session) Observe

func (s *Session) Observe(ctx context.Context, tx *sql.Tx) error

func (*Session) Run

func (s *Session) Run(ctx context.Context) error

func (*Session) TableName

func (s *Session) TableName() string