~tsileo/blobstash

ref: b178995e346d blobstash/pkg/client/kvstore/kvstore.go -rw-r--r-- 2.9 KiB
b178995eThomas Sileo vendor: update deps 2 years ago
                                                                                
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
package kvstore // import "a4.io/blobstash/pkg/client/kvstore"

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"net/url"
	"strconv"

	"a4.io/blobstash/pkg/client/clientutil"
	"a4.io/blobstash/pkg/client/response"
)

var ErrKeyNotFound = errors.New("key doest not exist")

func nextKey(key string) string {
	bkey := []byte(key)
	i := len(bkey)
	for i > 0 {
		i--
		bkey[i]++
		if bkey[i] != 0 {
			break
		}
	}
	return string(bkey)
}

type KvStore struct {
	client *clientutil.ClientUtil
}

func New(c *clientutil.ClientUtil) *KvStore {
	return &KvStore{c}
}

func (kvs *KvStore) Put(ctx context.Context, key, ref string, pdata []byte, version int) (*response.KeyValue, error) {
	data := url.Values{}
	data.Set("data", string(pdata))
	data.Set("ref", ref)
	if version != -1 {
		data.Set("version", strconv.Itoa(version))
	}
	resp, err := kvs.client.Post("/api/kvstore/key/"+key, []byte(data.Encode()))
	if err != nil {
		return nil, err
	}

	defer resp.Body.Close()

	if err := clientutil.ExpectStatusCode(resp, http.StatusOK); err != nil {
		return nil, err
	}

	// TODO(tsileo): better place than response?
	kv := &response.KeyValue{}
	if err := clientutil.Unmarshal(resp, kv); err != nil {
		return nil, err
	}

	return kv, nil
}

func (kvs *KvStore) Get(ctx context.Context, key string, version int) (*response.KeyValue, error) {
	resp, err := kvs.client.Get(fmt.Sprintf("/api/kvstore/key/%s?version=%v", key, version))
	if err != nil {
		return nil, err
	}

	defer resp.Body.Close()

	if err := clientutil.ExpectStatusCode(resp, 200); err != nil {
		if err.IsNotFound() {
			return nil, ErrKeyNotFound
		}
		return nil, err
	}

	kv := &response.KeyValue{}
	if err := clientutil.Unmarshal(resp, kv); err != nil {
		return nil, err
	}

	return kv, nil
}

func (kvs *KvStore) Versions(ctx context.Context, key string, start, end, limit int) (*response.KeyValueVersions, error) {
	// TODO handle start, end and limit
	resp, err := kvs.client.Get(fmt.Sprintf("/api/kvstore/key/%s/_versions?start=%d&end=%d&limit=%d", key, start, end, limit))
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	if err := clientutil.ExpectStatusCode(resp, http.StatusOK); err != nil {
		if err.IsNotFound() {
			return nil, ErrKeyNotFound
		}
		return nil, err
	}

	kvversions := &response.KeyValueVersions{}
	if err := clientutil.Unmarshal(resp, kvversions); err != nil {
		return nil, err
	}
	return kvversions, nil
}

func (kvs *KvStore) Keys(ctx context.Context, prefix, start, end string, limit int) ([]*response.KeyValue, error) {
	resp, err := kvs.client.Get(fmt.Sprintf("/api/kvstore/keys?prefix=%s&start=%s&end=%s&limit=%d", prefix, start, end, limit))
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	if err := clientutil.ExpectStatusCode(resp, http.StatusOK); err != nil {
		return nil, err
	}

	keys := &response.KeysResponse{}
	if err := clientutil.Unmarshal(resp, keys); err != nil {
		return nil, err
	}

	return keys.Keys, nil
}