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
|
//go:build tarantool
// +build tarantool
package tarantool
import (
"context"
"fmt"
"reflect"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/tarantool/go-tarantool/v2/crud"
"github.com/tarantool/go-tarantool/v2/pool"
)
const (
tarantoolKVSpaceName = "key_value"
)
func (store *TarantoolStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
var operations = []crud.Operation{
{
Operator: crud.Insert,
Field: "value",
Value: string(value),
},
}
req := crud.MakeUpsertRequest(tarantoolKVSpaceName).
Tuple([]interface{}{string(key), nil, string(value)}).
Operations(operations)
ret := crud.Result{}
if err := store.pool.Do(req, pool.RW).GetTyped(&ret); err != nil {
return fmt.Errorf("kv put: %w", err)
}
return nil
}
func (store *TarantoolStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
getOpts := crud.GetOpts{
Fields: crud.MakeOptTuple([]interface{}{"value"}),
Mode: crud.MakeOptString("read"),
PreferReplica: crud.MakeOptBool(true),
Balance: crud.MakeOptBool(true),
}
req := crud.MakeGetRequest(tarantoolKVSpaceName).
Key(crud.Tuple([]interface{}{string(key)})).
Opts(getOpts)
resp := crud.Result{}
err = store.pool.Do(req, pool.PreferRO).GetTyped(&resp)
if err != nil {
return nil, err
}
results, ok := resp.Rows.([]interface{})
if !ok || len(results) != 1 {
return nil, filer.ErrKvNotFound
}
rows, ok := results[0].([]interface{})
if !ok || len(rows) != 1 {
return nil, filer.ErrKvNotFound
}
row, ok := rows[0].(string)
if !ok {
return nil, fmt.Errorf("Can't convert rows[0] field to string. Actual type: %v, value: %v", reflect.TypeOf(rows[0]), rows[0])
}
return []byte(row), nil
}
func (store *TarantoolStore) KvDelete(ctx context.Context, key []byte) (err error) {
delOpts := crud.DeleteOpts{
Noreturn: crud.MakeOptBool(true),
}
req := crud.MakeDeleteRequest(tarantoolKVSpaceName).
Key(crud.Tuple([]interface{}{string(key)})).
Opts(delOpts)
if _, err := store.pool.Do(req, pool.RW).Get(); err != nil {
return fmt.Errorf("kv delete: %w", err)
}
return nil
}
|
