forked from dolthub/vitess
/
file.go
129 lines (115 loc) · 3.57 KB
/
file.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
119
120
121
122
123
124
125
126
127
128
129
package etcd2topo
import (
"path"
"github.com/coreos/etcd/clientv3"
"golang.org/x/net/context"
"github.com/youtube/vitess/go/vt/topo"
)
// Create is part of the topo.Backend interface.
func (s *Server) Create(ctx context.Context, cell, filePath string, contents []byte) (topo.Version, error) {
c, err := s.clientForCell(ctx, cell)
if err != nil {
return nil, err
}
nodePath := path.Join(c.root, filePath)
// We have to do a transaction, comparing existing version with 0.
// This means: if the file doesn't exist, create it.
txnresp, err := c.cli.Txn(ctx).
If(clientv3.Compare(clientv3.Version(nodePath), "=", 0)).
Then(clientv3.OpPut(nodePath, string(contents))).
Commit()
if err != nil {
return nil, convertError(err)
}
if !txnresp.Succeeded {
return nil, topo.ErrNodeExists
}
return EtcdVersion(txnresp.Header.Revision), nil
}
// Update is part of the topo.Backend interface.
func (s *Server) Update(ctx context.Context, cell, filePath string, contents []byte, version topo.Version) (topo.Version, error) {
c, err := s.clientForCell(ctx, cell)
if err != nil {
return nil, err
}
nodePath := path.Join(c.root, filePath)
if version != nil {
// We have to do a transaction. This means: if the
// current file revision is what we expect, save it.
txnresp, err := c.cli.Txn(ctx).
If(clientv3.Compare(clientv3.ModRevision(nodePath), "=", int64(version.(EtcdVersion)))).
Then(clientv3.OpPut(nodePath, string(contents))).
Commit()
if err != nil {
return nil, convertError(err)
}
if !txnresp.Succeeded {
return nil, topo.ErrBadVersion
}
return EtcdVersion(txnresp.Header.Revision), nil
}
// No version specified. We can use a simple unconditional Put.
resp, err := c.cli.Put(ctx, nodePath, string(contents))
if err != nil {
return nil, convertError(err)
}
return EtcdVersion(resp.Header.Revision), nil
}
// Get is part of the topo.Backend interface.
func (s *Server) Get(ctx context.Context, cell, filePath string) ([]byte, topo.Version, error) {
c, err := s.clientForCell(ctx, cell)
if err != nil {
return nil, nil, err
}
nodePath := path.Join(c.root, filePath)
resp, err := c.cli.Get(ctx, nodePath)
if err != nil {
return nil, nil, convertError(err)
}
if len(resp.Kvs) != 1 {
return nil, nil, topo.ErrNoNode
}
return resp.Kvs[0].Value, EtcdVersion(resp.Kvs[0].ModRevision), nil
}
// Delete is part of the topo.Backend interface.
func (s *Server) Delete(ctx context.Context, cell, filePath string, version topo.Version) error {
c, err := s.clientForCell(ctx, cell)
if err != nil {
return err
}
nodePath := path.Join(c.root, filePath)
if version != nil {
// We have to do a transaction. This means: if the
// node revision is what we expect, delete it,
// otherwise get the file. If the transaction doesn't
// succeed, we also ask for the value of the
// node. That way we'll know if it failed because it
// didn't exist, or because the version was wrong.
txnresp, err := c.cli.Txn(ctx).
If(clientv3.Compare(clientv3.ModRevision(nodePath), "=", int64(version.(EtcdVersion)))).
Then(clientv3.OpDelete(nodePath)).
Else(clientv3.OpGet(nodePath)).
Commit()
if err != nil {
return convertError(err)
}
if !txnresp.Succeeded {
if len(txnresp.Responses) > 0 {
if len(txnresp.Responses[0].GetResponseRange().Kvs) > 0 {
return topo.ErrBadVersion
}
}
return topo.ErrNoNode
}
return nil
}
// This is just a regular unconditional Delete here.
resp, err := c.cli.Delete(ctx, nodePath)
if err != nil {
return convertError(err)
}
if resp.Deleted != 1 {
return topo.ErrNoNode
}
return nil
}