gotx

README

gotx

A transaction manager based on sqlx(github.com/jmoiron/sqlx) for Golang. Since it is based on sqlx, the query methods in gotx(GetOne, Select, Insert, etc.) are similar to those in sqlx.

GoTx manages logical transactions binded to a goroutine in a group when the propagation type is set to PropagationRequired. These logical transactions are then binded to one underlying database transaction. GoTx uses goroutine ID to track logical transactions in a group.

When the propagation type is set to PropagationNew. Each logical transaction is binded to one underlying database transaction.

In the case of PropagationRequired, GoTx supports nested logical transactions so please feel free to call other transaction wrapped functions/methods from a transaction function/method in your service layer. With a nested transaction a commit does not persist changes until the top level transaction commit. While a rollback works regardless of the level of the transactions. That is when a rollback is triggered, all changes of nested logical transactions are discarded. When using PropagationNew every transaction do commit and rollback independently.

NOT READY FOR PRODUCTION USE. PLEASE USE IT AT YOUR OWN RISK.

install

go get github.com/oligo/gotx

usage

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

	"github.com/jmoiron/sqlx"
	"github.com/oligo/gotx"
)

// account model as an example here.
// Gotx use sqlx db tag here to do column mapping
type Account struct {
	ID   int64  `db:"id"`
	Name string `db:"name"`
    Articles []*Article
}

type Article struct {
    ID   int64  `db:"id"`
	Content string `db:"content"`
}

func main() {
	//using a DNS for mysql here
	dsn := "root:12345@tcp(127.0.0.1:3306)/db"

	db, err := sqlx.Connect("mysql", dsn)

	if err != nil {
		panic(err)
	}

	txManager := gotx.NewTxManager(db)

	// declares the tx options here. Please note that if options is omitted,
	// the default option is used. By default we use PropagationRequired for propagation
	// and RepeatableRead for isolation level.
	opts := &gotx.Options{
		Propagation:    gotx.PropagationRequired,
		IsolationLevel: sql.LevelRepeatableRead,
	}

	// use userID=10 for example
	userID := 10
	var account Account
	_ = txManager.Exec(context.Background(), func(tx *gotx.Transaction) error {
		// for more query method, please consult the godoc of this project
		err := tx.GetOne(&account, "select id, name from account where id=?", userID)
		if err != nil {
			return err
		}

        var articles []*Article
        // execute a nested transaction
        err = txManager.Exec(context.Background(), func(tx *gotx.Transaction) error {
            // for more query method, please consult the godoc of this project
            err := tx.Select(&articles, "select * from article where user_id=?", userID)
            if err != nil {
                return err
            }

            account.Articles = articles
            return nil
        }, opts)

		// Any error returned in this function will trigger the rollback of the transaction.
		return err

	}, opts)

}

Issues

Gotx uses goroutine ID to track nested transactions. This will be a problem when any of the nested transactions is executed in a seperate goroutine and we demand a PropagationRequired propagation setting. This may cause db locked when we are using a database system that does not support nested transaction such as Sqlite3.

Why not using context to track nested transactions?

Go deliberately choose not to expose goroutine ID and there's not such ThreadLocal objects. This makes the management of states bound to the thread/goroutine hard. It is encouraged to use content.Contenxt to pass state around。In the case of Gotx, it is quite tricky to track transactions using content.WithValue. We have to return contenxt from every tx execution and pass it down to nested transaction. This leads to complicated API design. Gotx chooses to go with the hacking way that leverages the Goroutine identifier internally and this works.

Documentation

Index

• Variables
• type LogLevel
• type LoggerConfig
• type Options
• type PropagationType
• type Transaction
  • func NewTx(t *rawTx, txID string, requiredNew bool) *Transaction
  • func (t *Transaction) Commit() error
  • func (t *Transaction) Delete(query string, arg interface{}) (int64, error)
  • func (t *Transaction) GetOne(dest interface{}, query string, args ...interface{}) error
  • func (t *Transaction) Insert(query string, arg interface{}) (int64, error)
  • func (t *Transaction) NamedExec(query string, arg interface{}) (int64, error)
  • func (t *Transaction) NamedQuery(dest interface{}, query string, arg interface{}) error
  • func (t *Transaction) Rollback() error
  • func (t *Transaction) Select(dest interface{}, query string, args ...interface{}) error
  • func (t *Transaction) String() string
  • func (t *Transaction) Update(query string, arg interface{}) (int64, error)
• type TxManager
  • func NewTxManager(db *sqlx.DB) *TxManager
  • func (tm *TxManager) Exec(ctx context.Context, txFunc func(tx *Transaction) error, options *Options) error
• type ZapLogger
  • func NewLogger(cfg *LoggerConfig) *ZapLogger

Variables

var (
	// ErrInvalidTxState is returned when transaction is not initialized
	ErrInvalidTxState = errors.New("gotx: tx is already committed or rolled back")
)

Types

type LogLevel

type LogLevel string

const (
	LOG_LEVEL_DEBUG LogLevel = "debug"
	LOG_LEVEL_INFO  LogLevel = "info"
	LOG_LEVEL_ERROR LogLevel = "error"
)

type LoggerConfig

type LoggerConfig struct {
	Level  LogLevel `json:"level" yaml:"level"`
	Format string   `json:"format" yaml:"format"`
}

LoggerConfig for zap

type Options

type Options struct {
	// PropagationType specifies how the tx manager manages transaction propagation
	Propagation PropagationType

	IsolationLevel sql.IsolationLevel
}

Options declares some configurable options when starts a transaction

type PropagationType

type PropagationType uint8

PropagationType is an alias of uint8

const (
	// Required specifies the txFunc will be run in an existing db tx. If there's no existing tx,
	// tx manager will create a new one for it.
	PropagationRequired PropagationType = iota

	// New specifies the txFunc will be run in a separated new db tx.
	PropagationNew
)

constants that defines transaction propagation patterns

type Transaction

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

Transaction is a logical transaction which wraps a underlying db transaction (physical transaction)

func NewTx

func NewTx(t *rawTx, txID string, requiredNew bool) *Transaction

func (*Transaction) Commit

func (t *Transaction) Commit() error

func (*Transaction) Delete

func (t *Transaction) Delete(query string, arg interface{}) (int64, error)

func (*Transaction) GetOne

func (t *Transaction) GetOne(dest interface{}, query string, args ...interface{}) error

GetOne is the sqlx.Get wrapper

func (*Transaction) Insert

func (t *Transaction) Insert(query string, arg interface{}) (int64, error)

Insert implements sql insert logic and returns generated ID

func (*Transaction) NamedExec

func (t *Transaction) NamedExec(query string, arg interface{}) (int64, error)

NamedExec is a generic executor that does not return rows. It inspect the query to check if it has IN clause and process it with sqlx.In. Docs from sqlx doc:

Named queries are common to many other database packages. They allow you to use a bindvar syntax which refers to the names of struct fields or map keys to bind variables a query, rather than having to refer to everything positionally. The struct field naming conventions follow that of StructScan, using the NameMapper and the db struct tag.

func (*Transaction) NamedQuery

func (t *Transaction) NamedQuery(dest interface{}, query string, arg interface{}) error

func (*Transaction) Rollback

func (t *Transaction) Rollback() error

rollback reset the refCount and do the real rollback.

func (*Transaction) Select

func (t *Transaction) Select(dest interface{}, query string, args ...interface{}) error

func (*Transaction) String

func (t *Transaction) String() string

func (*Transaction) Update

func (t *Transaction) Update(query string, arg interface{}) (int64, error)

Update execute a update sql using sqlx NamedExec.

type TxManager

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

TxManager implements a basic transaction manager

func NewTxManager

func NewTxManager(db *sqlx.DB) *TxManager

func (*TxManager) Exec

func (tm *TxManager) Exec(ctx context.Context, txFunc func(tx *Transaction) error, options *Options) error

type ZapLogger

type ZapLogger struct {
	*zap.SugaredLogger
}

var (
	// A package wide logger instance
	Logger *ZapLogger
)

func NewLogger

func NewLogger(cfg *LoggerConfig) *ZapLogger