transaction

Documentation

Index

• Variables
• type BaseTx
• type Manager
  • func NewManager(kvStore kv.KeyValueStore) *Manager
  • func (m *Manager) StartTx(ctx context.Context) (Tx, error)
• type SessionCtx
  • func (c *SessionCtx) GetStagedDatabase() StagedDB
  • func (c *SessionCtx) StageDatabase(db StagedDB)
• type StagedDB
• type Tx
• type TxSession
  • func (s *TxSession) Commit(ctx context.Context) error
  • func (s *TxSession) Context() *SessionCtx
  • func (s *TxSession) Delete(ctx context.Context, key keys.Key) error
  • func (s *TxSession) Get(ctx context.Context, key []byte, isSnapshot bool) (kv.Future, error)
  • func (s *TxSession) GetTxCtx() *api.TransactionCtx
  • func (s *TxSession) Insert(ctx context.Context, key keys.Key, data *internal.TableData) error
  • func (s *TxSession) Read(ctx context.Context, key keys.Key) (kv.Iterator, error)
  • func (s *TxSession) ReadRange(ctx context.Context, lKey keys.Key, rKey keys.Key, isSnapshot bool) (kv.Iterator, error)
  • func (s *TxSession) Replace(ctx context.Context, key keys.Key, data *internal.TableData, isUpdate bool) error
  • func (s *TxSession) Rollback(ctx context.Context) error
  • func (s *TxSession) SetVersionstampedKey(ctx context.Context, key []byte, value []byte) error
  • func (s *TxSession) SetVersionstampedValue(ctx context.Context, key []byte, value []byte) error
  • func (s *TxSession) Update(ctx context.Context, key keys.Key, ...) (int32, error)

Variables

var (
	// ErrSessionIsNotStarted is returned when the session is not started but is getting used
	ErrSessionIsNotStarted = api.Errorf(api.Code_INTERNAL, "session not started")

	// ErrSessionIsGone is returned when the session is gone but getting used
	ErrSessionIsGone = api.Errorf(api.Code_INTERNAL, "session is gone")
)

Types

type BaseTx

type BaseTx interface {
	Context() *SessionCtx
	GetTxCtx() *api.TransactionCtx
	Insert(ctx context.Context, key keys.Key, data *internal.TableData) error
	Replace(ctx context.Context, key keys.Key, data *internal.TableData, isUpdate bool) error
	Update(ctx context.Context, key keys.Key, apply func(*internal.TableData) (*internal.TableData, error)) (int32, error)
	Delete(ctx context.Context, key keys.Key) error
	Read(ctx context.Context, key keys.Key) (kv.Iterator, error)
	ReadRange(ctx context.Context, lKey keys.Key, rKey keys.Key, isSnapshot bool) (kv.Iterator, error)
	Get(ctx context.Context, key []byte, isSnapshot bool) (kv.Future, error)
	SetVersionstampedValue(ctx context.Context, key []byte, value []byte) error
	SetVersionstampedKey(ctx context.Context, key []byte, value []byte) error
}

BaseTx interface exposes base methods that can be used on a transactional object.

type Manager

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

func NewManager

func NewManager(kvStore kv.KeyValueStore) *Manager

func (*Manager) StartTx

func (m *Manager) StartTx(ctx context.Context) (Tx, error)

StartTx starts a new read-write tx session.

type SessionCtx

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

SessionCtx is used to store any baggage for the lifetime of the transaction. We use it to stage the database inside a transaction when the transaction is performing any DDLs.

func (*SessionCtx) GetStagedDatabase

func (c *SessionCtx) GetStagedDatabase() StagedDB

func (*SessionCtx) StageDatabase

func (c *SessionCtx) StageDatabase(db StagedDB)

type StagedDB

type StagedDB interface {
	Name() string
	GetCollection(string) *schema.DefaultCollection
}

type Tx

type Tx interface {
	BaseTx

	Commit(ctx context.Context) error
	Rollback(ctx context.Context) error
}

type TxSession

type TxSession struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

TxSession is used to start an explicit transaction. Caller can control whether this transaction's session needs to be tracked inside session tracker. Tracker a session is useful if the object is shared across the requests otherwise it is not useful in the same request flow.

func (*TxSession) Commit

func (s *TxSession) Commit(ctx context.Context) error

func (*TxSession) Context

func (s *TxSession) Context() *SessionCtx

func (*TxSession) Delete

func (s *TxSession) Delete(ctx context.Context, key keys.Key) error

func (*TxSession) Get

func (s *TxSession) Get(ctx context.Context, key []byte, isSnapshot bool) (kv.Future, error)

func (*TxSession) GetTxCtx

func (s *TxSession) GetTxCtx() *api.TransactionCtx

func (*TxSession) Insert

func (s *TxSession) Insert(ctx context.Context, key keys.Key, data *internal.TableData) error

func (*TxSession) Read

func (s *TxSession) Read(ctx context.Context, key keys.Key) (kv.Iterator, error)

func (*TxSession) ReadRange

func (s *TxSession) ReadRange(ctx context.Context, lKey keys.Key, rKey keys.Key, isSnapshot bool) (kv.Iterator, error)

func (*TxSession) Replace

func (s *TxSession) Replace(ctx context.Context, key keys.Key, data *internal.TableData, isUpdate bool) error

func (*TxSession) Rollback

func (s *TxSession) Rollback(ctx context.Context) error

func (*TxSession) SetVersionstampedKey

func (s *TxSession) SetVersionstampedKey(ctx context.Context, key []byte, value []byte) error

func (*TxSession) SetVersionstampedValue

func (s *TxSession) SetVersionstampedValue(ctx context.Context, key []byte, value []byte) error

func (*TxSession) Update

func (s *TxSession) Update(ctx context.Context, key keys.Key, apply func(*internal.TableData) (*internal.TableData, error)) (int32, error)