livesql

Documentation

Overview

Example

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/edwardofclt/thunder/internal/testfixtures"
	"github.com/edwardofclt/thunder/livesql"
	"github.com/edwardofclt/thunder/reactive"
	"github.com/edwardofclt/thunder/sqlgen"
	_ "github.com/go-sql-driver/mysql"
)

type Cat struct {
	Id   int64 `sql:",primary"`
	Name string
}

func main() {
	config := testfixtures.DefaultDBConfig
	testDb, err := testfixtures.NewTestDatabase()
	if err != nil {
		panic(err)
	}
	defer testDb.Close()

	// Create users table.
	testDb.Exec("CREATE TABLE `cats` (`id` BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY, `name` VARCHAR(100) NOT NULL);")

	// Register the User type in the schema.
	schema := sqlgen.NewSchema()
	schema.MustRegisterType("cats", sqlgen.AutoIncrement, Cat{})

	// Open a connection to the live DB.
	db, err := livesql.Open(config.Hostname, config.Port, config.Username, config.Password, testDb.DBName, schema)
	if err != nil {
		panic(err)
	}

	// Insert a dummy row.
	result, err := db.InsertRow(context.Background(), &Cat{Name: "Fluffles"})
	if err != nil {
		panic(err)
	}
	catId, err := result.LastInsertId()
	if err != nil {
		panic(err)
	}

	ch := make(chan struct{})
	// Wrap the query in a re-runner. This runner will re-trigger every time a change that would
	// affect queries inside of it being made.
	rerunner := reactive.NewRerunner(context.Background(), func(ctx context.Context) (interface{}, error) {
		user := &Cat{Id: catId}
		err := db.QueryRow(ctx, &user, sqlgen.Filter{}, nil)
		if err != nil {
			panic(err)
		}
		fmt.Println(user.Name)
		ch <- struct{}{}
		return nil, nil
	}, 200*time.Millisecond, false)
	defer rerunner.Stop()

	<-ch
	if err := db.UpdateRow(context.Background(), &Cat{Id: 1, Name: "Ruffles"}); err != nil {
		panic(err)
	}

	<-ch
}

Output:

Fluffles
Ruffles

Index

• func FieldToValue(field *thunderpb.Field) (driver.Value, error)
• func FilterFromProto(schema *sqlgen.Schema, proto *thunderpb.SQLFilter) (string, sqlgen.Filter, error)
• func FilterToProto(schema *sqlgen.Schema, tableName string, filter sqlgen.Filter) (*thunderpb.SQLFilter, error)
• type Binlog
  • func NewBinlog(ldb *LiveDB, host string, port uint16, username, password, database string) (*Binlog, error)
  • func NewBinlogWithSource(ldb *LiveDB, sourceDB *sql.DB, host string, port uint16, ...) (*Binlog, error)
  • func (b *Binlog) Close() error
  • func (b *Binlog) RunPollLoop() error
  • func (b *Binlog) SetLogger(l logger.Logger)
  • func (b *Binlog) SetUpdateDelay(d time.Duration)
• type LiveDB
  • func NewLiveDB(db *sqlgen.DB) *LiveDB
  • func Open(hostname string, port uint16, username, password, database string, ...) (*LiveDB, error)
  • func (ldb *LiveDB) AddDependency(ctx context.Context, dep QueryDependency) error
  • func (ldb *LiveDB) Close() error
  • func (ldb *LiveDB) FullScanQuery(ctx context.Context, result interface{}, filter sqlgen.Filter, ...) error
  • func (ldb *LiveDB) Query(ctx context.Context, result interface{}, filter sqlgen.Filter, ...) error
  • func (ldb *LiveDB) QueryRow(ctx context.Context, result interface{}, filter sqlgen.Filter, ...) error
• type QueryDependency

Examples

• Package

Functions

func FieldToValue

func FieldToValue(field *thunderpb.Field) (driver.Value, error)

FieldToValue converts thunderpb.Field to driver.Value.

func FilterFromProto

func FilterFromProto(schema *sqlgen.Schema, proto *thunderpb.SQLFilter) (string, sqlgen.Filter, error)

FilterFromProto takes a thunderpb.SQLFilter, runs Scanner on each field value, and returns a sqlgen.Filter.

func FilterToProto

func FilterToProto(schema *sqlgen.Schema, tableName string, filter sqlgen.Filter) (*thunderpb.SQLFilter, error)

FilterToProto takes a sqlgen.Filter, runs Valuer on each filter value, and returns a thunderpb.SQLFilter.

Types

type Binlog

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

Binlog streams the MySQL binary replication log, parses it, and broadcasts updates

func NewBinlog

func NewBinlog(ldb *LiveDB, host string, port uint16, username, password, database string) (*Binlog, error)

NewBinlog constructs a new Binlog for a given DB.

func NewBinlogWithSource

func NewBinlogWithSource(ldb *LiveDB, sourceDB *sql.DB, host string, port uint16, username, password, database string) (*Binlog, error)

NewBinlogWithSource constructs a new Binlog for a given DB and a source DB

NewBinlogWithSource verifies that the given DB has been correctly configured for streaming changes.

func (*Binlog) Close

func (b *Binlog) Close() error

func (*Binlog) RunPollLoop

func (b *Binlog) RunPollLoop() error

RunPollLoop is the core binlog function that fetches and distributes updates from MySQL

func (*Binlog) SetLogger

func (b *Binlog) SetLogger(l logger.Logger)

SetOutput sets the destination for the error logger.

func (*Binlog) SetUpdateDelay

func (b *Binlog) SetUpdateDelay(d time.Duration)

SetUpdateDelay sets the duration by which future updates will be delayed.

type LiveDB

type LiveDB struct {
	*sqlgen.DB
	// contains filtered or unexported fields
}

LiveDB is a SQL client that supports live updating queries. It relies on a reactive.Rerunner being in the context to register changes in the database (which are propagated through said rerunner to its clients). Without this rerunner being in the context it falls back to non-live (sqlgen) behavior. See https://godoc.org/github.com/samsarahq/thunder/reactive for information on reactive.

func NewLiveDB

func NewLiveDB(db *sqlgen.DB) *LiveDB

NewLiveDB constructs a new LiveDB

func Open

func Open(hostname string, port uint16, username, password, database string, schema *sqlgen.Schema) (*LiveDB, error)

func (*LiveDB) AddDependency

func (ldb *LiveDB) AddDependency(ctx context.Context, dep QueryDependency) error

func (*LiveDB) Close

func (ldb *LiveDB) Close() error

func (*LiveDB) FullScanQuery

func (ldb *LiveDB) FullScanQuery(ctx context.Context, result interface{}, filter sqlgen.Filter, options *sqlgen.SelectOptions) error

FullScanQuery bypasses any index checking on a query. Normal LiveDB.Query will check during tests if the query uses an index and will fail tests if not. This function will skip those checks. There are cases where we explicitly want to support full table scans such as

1. During tests to verify results (eg get all)
2. Some rare operations are infrequent and its better to have no index and instead perform full table scans when that query is run.

func (*LiveDB) Query

func (ldb *LiveDB) Query(ctx context.Context, result interface{}, filter sqlgen.Filter, options *sqlgen.SelectOptions) error

Query fetches a collection of rows from the database and will invalidate ctx when the query result changes

result should be a pointer to a slice of pointers to structs, for example:

var users []*User
if err := ldb.Query(ctx, &users, nil, nil); err != nil {

func (*LiveDB) QueryRow

func (ldb *LiveDB) QueryRow(ctx context.Context, result interface{}, filter sqlgen.Filter, options *sqlgen.SelectOptions) error

QueryRow fetches a single row from the database and will invalidate ctx when the query result changes

result should be a pointer to a pointer to a struct, for example:

var user *User
if err := ldb.Query(ctx, &user, Filter{"id": 10}, nil); err != nil {

type QueryDependency

type QueryDependency struct {
	Table  string
	Filter sqlgen.Filter
}

QueryDependency represents a dependency on SQL query.