V1 transaction options

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/golang/mock v1.6.0
 	github.com/jinzhu/copier v0.3.5
 	github.com/jmoiron/sqlx v1.3.5
-	github.com/mattn/go-sqlite3 v1.14.15
+	github.com/mattn/go-sqlite3 v1.14.16
 	github.com/stretchr/testify v1.8.1
 	go.mongodb.org/mongo-driver v1.10.3
 	go.uber.org/multierr v1.8.0

go.sum

@@ -25,8 +25,8 @@ github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/lib/pq v1.2.0 h1:LXpIM/LZ5xGFhOpXAQUIMM1HdyqzVYM13zNdjCEEcA0=
 github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/mattn/go-sqlite3 v1.14.6/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
-github.com/mattn/go-sqlite3 v1.14.15 h1:vfoHhTN1af61xCRSWzFIWzx2YskyMTwHLrExkBOjvxI=
-github.com/mattn/go-sqlite3 v1.14.15/go.mod h1:2eHXhiwb8IkHr+BDWZGa96P6+rkvnG63S2DGjv9HUNg=
+github.com/mattn/go-sqlite3 v1.14.16 h1:yOQRA0RpS5PFz/oikGwBEqvAWhWg5ufRz4ETLjwpU1Y=
+github.com/mattn/go-sqlite3 v1.14.16/go.mod h1:2eHXhiwb8IkHr+BDWZGa96P6+rkvnG63S2DGjv9HUNg=
 github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe h1:iruDEfMl2E6fbMZ9s0scYfZQ84/6SPL6zC8ACM2oIL0=
 github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=

mongo/contract.go

@@ -0,0 +1,23 @@
+package mongo
+
+import (
+	"context"
+
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	"go.mongodb.org/mongo-driver/mongo/readpref"
+)
+
+type client interface {
+	Connect(ctx context.Context) error
+	Disconnect(ctx context.Context) error
+	Ping(ctx context.Context, rp *readpref.ReadPref) error
+	StartSession(opts ...*options.SessionOptions) (mongo.Session, error)
+	Database(name string, opts ...*options.DatabaseOptions) *mongo.Database
+	ListDatabases(ctx context.Context, filter interface{}, opts ...*options.ListDatabasesOptions) (mongo.ListDatabasesResult, error)
+	ListDatabaseNames(ctx context.Context, filter interface{}, opts ...*options.ListDatabasesOptions) ([]string, error)
+	UseSession(ctx context.Context, fn func(sessionContext mongo.SessionContext) error) error
+	UseSessionWithOptions(ctx context.Context, opts *options.SessionOptions, fn func(sessionContext mongo.SessionContext) error) error
+	Watch(ctx context.Context, pipeline interface{}, opts ...*options.ChangeStreamOptions) (*mongo.ChangeStream, error)
+	NumberSessionsInProgress() int
+}

mongo/factory.go

@@ -3,15 +3,14 @@ package mongo
 import (
 	"context"
 
-	"go.mongodb.org/mongo-driver/mongo"
-
 	"github.com/avito-tech/go-transaction-manager/transaction"
 )
 
 // NewDefaultFactory creates default transaction.Transaction(mongo.Session).
-// TODO add options.
-func NewDefaultFactory(client *mongo.Client) transaction.TrFactory {
-	return func(ctx context.Context, s transaction.Settings) (context.Context, transaction.Transaction, error) {
-		return NewTransaction(ctx, nil, nil, client)
+func NewDefaultFactory(client client) transaction.TrFactory {
+	return func(ctx context.Context, trms transaction.Settings) (context.Context, transaction.Transaction, error) {
+		s, _ := trms.(Settings)
+
+		return NewTransaction(ctx, s.SessionOpts(), s.TransactionOpts(), client)
 	}
 }

mongo/settings.go

@@ -0,0 +1,82 @@
+package mongo
+
+import (
+	"go.mongodb.org/mongo-driver/mongo/options"
+
+	"github.com/avito-tech/go-transaction-manager/transaction"
+)
+
+// Opt is a type to configure Settings.
+type Opt func(s *Settings)
+
+// WithSessionOpts sets up options.SessionOptions for the Settings.
+func WithSessionOpts(opts *options.SessionOptions) Opt {
+	return func(s *Settings) {
+		*s = s.setSessionOpts(opts)
+	}
+}
+
+// WithTransactionOpts sets up options.TransactionOptions for the Settings.
+func WithTransactionOpts(opts *options.TransactionOptions) Opt {
+	return func(s *Settings) {
+		*s = s.setTransactionOpts(opts)
+	}
+}
+
+// Settings contains settings for mongo.Transaction.
+type Settings struct {
+	transaction.Settings
+	sessionOpts     *options.SessionOptions
+	transactionOpts *options.TransactionOptions
+}
+
+// NewSettings creates Settings.
+func NewSettings(trms transaction.Settings, oo ...Opt) Settings {
+	s := &Settings{Settings: trms}
+
+	for _, o := range oo {
+		o(s)
+	}
+
+	return *s
+}
+
+//revive:disable:exported
+func (s Settings) EnrichBy(in transaction.Settings) (res transaction.Settings) { //nolint:ireturn,nolintlint
+	external, ok := in.(Settings)
+	if ok {
+		if s.SessionOpts() == nil {
+			s = s.setSessionOpts(external.SessionOpts())
+		}
+
+		if s.TransactionOpts() == nil {
+			s = s.setTransactionOpts(external.TransactionOpts())
+		}
+	}
+
+	s.Settings = s.Settings.EnrichBy(in)
+
+	return s
+}
+
+// SessionOpts returns *options.SessionOptions for the transaction.Transaction.
+func (s Settings) SessionOpts() *options.SessionOptions {
+	return s.sessionOpts
+}
+
+func (s Settings) setSessionOpts(opts *options.SessionOptions) Settings {
+	s.sessionOpts = opts
+
+	return s
+}
+
+// TransactionOpts returns transaction.CtxKey for the transaction.Transaction.
+func (s Settings) TransactionOpts() *options.TransactionOptions {
+	return s.transactionOpts
+}
+
+func (s Settings) setTransactionOpts(opts *options.TransactionOptions) Settings {
+	s.transactionOpts = opts
+
+	return s
+}

mongo/settings_test.go

@@ -0,0 +1,96 @@
+package mongo
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	"go.mongodb.org/mongo-driver/mongo/readconcern"
+
+	"github.com/avito-tech/go-transaction-manager/transaction"
+	"github.com/avito-tech/go-transaction-manager/transaction/settings"
+)
+
+func TestSettings_EnrichBy(t *testing.T) {
+	t.Parallel()
+
+	type args struct {
+		external transaction.Settings
+	}
+
+	tests := map[string]struct {
+		settings Settings
+		args     args
+		want     transaction.Settings
+	}{
+		"update_default": {
+			settings: NewSettings(settings.New()),
+			args: args{
+				external: NewSettings(
+					settings.New(settings.WithCancelable(true)),
+					WithSessionOpts((&options.SessionOptions{}).
+						SetCausalConsistency(true)),
+					WithTransactionOpts((&options.TransactionOptions{}).
+						SetReadConcern(readconcern.Majority())),
+				),
+			},
+			want: NewSettings(
+				settings.New(settings.WithCancelable(true)),
+				WithSessionOpts((&options.SessionOptions{}).
+					SetCausalConsistency(true)),
+				WithTransactionOpts((&options.TransactionOptions{}).
+					SetReadConcern(readconcern.Majority())),
+			),
+		},
+		"without_update": {
+			settings: NewSettings(
+				settings.New(settings.WithCancelable(true)),
+				WithSessionOpts((&options.SessionOptions{}).
+					SetCausalConsistency(true)),
+				WithTransactionOpts((&options.TransactionOptions{}).
+					SetReadConcern(readconcern.Majority())),
+			),
+			args: args{
+				external: NewSettings(
+					settings.New(settings.WithCancelable(false)),
+					WithSessionOpts((&options.SessionOptions{}).
+						SetCausalConsistency(false)),
+					WithTransactionOpts((&options.TransactionOptions{}).
+						SetReadConcern(readconcern.Local())),
+				),
+			},
+			want: NewSettings(
+				settings.New(settings.WithCancelable(true)),
+				WithSessionOpts((&options.SessionOptions{}).
+					SetCausalConsistency(true)),
+				WithTransactionOpts((&options.TransactionOptions{}).
+					SetReadConcern(readconcern.Majority())),
+			),
+		},
+		"update_only_transaction.Settings": {
+			settings: NewSettings(
+				settings.New(),
+				WithSessionOpts((&options.SessionOptions{}).
+					SetCausalConsistency(true)),
+			),
+			args: args{
+				external: settings.New(settings.WithCancelable(true)),
+			},
+			want: NewSettings(
+				settings.New(settings.WithCancelable(true)),
+				WithSessionOpts((&options.SessionOptions{}).
+					SetCausalConsistency(true)),
+			),
+		},
+	}
+	for name, tt := range tests {
+		tt := tt
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			got := tt.settings.EnrichBy(tt.args.external)
+
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}

mongo/transaction.go

@@ -20,14 +20,16 @@ func NewTransaction(
 	ctx context.Context,
 	sessionOptions *options.SessionOptions,
 	trOpts *options.TransactionOptions,
-	client *mongo.Client,
+	client client,
 ) (context.Context, *Transaction, error) {
 	s, err := client.StartSession(sessionOptions)
 	if err != nil {
 		return ctx, nil, err
 	}
 
 	if err = s.StartTransaction(trOpts); err != nil {
+		defer s.EndSession(ctx)
+
 		return ctx, nil, err
 	}
 
@@ -66,7 +68,7 @@ func (t *Transaction) Commit(ctx context.Context) error {
 	return nil
 }
 
-// Rollback the transaction.Transaction..
+// Rollback the transaction.Transaction.
 func (t *Transaction) Rollback(ctx context.Context) error {
 	t.deactivate()