/
transaction.go
118 lines (87 loc) · 2.15 KB
/
transaction.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
// Package gorm is an implementation of trm.Transaction interface by Transaction for *gorm.DB.
package gorm
import (
"context"
"database/sql"
"errors"
"sync"
"sync/atomic"
"gorm.io/gorm"
"github.com/avito-tech/go-transaction-manager/trm/v2"
)
var errRollbackTx = errors.New("rollback transaction")
// Transaction is trm.Transaction for sqlx.Tx.
type Transaction struct {
tx *gorm.DB
err chan error
isActive int64
}
// NewTransaction creates trm.Transaction for sqlx.Tx.
func NewTransaction(
ctx context.Context,
opts *sql.TxOptions,
db *gorm.DB,
) (context.Context, *Transaction, error) {
tr := &Transaction{isActive: 1, err: make(chan error), tx: nil}
var err error
wg := sync.WaitGroup{}
wg.Add(1)
go func() {
db = db.WithContext(ctx)
err = db.Transaction(func(tx *gorm.DB) error {
tr.tx = tx
wg.Done()
return <-tr.err
}, opts)
if tr.tx != nil {
tr.err <- err
} else {
wg.Done()
}
}()
wg.Wait()
if err != nil {
return ctx, nil, err
}
go tr.awaitDone(ctx)
return ctx, tr, nil
}
func (t *Transaction) awaitDone(ctx context.Context) {
if ctx.Done() == nil {
return
}
<-ctx.Done()
t.deactivate()
}
// Transaction returns the real transaction sqlx.Tx.
// trm.NestedTrFactory returns IsActive as true while trm.Transaction is opened.
func (t *Transaction) Transaction() interface{} {
return t.tx
}
// Begin nested transaction by save point.
func (t *Transaction) Begin(ctx context.Context, s trm.Settings) (context.Context, trm.Transaction, error) {
return NewDefaultFactory(t.tx)(ctx, s)
}
// Commit closes the trm.Transaction.
func (t *Transaction) Commit(_ context.Context) error {
defer t.deactivate()
t.err <- nil
return <-t.err
}
// Rollback the trm.Transaction.
func (t *Transaction) Rollback(_ context.Context) error {
defer t.deactivate()
t.err <- errRollbackTx
err := <-t.err
if errors.Is(err, errRollbackTx) {
return nil
}
return err
}
// IsActive returns true if the transaction started but not committed or rolled back.
func (t *Transaction) IsActive() bool {
return atomic.LoadInt64(&t.isActive) == 1
}
func (t *Transaction) deactivate() {
atomic.SwapInt64(&t.isActive, 0)
}