postgres

Documentation

Overview

Package postgres is the implementation of the postgres data store.

Index

• Constants
• Variables
• type Postgres
  • func WithInstance(ctx context.Context, db *sql.DB) (*Postgres, error)
  • func (p *Postgres) Add(ctx context.Context, evt *outboxer.OutboxMessage) error
  • func (p *Postgres) AddWithinTx(ctx context.Context, evt *outboxer.OutboxMessage, ...) error
  • func (p *Postgres) Close() error
  • func (p *Postgres) GetEvents(ctx context.Context, batchSize int32) ([]*outboxer.OutboxMessage, error)
  • func (p *Postgres) Remove(ctx context.Context, dispatchedBefore time.Time, batchSize int32) error
  • func (p *Postgres) SetAsDispatched(ctx context.Context, id int64) error

Examples

• Postgres

Constants

const (
	// DefaultEventStoreTable is the default table name.
	DefaultEventStoreTable = "event_store"
)

Variables

var (
	// ErrLocked is used when we can't acquire an explicit lock.
	ErrLocked = errors.New("can't acquire lock")

	// ErrNoDatabaseName is used when the database name is blank.
	ErrNoDatabaseName = errors.New("no database name")

	// ErrNoSchema is used when the schema name is blank.
	ErrNoSchema = errors.New("no schema")
)

Types

type Postgres

type Postgres struct {
	DatabaseName    string
	SchemaName      string
	EventStoreTable string
	// contains filtered or unexported fields
}

Postgres is the implementation of the data store.

Example

package main

import (
	"context"
	"database/sql"
	"fmt"
	"os"

	"github.com/artsv79/outboxer/storage/postgres"
)

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	db, err := sql.Open("postgres", os.Getenv("DS_DSN"))
	if err != nil {
		fmt.Printf("failed to connect to postgres: %s", err)
		return
	}

	ds, err := postgres.WithInstance(ctx, db)
	if err != nil {
		fmt.Printf("failed to setup the data store: %s", err)
		return
	}

	ds.Close()
}

func WithInstance

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

WithInstance creates a postgres data store with an existing db connection.

func (*Postgres) Add

func (p *Postgres) Add(ctx context.Context, evt *outboxer.OutboxMessage) error

Add adds the message to the data store.

func (*Postgres) AddWithinTx

func (p *Postgres) AddWithinTx(ctx context.Context, evt *outboxer.OutboxMessage, fn func(outboxer.ExecerContext) error) error

AddWithinTx creates a transaction and then tries to execute anything within it.

func (*Postgres) Close

func (p *Postgres) Close() error

Close closes the db connection.

func (*Postgres) GetEvents

func (p *Postgres) GetEvents(ctx context.Context, batchSize int32) ([]*outboxer.OutboxMessage, error)

GetEvents retrieves all the relevant events.

func (*Postgres) Remove

func (p *Postgres) Remove(ctx context.Context, dispatchedBefore time.Time, batchSize int32) error

Remove removes old messages from the data store.

func (*Postgres) SetAsDispatched

func (p *Postgres) SetAsDispatched(ctx context.Context, id int64) error

SetAsDispatched sets one message as dispatched.