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
|
// Copyright (C) MongoDB, Inc. 2017-present.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License. You may obtain
// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
package unified
import (
"context"
"fmt"
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
)
func executeAbortTransaction(ctx context.Context, operation *operation) (*operationResult, error) {
sess, err := entities(ctx).session(operation.Object)
if err != nil {
return nil, err
}
// AbortTransaction takes no options, so the arguments doc must be nil or empty.
elems, _ := operation.Arguments.Elements()
if len(elems) > 0 {
return nil, fmt.Errorf("unrecognized abortTransaction options %v", operation.Arguments)
}
return newErrorResult(sess.AbortTransaction(ctx)), nil
}
func executeEndSession(ctx context.Context, operation *operation) error {
sess, err := entities(ctx).session(operation.Object)
if err != nil {
return err
}
// EnsSession takes no options, so the arguments doc must be nil or empty.
elems, _ := operation.Arguments.Elements()
if len(elems) > 0 {
return fmt.Errorf("unrecognized endSession options %v", operation.Arguments)
}
sess.EndSession(ctx)
return nil
}
func executeCommitTransaction(ctx context.Context, operation *operation) (*operationResult, error) {
sess, err := entities(ctx).session(operation.Object)
if err != nil {
return nil, err
}
// CommitTransaction takes no options, so the arguments doc must be nil or empty.
elems, _ := operation.Arguments.Elements()
if len(elems) > 0 {
return nil, fmt.Errorf("unrecognized commitTransaction options %v", operation.Arguments)
}
return newErrorResult(sess.CommitTransaction(ctx)), nil
}
func executeStartTransaction(ctx context.Context, operation *operation) (*operationResult, error) {
sess, err := entities(ctx).session(operation.Object)
if err != nil {
return nil, err
}
opts := options.Transaction()
if operation.Arguments != nil {
var temp transactionOptions
if err := bson.Unmarshal(operation.Arguments, &temp); err != nil {
return nil, fmt.Errorf("error unmarshalling arguments to transactionOptions: %v", err)
}
opts = temp.TransactionOptions
}
return newErrorResult(sess.StartTransaction(opts)), nil
}
func executeWithTransaction(ctx context.Context, op *operation, loopDone <-chan struct{}) error {
sess, err := entities(ctx).session(op.Object)
if err != nil {
return err
}
// Process the "callback" argument. This is an array of operation objects, each of which should be executed inside
// the transaction.
callback, err := op.Arguments.LookupErr("callback")
if err != nil {
return newMissingArgumentError("callback")
}
var operations []*operation
if err := callback.Unmarshal(&operations); err != nil {
return fmt.Errorf("error transforming callback option to slice of operations: %v", err)
}
// Remove the "callback" field and process the other options.
var temp transactionOptions
if err := bson.Unmarshal(removeFieldsFromDocument(op.Arguments, "callback"), &temp); err != nil {
return fmt.Errorf("error unmarshalling arguments to transactionOptions: %v", err)
}
_, err = sess.WithTransaction(ctx, func(sessCtx mongo.SessionContext) (interface{}, error) {
for idx, oper := range operations {
if err := oper.execute(ctx, loopDone); err != nil {
return nil, fmt.Errorf("error executing operation %q at index %d: %v", oper.Name, idx, err)
}
}
return nil, nil
}, temp.TransactionOptions)
return err
}
|
