sqlite

Documentation

Overview

Package sqlite provides a durable, embedded stream.Service backed by SQLite.

Use it with any database/sql SQLite driver (modernc.org/sqlite, mattn/go-sqlite3). The caller is responsible for opening the *sql.DB; stream/sqlite does not own the connection lifetime.

SQLite has no FOR UPDATE SKIP LOCKED. Claim queries run inside a BEGIN IMMEDIATE transaction; concurrent claimers serialize on the database write lock. This is fine for embedded single-process workloads. For high throughput across many processes, use the pgx backend instead.

The database should be opened with WAL mode and a non-trivial busy_timeout for adequate write throughput. Recommended DSN suffix: "?_pragma=journal_mode(WAL)&_pragma=synchronous(NORMAL)&_pragma=busy_timeout(5000)"

Index

• Variables
• func Apply(ctx context.Context, db *sql.DB) error
• type Service
  • func New(db *sql.DB, opts ...stream.ServiceOption) *Service
  • func (s *Service) Close(ctx context.Context) error
  • func (s *Service) Publish(ctx context.Context, name string, payload []byte, opts ...stream.PublishOption) error
  • func (s *Service) Subscribe(ctx context.Context, name, group string, h stream.Handler, ...) (stream.Consumer, error)

Variables

var ErrClosed = errors.New("stream/sqlite: service closed")

ErrClosed is returned by Publish or Subscribe after Close.

var Schema string

Functions

func Apply

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

Apply executes the embedded schema against db. Idempotent.

Types

type Service

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

Service is the SQLite stream.Service implementation.

func New

func New(db *sql.DB, opts ...stream.ServiceOption) *Service

New constructs a stream/sqlite Service. The caller owns db.

func (*Service) Close

func (s *Service) Close(ctx context.Context) error

Close shuts down all consumers. Idempotent. ctx bounds the overall close: cancellation aborts before the next consumer is drained and returns ctx.Err() — already-drained consumers stay drained.

func (*Service) Publish

func (s *Service) Publish(ctx context.Context, name string, payload []byte, opts ...stream.PublishOption) error

Publish appends a message to the stream. After insert, signals any local subscribers via the per-stream wake channel and runs trim if MaxLen/MaxAge are set.

func (*Service) Subscribe

func (s *Service) Subscribe(ctx context.Context, name, group string, h stream.Handler, opts ...stream.SubscribeOption) (stream.Consumer, error)

Subscribe lazily creates the consumer group at the requested start position and spawns concurrency-many workers.