@cryptotaxi247 / kubo / commits / 137c0ac4a

use batching transaction interface from datastore

License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>

Jeromy committed Jun 26, 2015 at 09:44 UTC 137c0ac4ac5591f5011667dd413e0bd4aab5c3ba
39 files changed +857 -1546
Godeps/Godeps.json
+1 -6
@@ -87,11 +87,6 @@
87 "ImportPath": "github.com/fd/go-nat",
88 "Rev": "50e7633d5f27d81490026a13e5b92d2e42d8c6bb"
89 },
90 - {
91 - "ImportPath": "github.com/fzzy/radix/redis",
92 - "Comment": "v0.5.1",
93 - "Rev": "27a863cdffdb0998d13e1e11992b18489aeeaa25"
94 - },
90 {
91 "ImportPath": "github.com/gogo/protobuf/io",
92 "Rev": "0ac967c269268f1af7d9bcc7927ccc9a589b2b36"
@@ -139,7 +134,7 @@
134 },
135 {
136 "ImportPath": "github.com/jbenet/go-datastore",
142 - "Rev": "245a981af3750d7710db13dca731ba8461aa1095"
137 + "Rev": "7d6acaf7c0164c335f2ca4100f8fe30a7e2943dd"
138 },
139 {
140 "ImportPath": "github.com/jbenet/go-detect-race",
Godeps/_workspace/src/github.com/fzzy/radix/redis/client.go deleted
-244
@@ -1,244 +0,0 @@
1 -package redis
2 -
3 -import (
4 - "bufio"
5 - "errors"
6 - "net"
7 - "strings"
8 - "time"
9 -
10 - "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/fzzy/radix/redis/resp"
11 -)
12 -
13 -const (
14 - bufSize int = 4096
15 -)
16 -
17 -//* Common errors
18 -
19 -var LoadingError error = errors.New("server is busy loading dataset in memory")
20 -var PipelineQueueEmptyError error = errors.New("pipeline queue empty")
21 -
22 -//* Client
23 -
24 -// Client describes a Redis client.
25 -type Client struct {
26 - // The connection the client talks to redis over. Don't touch this unless
27 - // you know what you're doing.
28 - Conn net.Conn
29 - timeout time.Duration
30 - reader *bufio.Reader
31 - pending []*request
32 - completed []*Reply
33 -}
34 -
35 -// request describes a client's request to the redis server
36 -type request struct {
37 - cmd string
38 - args []interface{}
39 -}
40 -
41 -// Dial connects to the given Redis server with the given timeout, which will be
42 -// used as the read/write timeout when communicating with redis
43 -func DialTimeout(network, addr string, timeout time.Duration) (*Client, error) {
44 - // establish a connection
45 - conn, err := net.Dial(network, addr)
46 - if err != nil {
47 - return nil, err
48 - }
49 -
50 - c := new(Client)
51 - c.Conn = conn
52 - c.timeout = timeout
53 - c.reader = bufio.NewReaderSize(conn, bufSize)
54 - return c, nil
55 -}
56 -
57 -// Dial connects to the given Redis server.
58 -func Dial(network, addr string) (*Client, error) {
59 - return DialTimeout(network, addr, time.Duration(0))
60 -}
61 -
62 -//* Public methods
63 -
64 -// Close closes the connection.
65 -func (c *Client) Close() error {
66 - return c.Conn.Close()
67 -}
68 -
69 -// Cmd calls the given Redis command.
70 -func (c *Client) Cmd(cmd string, args ...interface{}) *Reply {
71 - err := c.writeRequest(&request{cmd, args})
72 - if err != nil {
73 - return &Reply{Type: ErrorReply, Err: err}
74 - }
75 - return c.ReadReply()
76 -}
77 -
78 -// Append adds the given call to the pipeline queue.
79 -// Use GetReply() to read the reply.
80 -func (c *Client) Append(cmd string, args ...interface{}) {
81 - c.pending = append(c.pending, &request{cmd, args})
82 -}
83 -
84 -// GetReply returns the reply for the next request in the pipeline queue.
85 -// Error reply with PipelineQueueEmptyError is returned,
86 -// if the pipeline queue is empty.
87 -func (c *Client) GetReply() *Reply {
88 - if len(c.completed) > 0 {
89 - r := c.completed[0]
90 - c.completed = c.completed[1:]
91 - return r
92 - }
93 - c.completed = nil
94 -
95 - if len(c.pending) == 0 {
96 - return &Reply{Type: ErrorReply, Err: PipelineQueueEmptyError}
97 - }
98 -
99 - nreqs := len(c.pending)
100 - err := c.writeRequest(c.pending...)
101 - c.pending = nil
102 - if err != nil {
103 - return &Reply{Type: ErrorReply, Err: err}
104 - }
105 - r := c.ReadReply()
106 - c.completed = make([]*Reply, nreqs-1)
107 - for i := 0; i < nreqs-1; i++ {
108 - c.completed[i] = c.ReadReply()
109 - }
110 -
111 - return r
112 -}
113 -
114 -//* Private methods
115 -
116 -func (c *Client) setReadTimeout() {
117 - if c.timeout != 0 {
118 - c.Conn.SetReadDeadline(time.Now().Add(c.timeout))
119 - }
120 -}
121 -
122 -func (c *Client) setWriteTimeout() {
123 - if c.timeout != 0 {
124 - c.Conn.SetWriteDeadline(time.Now().Add(c.timeout))
125 - }
126 -}
127 -
128 -// This will read a redis reply off of the connection without sending anything
129 -// first (useful after you've sent a SUSBSCRIBE command). This will block until
130 -// a reply is received or the timeout is reached. On timeout an ErrorReply will
131 -// be returned, you can check if it's a timeout like so:
132 -//
133 -// r := conn.ReadReply()
134 -// if r.Err != nil {
135 -// if t, ok := r.Err.(*net.OpError); ok && t.Timeout() {
136 -// // Is timeout
137 -// } else {
138 -// // Not timeout
139 -// }
140 -// }
141 -//
142 -// Note: this is a more low-level function, you really shouldn't have to
143 -// actually use it unless you're writing your own pub/sub code
144 -func (c *Client) ReadReply() *Reply {
145 - c.setReadTimeout()
146 - return c.parse()
147 -}
148 -
149 -func (c *Client) writeRequest(requests ...*request) error {
150 - c.setWriteTimeout()
151 - for i := range requests {
152 - req := make([]interface{}, 0, len(requests[i].args)+1)
153 - req = append(req, requests[i].cmd)
154 - req = append(req, requests[i].args...)
155 - err := resp.WriteArbitraryAsFlattenedStrings(c.Conn, req)
156 - if err != nil {
157 - c.Close()
158 - return err
159 - }
160 - }
161 - return nil
162 -}
163 -
164 -func (c *Client) parse() *Reply {
165 - m, err := resp.ReadMessage(c.reader)
166 - if err != nil {
167 - if t, ok := err.(*net.OpError); !ok || !t.Timeout() {
168 - // close connection except timeout
169 - c.Close()
170 - }
171 - return &Reply{Type: ErrorReply, Err: err}
172 - }
173 - r, err := messageToReply(m)
174 - if err != nil {
175 - return &Reply{Type: ErrorReply, Err: err}
176 - }
177 - return r
178 -}
179 -
180 -// The error return parameter is for bubbling up parse errors and the like, if
181 -// the error is sent by redis itself as an Err message type, then it will be
182 -// sent back as an actual Reply (wrapped in a CmdError)
183 -func messageToReply(m *resp.Message) (*Reply, error) {
184 - r := &Reply{}
185 -
186 - switch m.Type {
187 - case resp.Err:
188 - errMsg, err := m.Err()
189 - if err != nil {
190 - return nil, err
191 - }
192 - if strings.HasPrefix(errMsg.Error(), "LOADING") {
193 - err = LoadingError
194 - } else {
195 - err = &CmdError{errMsg}
196 - }
197 - r.Type = ErrorReply
198 - r.Err = err
199 -
200 - case resp.SimpleStr:
201 - status, err := m.Bytes()
202 - if err != nil {
203 - return nil, err
204 - }
205 - r.Type = StatusReply
206 - r.buf = status
207 -
208 - case resp.Int:
209 - i, err := m.Int()
210 - if err != nil {
211 - return nil, err
212 - }
213 - r.Type = IntegerReply
214 - r.int = i
215 -
216 - case resp.BulkStr:
217 - b, err := m.Bytes()
218 - if err != nil {
219 - return nil, err
220 - }
221 - r.Type = BulkReply
222 - r.buf = b
223 -
224 - case resp.Nil:
225 - r.Type = NilReply
226 -
227 - case resp.Array:
228 - ms, err := m.Array()
229 - if err != nil {
230 - return nil, err
231 - }
232 - r.Type = MultiReply
233 - r.Elems = make([]*Reply, len(ms))
234 - for i := range ms {
235 - r.Elems[i], err = messageToReply(ms[i])
236 - if err != nil {
237 - return nil, err
238 - }
239 - }
240 - }
241 -
242 - return r, nil
243 -
244 -}
Godeps/_workspace/src/github.com/fzzy/radix/redis/client_test.go deleted
-106
@@ -1,106 +0,0 @@
1 -package redis
2 -
3 -import (
4 - "bufio"
5 - "bytes"
6 - "github.com/stretchr/testify/assert"
7 - . "testing"
8 - "time"
9 -)
10 -
11 -func dial(t *T) *Client {
12 - client, err := DialTimeout("tcp", "127.0.0.1:6379", 10*time.Second)
13 - assert.Nil(t, err)
14 - return client
15 -}
16 -
17 -func TestCmd(t *T) {
18 - c := dial(t)
19 - v, _ := c.Cmd("echo", "Hello, World!").Str()
20 - assert.Equal(t, "Hello, World!", v)
21 -
22 - // Test that a bad command properly returns a *CmdError
23 - err := c.Cmd("non-existant-cmd").Err
24 - assert.NotEqual(t, "", err.(*CmdError).Error())
25 -
26 - // Test that application level errors propagate correctly
27 - c.Cmd("sadd", "foo", "bar")
28 - _, err = c.Cmd("get", "foo").Str()
29 - assert.NotEqual(t, "", err.(*CmdError).Error())
30 -}
31 -
32 -func TestPipeline(t *T) {
33 - c := dial(t)
34 - c.Append("echo", "foo")
35 - c.Append("echo", "bar")
36 - c.Append("echo", "zot")
37 -
38 - v, _ := c.GetReply().Str()
39 - assert.Equal(t, "foo", v)
40 -
41 - v, _ = c.GetReply().Str()
42 - assert.Equal(t, "bar", v)
43 -
44 - v, _ = c.GetReply().Str()
45 - assert.Equal(t, "zot", v)
46 -
47 - r := c.GetReply()
48 - assert.Equal(t, ErrorReply, r.Type)
49 - assert.Equal(t, PipelineQueueEmptyError, r.Err)
50 -}
51 -
52 -func TestParse(t *T) {
53 - c := dial(t)
54 -
55 - parseString := func(b string) *Reply {
56 - c.reader = bufio.NewReader(bytes.NewBufferString(b))
57 - return c.parse()
58 - }
59 -
60 - // missing \n trailing
61 - r := parseString("foo")
62 - assert.Equal(t, ErrorReply, r.Type)
63 - assert.NotNil(t, r.Err)
64 -
65 - // error reply
66 - r = parseString("-ERR unknown command 'foobar'\r\n")
67 - assert.Equal(t, ErrorReply, r.Type)
68 - assert.Equal(t, "ERR unknown command 'foobar'", r.Err.Error())
69 -
70 - // LOADING error
71 - r = parseString("-LOADING Redis is loading the dataset in memory\r\n")
72 - assert.Equal(t, ErrorReply, r.Type)
73 - assert.Equal(t, LoadingError, r.Err)
74 -
75 - // status reply
76 - r = parseString("+OK\r\n")
77 - assert.Equal(t, StatusReply, r.Type)
78 - assert.Equal(t, []byte("OK"), r.buf)
79 -
80 - // integer reply
81 - r = parseString(":1337\r\n")
82 - assert.Equal(t, IntegerReply, r.Type)
83 - assert.Equal(t, int64(1337), r.int)
84 -
85 - // null bulk reply
86 - r = parseString("$-1\r\n")
87 - assert.Equal(t, NilReply, r.Type)
88 -
89 - // bulk reply
90 - r = parseString("$6\r\nfoobar\r\n")
91 - assert.Equal(t, BulkReply, r.Type)
92 - assert.Equal(t, []byte("foobar"), r.buf)
93 -
94 - // null multi bulk reply
95 - r = parseString("*-1\r\n")
96 - assert.Equal(t, NilReply, r.Type)
97 -
98 - // multi bulk reply
99 - r = parseString("*5\r\n:0\r\n:1\r\n:2\r\n:3\r\n$6\r\nfoobar\r\n")
100 - assert.Equal(t, MultiReply, r.Type)
101 - assert.Equal(t, 5, len(r.Elems))
102 - for i := 0; i < 4; i++ {
103 - assert.Equal(t, int64(i), r.Elems[i].int)
104 - }
105 - assert.Equal(t, []byte("foobar"), r.Elems[4].buf)
106 -}
Godeps/_workspace/src/github.com/fzzy/radix/redis/doc.go deleted
-87
@@ -1,87 +0,0 @@
1 -// A simple client for connecting and interacting with redis.
2 -//
3 -// To import inside your package do:
4 -//
5 -// import "github.com/fzzy/radix/redis"
6 -//
7 -// Connecting
8 -//
9 -// Use either Dial or DialTimeout:
10 -//
11 -// client, err := redis.Dial("tcp", "localhost:6379")
12 -// if err != nil {
13 -// // handle err
14 -// }
15 -//
16 -// Make sure to call Close on the client if you want to clean it up before the
17 -// end of the program.
18 -//
19 -// Cmd and Reply
20 -//
21 -// The Cmd method returns a Reply, which has methods for converting to various
22 -// types. Each of these methods returns an error which can either be a
23 -// connection error (e.g. timeout), an application error (e.g. key is wrong
24 -// type), or a conversion error (e.g. cannot convert to integer). You can also
25 -// directly check the error using the Err field:
26 -//
27 -// foo, err := client.Cmd("GET", "foo").Str()
28 -// if err != nil {
29 -// // handle err
30 -// }
31 -//
32 -// // Checking Err field directly
33 -//
34 -// err = client.Cmd("PING").Err
35 -// if err != nil {
36 -// // handle err
37 -// }
38 -//
39 -// Multi Replies
40 -//
41 -// The elements to Multi replies can be accessed as strings using List or
42 -// ListBytes, or you can use the Elems field for more low-level access:
43 -//
44 -// r := client.Cmd("MGET", "foo", "bar", "baz")
45 -//
46 -// // This:
47 -// for _, elemStr := range r.List() {
48 -// fmt.Println(elemStr)
49 -// }
50 -//
51 -// // is equivalent to this:
52 -// for i := range r.Elems {
53 -// elemStr, _ := r.Elems[i].Str()
54 -// fmt.Println(elemStr)
55 -// }
56 -//
57 -// Pipelining
58 -//
59 -// Pipelining is when the client sends a bunch of commands to the server at
60 -// once, and only once all the commands have been sent does it start reading the
61 -// replies off the socket. This is supported using the Append and GetReply
62 -// methods. Append will simply append the command to a buffer without sending
63 -// it, the first time GetReply is called it will send all the commands in the
64 -// buffer and return the Reply for the first command that was sent. Subsequent
65 -// calls to GetReply return Replys for subsequent commands:
66 -//
67 -// client.Append("GET", "foo")
68 -// client.Append("SET", "bar", "foo")
69 -// client.Append("DEL", "baz")
70 -//
71 -// // Read GET foo reply
72 -// foo, err := client.GetReply().Str()
73 -// if err != nil {
74 -// // handle err
75 -// }
76 -//
77 -// // Read SET bar foo reply
78 -// if err := client.GetReply().Err; err != nil {
79 -// // handle err
80 -// }
81 -//
82 -// // Read DEL baz reply
83 -// if err := client.GetReply().Err; err != nil {
84 -// // handle err
85 -// }
86 -//
87 -package redis
Godeps/_workspace/src/github.com/fzzy/radix/redis/reply.go deleted
-275
@@ -1,275 +0,0 @@
1 -package redis
2 -
3 -import (
4 - "errors"
5 - "strconv"
6 - "strings"
7 -)
8 -
9 -// A CmdError implements the error interface and is what is returned when the
10 -// server returns an error on the application level (e.g. key doesn't exist or
11 -// is the wrong type), as opposed to a connection/transport error.
12 -//
13 -// You can test if a reply is a CmdError like so:
14 -//
15 -// r := conn.Cmd("GET", "key-which-isnt-a-string")
16 -// if r.Err != nil {
17 -// if cerr, ok := r.Err.(*redis.CmdError); ok {
18 -// // Is CmdError
19 -// } else {
20 -// // Is other error
21 -// }
22 -// }
23 -type CmdError struct {
24 - Err error
25 -}
26 -
27 -func (cerr *CmdError) Error() string {
28 - return cerr.Err.Error()
29 -}
30 -
31 -// Returns true if error returned was due to the redis server being read only
32 -func (cerr *CmdError) Readonly() bool {
33 - return strings.HasPrefix(cerr.Err.Error(), "READONLY")
34 -}
35 -
36 -//* Reply
37 -
38 -/*
39 -ReplyType describes type of a reply.
40 -
41 -Possible values are:
42 -
43 -StatusReply -- status reply
44 -ErrorReply -- error reply
45 -IntegerReply -- integer reply
46 -NilReply -- nil reply
47 -BulkReply -- bulk reply
48 -MultiReply -- multi bulk reply
49 -*/
50 -type ReplyType uint8
51 -
52 -const (
53 - StatusReply ReplyType = iota
54 - ErrorReply
55 - IntegerReply
56 - NilReply
57 - BulkReply
58 - MultiReply
59 -)
60 -
61 -// Reply holds a Redis reply.
62 -type Reply struct {
63 - Type ReplyType // Reply type
64 - Elems []*Reply // Sub-replies
65 - Err error // Reply error
66 - buf []byte
67 - int int64
68 -}
69 -
70 -// Bytes returns the reply value as a byte string or
71 -// an error, if the reply type is not StatusReply or BulkReply.
72 -func (r *Reply) Bytes() ([]byte, error) {
73 - if r.Type == ErrorReply {
74 - return nil, r.Err
75 - }
76 - if !(r.Type == StatusReply || r.Type == BulkReply) {
77 - return nil, errors.New("string value is not available for this reply type")
78 - }
79 -
80 - return r.buf, nil
81 -}
82 -
83 -// Str is a convenience method for calling Reply.Bytes() and converting it to string
84 -func (r *Reply) Str() (string, error) {
85 - b, err := r.Bytes()
86 - if err != nil {
87 - return "", err
88 - }
89 -
90 - return string(b), nil
91 -}
92 -
93 -// Int64 returns the reply value as a int64 or an error,
94 -// if the reply type is not IntegerReply or the reply type
95 -// BulkReply could not be parsed to an int64.
96 -func (r *Reply) Int64() (int64, error) {
97 - if r.Type == ErrorReply {
98 - return 0, r.Err
99 - }
100 - if r.Type != IntegerReply {
101 - s, err := r.Str()
102 - if err == nil {
103 - i64, err := strconv.ParseInt(s, 10, 64)
104 - if err != nil {
105 - return 0, errors.New("failed to parse integer value from string value")
106 - } else {
107 - return i64, nil
108 - }
109 - }
110 -
111 - return 0, errors.New("integer value is not available for this reply type")
112 - }
113 -
114 - return r.int, nil
115 -}
116 -
117 -// Int is a convenience method for calling Reply.Int64() and converting it to int.
118 -func (r *Reply) Int() (int, error) {
119 - i64, err := r.Int64()
120 - if err != nil {
121 - return 0, err
122 - }
123 -
124 - return int(i64), nil
125 -}
126 -
127 -// Bool returns false, if the reply value equals to 0 or "0", otherwise true; or
128 -// an error, if the reply type is not IntegerReply or BulkReply.
129 -func (r *Reply) Bool() (bool, error) {
130 - if r.Type == ErrorReply {
131 - return false, r.Err
132 - }
133 - i, err := r.Int()
134 - if err == nil {
135 - if i == 0 {
136 - return false, nil
137 - }
138 -
139 - return true, nil
140 - }
141 -
142 - s, err := r.Str()
143 - if err == nil {
144 - if s == "0" {
145 - return false, nil
146 - }
147 -
148 - return true, nil
149 - }
150 -
151 - return false, errors.New("boolean value is not available for this reply type")
152 -}
153 -
154 -// List returns a multi bulk reply as a slice of strings or an error.
155 -// The reply type must be MultiReply and its elements' types must all be either BulkReply or NilReply.
156 -// Nil elements are returned as empty strings.
157 -// Useful for list commands.
158 -func (r *Reply) List() ([]string, error) {
159 - // Doing all this in two places instead of just calling ListBytes() so we don't have
160 - // to iterate twice
161 - if r.Type == ErrorReply {
162 - return nil, r.Err
163 - }
164 - if r.Type != MultiReply {
165 - return nil, errors.New("reply type is not MultiReply")
166 - }
167 -
168 - strings := make([]string, len(r.Elems))
169 - for i, v := range r.Elems {
170 - if v.Type == BulkReply {
171 - strings[i] = string(v.buf)
172 - } else if v.Type == NilReply {
173 - strings[i] = ""
174 - } else {
175 - return nil, errors.New("element type is not BulkReply or NilReply")
176 - }
177 - }
178 -
179 - return strings, nil
180 -}
181 -
182 -// ListBytes returns a multi bulk reply as a slice of bytes slices or an error.
183 -// The reply type must be MultiReply and its elements' types must all be either BulkReply or NilReply.
184 -// Nil elements are returned as nil.
185 -// Useful for list commands.
186 -func (r *Reply) ListBytes() ([][]byte, error) {
187 - if r.Type == ErrorReply {
188 - return nil, r.Err
189 - }
190 - if r.Type != MultiReply {
191 - return nil, errors.New("reply type is not MultiReply")
192 - }
193 -
194 - bufs := make([][]byte, len(r.Elems))
195 - for i, v := range r.Elems {
196 - if v.Type == BulkReply {
197 - bufs[i] = v.buf
198 - } else if v.Type == NilReply {
199 - bufs[i] = nil
200 - } else {
201 - return nil, errors.New("element type is not BulkReply or NilReply")
202 - }
203 - }
204 -
205 - return bufs, nil
206 -}
207 -
208 -// Hash returns a multi bulk reply as a map[string]string or an error.
209 -// The reply type must be MultiReply,
210 -// it must have an even number of elements,
211 -// they must be in a "key value key value..." order and
212 -// values must all be either BulkReply or NilReply.
213 -// Nil values are returned as empty strings.
214 -// Useful for hash commands.
215 -func (r *Reply) Hash() (map[string]string, error) {
216 - if r.Type == ErrorReply {
217 - return nil, r.Err
218 - }
219 - rmap := map[string]string{}
220 -
221 - if r.Type != MultiReply {
222 - return nil, errors.New("reply type is not MultiReply")
223 - }
224 -
225 - if len(r.Elems)%2 != 0 {
226 - return nil, errors.New("reply has odd number of elements")
227 - }
228 -
229 - for i := 0; i < len(r.Elems)/2; i++ {
230 - var val string
231 -
232 - key, err := r.Elems[i*2].Str()
233 - if err != nil {
234 - return nil, errors.New("key element has no string reply")
235 - }
236 -
237 - v := r.Elems[i*2+1]
238 - if v.Type == BulkReply {
239 - val = string(v.buf)
240 - rmap[key] = val
241 - } else if v.Type == NilReply {
242 - } else {
243 - return nil, errors.New("value element type is not BulkReply or NilReply")
244 - }
245 - }
246 -
247 - return rmap, nil
248 -}
249 -
250 -// String returns a string representation of the reply and its sub-replies.
251 -// This method is for debugging.
252 -// Use method Reply.Str() for reading string reply.
253 -func (r *Reply) String() string {
254 - switch r.Type {
255 - case ErrorReply:
256 - return r.Err.Error()
257 - case StatusReply:
258 - fallthrough
259 - case BulkReply:
260 - return string(r.buf)
261 - case IntegerReply:
262 - return strconv.FormatInt(r.int, 10)
263 - case NilReply:
264 - return "<nil>"
265 - case MultiReply:
266 - s := "[ "
267 - for _, e := range r.Elems {
268 - s = s + e.String() + " "
269 - }
270 - return s + "]"
271 - }
272 -
273 - // This should never execute
274 - return ""
275 -}
Godeps/_workspace/src/github.com/fzzy/radix/redis/reply_test.go deleted
-125
@@ -1,125 +0,0 @@
1 -package redis
2 -
3 -import (
4 - "github.com/stretchr/testify/assert"
5 - . "testing"
6 -)
7 -
8 -func TestStr(t *T) {
9 - r := &Reply{Type: ErrorReply, Err: LoadingError}
10 - _, err := r.Str()
11 - assert.Equal(t, LoadingError, err)
12 -
13 - r = &Reply{Type: IntegerReply}
14 - _, err = r.Str()
15 - assert.NotNil(t, err)
16 -
17 - r = &Reply{Type: StatusReply, buf: []byte("foo")}
18 - b, err := r.Str()
19 - assert.Nil(t, err)
20 - assert.Equal(t, "foo", b)
21 -
22 - r = &Reply{Type: BulkReply, buf: []byte("foo")}
23 - b, err = r.Str()
24 - assert.Nil(t, err)
25 - assert.Equal(t, "foo", b)
26 -}
27 -
28 -func TestBytes(t *T) {
29 - r := &Reply{Type: BulkReply, buf: []byte("foo")}
30 - b, err := r.Bytes()
31 - assert.Nil(t, err)
32 - assert.Equal(t, []byte("foo"), b)
33 -}
34 -
35 -func TestInt64(t *T) {
36 - r := &Reply{Type: ErrorReply, Err: LoadingError}
37 - _, err := r.Int64()
38 - assert.Equal(t, LoadingError, err)
39 -
40 - r = &Reply{Type: IntegerReply, int: 5}
41 - b, err := r.Int64()
42 - assert.Nil(t, err)
43 - assert.Equal(t, int64(5), b)
44 -
45 - r = &Reply{Type: BulkReply, buf: []byte("5")}
46 - b, err = r.Int64()
47 - assert.Nil(t, err)
48 - assert.Equal(t, int64(5), b)
49 -
50 - r = &Reply{Type: BulkReply, buf: []byte("foo")}
51 - _, err = r.Int64()
52 - assert.NotNil(t, err)
53 -}
54 -
55 -func TestInt(t *T) {
56 - r := &Reply{Type: IntegerReply, int: 5}
57 - b, err := r.Int()
58 - assert.Nil(t, err)
59 - assert.Equal(t, 5, b)
60 -}
61 -
62 -func TestBool(t *T) {
63 - r := &Reply{Type: IntegerReply, int: 0}
64 - b, err := r.Bool()
65 - assert.Nil(t, err)
66 - assert.Equal(t, false, b)
67 -
68 - r = &Reply{Type: StatusReply, buf: []byte("0")}
69 - b, err = r.Bool()
70 - assert.Nil(t, err)
71 - assert.Equal(t, false, b)
72 -
73 - r = &Reply{Type: IntegerReply, int: 2}
74 - b, err = r.Bool()
75 - assert.Nil(t, err)
76 - assert.Equal(t, true, b)
77 -
78 - r = &Reply{Type: NilReply}
79 - _, err = r.Bool()
80 - assert.NotNil(t, err)
81 -}
82 -
83 -func TestList(t *T) {
84 - r := &Reply{Type: MultiReply}
85 - r.Elems = make([]*Reply, 3)
86 - r.Elems[0] = &Reply{Type: BulkReply, buf: []byte("0")}
87 - r.Elems[1] = &Reply{Type: NilReply}
88 - r.Elems[2] = &Reply{Type: BulkReply, buf: []byte("2")}
89 - l, err := r.List()
90 - assert.Nil(t, err)
91 - assert.Equal(t, 3, len(l))
92 - assert.Equal(t, "0", l[0])
93 - assert.Equal(t, "", l[1])
94 - assert.Equal(t, "2", l[2])
95 -}
96 -
97 -func TestBytesList(t *T) {
98 - r := &Reply{Type: MultiReply}
99 - r.Elems = make([]*Reply, 3)
100 - r.Elems[0] = &Reply{Type: BulkReply, buf: []byte("0")}
101 - r.Elems[1] = &Reply{Type: NilReply}
102 - r.Elems[2] = &Reply{Type: BulkReply, buf: []byte("2")}
103 - l, err := r.ListBytes()
104 - assert.Nil(t, err)
105 - assert.Equal(t, 3, len(l))
106 - assert.Equal(t, []byte("0"), l[0])
107 - assert.Nil(t, l[1])
108 - assert.Equal(t, []byte("2"), l[2])
109 -}
110 -
111 -func TestHash(t *T) {
112 - r := &Reply{Type: MultiReply}
113 - r.Elems = make([]*Reply, 6)
114 - r.Elems[0] = &Reply{Type: BulkReply, buf: []byte("a")}
115 - r.Elems[1] = &Reply{Type: BulkReply, buf: []byte("0")}
116 - r.Elems[2] = &Reply{Type: BulkReply, buf: []byte("b")}
117 - r.Elems[3] = &Reply{Type: NilReply}
118 - r.Elems[4] = &Reply{Type: BulkReply, buf: []byte("c")}
119 - r.Elems[5] = &Reply{Type: BulkReply, buf: []byte("2")}
120 - h, err := r.Hash()
121 - assert.Nil(t, err)
122 - assert.Equal(t, "0", h["a"])
123 - assert.Equal(t, "", h["b"])
124 - assert.Equal(t, "2", h["c"])
125 -}
Godeps/_workspace/src/github.com/fzzy/radix/redis/resp/resp.go deleted
-466
@@ -1,466 +0,0 @@
1 -// This package provides an easy to use interface for creating and parsing
2 -// messages encoded in the REdis Serialization Protocol (RESP). You can check
3 -// out more details about the protocol here: http://redis.io/topics/protocol
4 -package resp
5 -
6 -import (
7 - "bufio"
8 - "bytes"
9 - "errors"
10 - "fmt"
11 - "io"
12 - "reflect"
13 - "strconv"
14 -)
15 -
16 -var (
17 - delim = []byte{'\r', '\n'}
18 - delimEnd = delim[len(delim)-1]
19 -)
20 -
21 -type Type int
22 -
23 -const (
24 - SimpleStr Type = iota
25 - Err
26 - Int
27 - BulkStr
28 - Array
29 - Nil
30 -)
31 -
32 -const (
33 - simpleStrPrefix = '+'
34 - errPrefix = '-'
35 - intPrefix = ':'
36 - bulkStrPrefix = '$'
37 - arrayPrefix = '*'
38 -)
39 -
40 -// Parse errors
41 -var (
42 - badType = errors.New("wrong type")
43 - parseErr = errors.New("parse error")
44 -)
45 -
46 -type Message struct {
47 - Type
48 - val interface{}
49 - raw []byte
50 -}
51 -
52 -// NewMessagePParses the given raw message and returns a Message struct
53 -// representing it
54 -func NewMessage(b []byte) (*Message, error) {
55 - return ReadMessage(bytes.NewReader(b))
56 -}
57 -
58 -// Can be used when writing to a resp stream to write a simple-string-style
59 -// stream (e.g. +OK\r\n) instead of the default bulk-string-style strings.
60 -//
61 -// foo := NewSimpleString("foo")
62 -// bar := NewSimpleString("bar")
63 -// baz := NewSimpleString("baz")
64 -// resp.WriteArbitrary(w, foo)
65 -// resp.WriteArbitrary(w, []interface{}{bar, baz})
66 -//
67 -func NewSimpleString(s string) *Message {
68 - b := append(make([]byte, 0, len(s) + 3), '+')
69 - b = append(b, []byte(s)...)
70 - b = append(b, '\r', '\n')
71 - return &Message{
72 - Type: SimpleStr,
73 - val: s,
74 - raw: b,
75 - }
76 -}
77 -
78 -// ReadMessage attempts to read a message object from the given io.Reader, parse
79 -// it, and return a Message struct representing it
80 -func ReadMessage(reader io.Reader) (*Message, error) {
81 - r := bufio.NewReader(reader)
82 - return bufioReadMessage(r)
83 -}
84 -
85 -func bufioReadMessage(r *bufio.Reader) (*Message, error) {
86 - b, err := r.Peek(1)
87 - if err != nil {
88 - return nil, err
89 - }
90 - switch b[0] {
91 - case simpleStrPrefix:
92 - return readSimpleStr(r)
93 - case errPrefix:
94 - return readError(r)
95 - case intPrefix:
96 - return readInt(r)
97 - case bulkStrPrefix:
98 - return readBulkStr(r)
99 - case arrayPrefix:
100 - return readArray(r)
101 - default:
102 - return nil, badType
103 - }
104 -}
105 -
106 -func readSimpleStr(r *bufio.Reader) (*Message, error) {
107 - b, err := r.ReadBytes(delimEnd)
108 - if err != nil {
109 - return nil, err
110 - }
111 - return &Message{Type: SimpleStr, val: b[1 : len(b)-2], raw: b}, nil
112 -}
113 -
114 -func readError(r *bufio.Reader) (*Message, error) {
115 - b, err := r.ReadBytes(delimEnd)
116 - if err != nil {
117 - return nil, err
118 - }
119 - return &Message{Type: Err, val: b[1 : len(b)-2], raw: b}, nil
120 -}
121 -
122 -func readInt(r *bufio.Reader) (*Message, error) {
123 - b, err := r.ReadBytes(delimEnd)
124 - if err != nil {
125 - return nil, err
126 - }
127 - i, err := strconv.ParseInt(string(b[1:len(b)-2]), 10, 64)
128 - if err != nil {
129 - return nil, parseErr
130 - }
131 - return &Message{Type: Int, val: i, raw: b}, nil
132 -}
133 -
134 -func readBulkStr(r *bufio.Reader) (*Message, error) {
135 - b, err := r.ReadBytes(delimEnd)
136 - if err != nil {
137 - return nil, err
138 - }
139 - size, err := strconv.ParseInt(string(b[1:len(b)-2]), 10, 64)
140 - if err != nil {
141 - return nil, parseErr
142 - }
143 - if size < 0 {
144 - return &Message{Type: Nil, raw: b}, nil
145 - }
146 - total := make([]byte, size)
147 - b2 := total
148 - var n int
149 - for len(b2) > 0 {
150 - n, err = r.Read(b2)
151 - if err != nil {
152 - return nil, err
153 - }
154 - b2 = b2[n:]
155 - }
156 -
157 - // There's a hanging \r\n there, gotta read past it
158 - trail := make([]byte, 2)
159 - for i := 0; i < 2; i++ {
160 - if c, err := r.ReadByte(); err != nil {
161 - return nil, err
162 - } else {
163 - trail[i] = c
164 - }
165 - }
166 -
167 - blens := len(b) + len(total)
168 - raw := make([]byte, 0, blens+2)
169 - raw = append(raw, b...)
170 - raw = append(raw, total...)
171 - raw = append(raw, trail...)
172 - return &Message{Type: BulkStr, val: total, raw: raw}, nil
173 -}
174 -
175 -func readArray(r *bufio.Reader) (*Message, error) {
176 - b, err := r.ReadBytes(delimEnd)
177 - if err != nil {
178 - return nil, err
179 - }
180 - size, err := strconv.ParseInt(string(b[1:len(b)-2]), 10, 64)
181 - if err != nil {
182 - return nil, parseErr
183 - }
184 - if size < 0 {
185 - return &Message{Type: Nil, raw: b}, nil
186 - }
187 -
188 - arr := make([]*Message, size)
189 - for i := range arr {
190 - m, err := bufioReadMessage(r)
191 - if err != nil {
192 - return nil, err
193 - }
194 - arr[i] = m
195 - b = append(b, m.raw...)
196 - }
197 - return &Message{Type: Array, val: arr, raw: b}, nil
198 -}
199 -
200 -// Bytes returns a byte slice representing the value of the Message. Only valid
201 -// for a Message of type SimpleStr, Err, and BulkStr. Others will return an
202 -// error
203 -func (m *Message) Bytes() ([]byte, error) {
204 - if b, ok := m.val.([]byte); ok {
205 - return b, nil
206 - }
207 - return nil, badType
208 -}
209 -
210 -// Str is a Convenience method around Bytes which converts the output to a
211 -// string
212 -func (m *Message) Str() (string, error) {
213 - b, err := m.Bytes()
214 - if err != nil {
215 - return "", err
216 - }
217 - return string(b), nil
218 -}
219 -
220 -// Int returns an int64 representing the value of the Message. Only valid for
221 -// Int messages
222 -func (m *Message) Int() (int64, error) {
223 - if i, ok := m.val.(int64); ok {
224 - return i, nil
225 - }
226 - return 0, badType
227 -}
228 -
229 -// Err returns an error representing the value of the Message. Only valid for
230 -// Err messages
231 -func (m *Message) Err() (error, error) {
232 - if m.Type != Err {
233 - return nil, badType
234 - }
235 - s, err := m.Str()
236 - if err != nil {
237 - return nil, err
238 - }
239 - return errors.New(s), nil
240 -}
241 -
242 -// Array returns the Message slice encompassed by this Messsage, assuming the
243 -// Message is of type Array
244 -func (m *Message) Array() ([]*Message, error) {
245 - if a, ok := m.val.([]*Message); ok {
246 - return a, nil
247 - }
248 - return nil, badType
249 -}
250 -
251 -// WriteMessage takes in the given Message and writes its encoded form to the
252 -// given io.Writer
253 -func WriteMessage(w io.Writer, m *Message) error {
254 - _, err := w.Write(m.raw)
255 - return err
256 -}
257 -
258 -// WriteArbitrary takes in any primitive golang value, or Message, and writes
259 -// its encoded form to the given io.Writer, inferring types where appropriate.
260 -func WriteArbitrary(w io.Writer, m interface{}) error {
261 - b := format(m, false)
262 - _, err := w.Write(b)
263 - return err
264 -}
265 -
266 -// WriteArbitraryAsString is similar to WriteArbitraryAsFlattenedString except
267 -// that it won't flatten any embedded arrays.
268 -func WriteArbitraryAsString(w io.Writer, m interface{}) error {
269 - b := format(m, true)
270 - _, err := w.Write(b)
271 - return err
272 -}
273 -
274 -// WriteArbitraryAsFlattenedStrings is similar to WriteArbitrary except that it
275 -// will encode all types except Array as a BulkStr, converting the argument into
276 -// a string first as necessary. It will also flatten any embedded arrays into a
277 -// single long array. This is useful because commands to a redis server must be
278 -// given as an array of bulk strings. If the argument isn't already in a slice
279 -// or map it will be wrapped so that it is written as an Array of size one.
280 -//
281 -// Note that if a Message type is found it will *not* be encoded to a BulkStr,
282 -// but will simply be passed through as whatever type it already represents.
283 -func WriteArbitraryAsFlattenedStrings(w io.Writer, m interface{}) error {
284 - fm := flatten(m)
285 - return WriteArbitraryAsString(w, fm)
286 -}
287 -
288 -func format(m interface{}, forceString bool) []byte {
289 - switch mt := m.(type) {
290 - case []byte:
291 - return formatStr(mt)
292 - case string:
293 - return formatStr([]byte(mt))
294 - case bool:
295 - if mt {
296 - return formatStr([]byte("1"))
297 - } else {
298 - return formatStr([]byte("0"))
299 - }
300 - case nil:
301 - if forceString {
302 - return formatStr([]byte{})
303 - } else {
304 - return formatNil()
305 - }
306 - case int:
307 - return formatInt(int64(mt), forceString)
308 - case int8:
309 - return formatInt(int64(mt), forceString)
310 - case int16:
311 - return formatInt(int64(mt), forceString)
312 - case int32:
313 - return formatInt(int64(mt), forceString)
314 - case int64:
315 - return formatInt(mt, forceString)
316 - case uint:
317 - return formatInt(int64(mt), forceString)
318 - case uint8:
319 - return formatInt(int64(mt), forceString)
320 - case uint16:
321 - return formatInt(int64(mt), forceString)
322 - case uint32:
323 - return formatInt(int64(mt), forceString)
324 - case uint64:
325 - return formatInt(int64(mt), forceString)
326 - case float32:
327 - ft := strconv.FormatFloat(float64(mt), 'f', -1, 32)
328 - return formatStr([]byte(ft))
329 - case float64:
330 - ft := strconv.FormatFloat(mt, 'f', -1, 64)
331 - return formatStr([]byte(ft))
332 - case error:
333 - if forceString {
334 - return formatStr([]byte(mt.Error()))
335 - } else {
336 - return formatErr(mt)
337 - }
338 -
339 - // We duplicate the below code here a bit, since this is the common case and
340 - // it'd be better to not get the reflect package involved here
341 - case []interface{}:
342 - l := len(mt)
343 - b := make([]byte, 0, l*1024)
344 - b = append(b, '*')
345 - b = append(b, []byte(strconv.Itoa(l))...)
346 - b = append(b, []byte("\r\n")...)
347 - for i := 0; i < l; i++ {
348 - b = append(b, format(mt[i], forceString)...)
349 - }
350 - return b
351 -
352 - case *Message:
353 - return mt.raw
354 -
355 - default:
356 - // Fallback to reflect-based.
357 - switch reflect.TypeOf(m).Kind() {
358 - case reflect.Slice:
359 - rm := reflect.ValueOf(mt)
360 - l := rm.Len()
361 - b := make([]byte, 0, l*1024)
362 - b = append(b, '*')
363 - b = append(b, []byte(strconv.Itoa(l))...)
364 - b = append(b, []byte("\r\n")...)
365 - for i := 0; i < l; i++ {
366 - vv := rm.Index(i).Interface()
367 - b = append(b, format(vv, forceString)...)
368 - }
369 -
370 - return b
371 - case reflect.Map:
372 - rm := reflect.ValueOf(mt)
373 - l := rm.Len() * 2
374 - b := make([]byte, 0, l*1024)
375 - b = append(b, '*')
376 - b = append(b, []byte(strconv.Itoa(l))...)
377 - b = append(b, []byte("\r\n")...)
378 - keys := rm.MapKeys()
379 - for _, k := range keys {
380 - kv := k.Interface()
381 - vv := rm.MapIndex(k).Interface()
382 - b = append(b, format(kv, forceString)...)
383 - b = append(b, format(vv, forceString)...)
384 - }
385 - return b
386 - default:
387 - return formatStr([]byte(fmt.Sprint(m)))
388 - }
389 - }
390 -}
391 -
392 -var typeOfBytes = reflect.TypeOf([]byte(nil))
393 -
394 -func flatten(m interface{}) []interface{} {
395 - t := reflect.TypeOf(m)
396 -
397 - // If it's a byte-slice we don't want to flatten
398 - if t == typeOfBytes {
399 - return []interface{}{m}
400 - }
401 -
402 - switch t.Kind() {
403 - case reflect.Slice:
404 - rm := reflect.ValueOf(m)
405 - l := rm.Len()
406 - ret := make([]interface{}, 0, l)
407 - for i := 0; i < l; i++ {
408 - ret = append(ret, flatten(rm.Index(i).Interface())...)
409 - }
410 - return ret
411 -
412 - case reflect.Map:
413 - rm := reflect.ValueOf(m)
414 - l := rm.Len() * 2
415 - keys := rm.MapKeys()
416 - ret := make([]interface{}, 0, l)
417 - for _, k := range keys {
418 - kv := k.Interface()
419 - vv := rm.MapIndex(k).Interface()
420 - ret = append(ret, flatten(kv)...)
421 - ret = append(ret, flatten(vv)...)
422 - }
423 - return ret
424 -
425 - default:
426 - return []interface{}{m}
427 - }
428 -}
429 -
430 -func formatStr(b []byte) []byte {
431 - l := strconv.Itoa(len(b))
432 - bs := make([]byte, 0, len(l)+len(b)+5)
433 - bs = append(bs, bulkStrPrefix)
434 - bs = append(bs, []byte(l)...)
435 - bs = append(bs, delim...)
436 - bs = append(bs, b...)
437 - bs = append(bs, delim...)
438 - return bs
439 -}
440 -
441 -func formatErr(ierr error) []byte {
442 - ierrstr := []byte(ierr.Error())
443 - bs := make([]byte, 0, len(ierrstr)+3)
444 - bs = append(bs, errPrefix)
445 - bs = append(bs, ierrstr...)
446 - bs = append(bs, delim...)
447 - return bs
448 -}
449 -
450 -func formatInt(i int64, forceString bool) []byte {
451 - istr := strconv.FormatInt(i, 10)
452 - if forceString {
453 - return formatStr([]byte(istr))
454 - }
455 - bs := make([]byte, 0, len(istr)+3)
456 - bs = append(bs, intPrefix)
457 - bs = append(bs, istr...)
458 - bs = append(bs, delim...)
459 - return bs
460 -}
461 -
462 -var nilFormatted = []byte("$-1\r\n")
463 -
464 -func formatNil() []byte {
465 - return nilFormatted
466 -}
Godeps/_workspace/src/github.com/fzzy/radix/redis/resp/resp_test.go deleted
-209
@@ -1,209 +0,0 @@
1 -package resp
2 -
3 -import (
4 - "bytes"
5 - "errors"
6 - "github.com/stretchr/testify/assert"
7 - . "testing"
8 -)
9 -
10 -func TestRead(t *T) {
11 - var m *Message
12 - var err error
13 -
14 - _, err = NewMessage(nil)
15 - assert.NotNil(t, err)
16 -
17 - _, err = NewMessage([]byte{})
18 - assert.NotNil(t, err)
19 -
20 - // Simple string
21 - m, _ = NewMessage([]byte("+ohey\r\n"))
22 - assert.Equal(t, SimpleStr, m.Type)
23 - assert.Equal(t, []byte("ohey"), m.val)
24 -
25 - // Empty simple string
26 - m, _ = NewMessage([]byte("+\r\n"))
27 - assert.Equal(t, SimpleStr, m.Type)
28 - assert.Equal(t, []byte(""), m.val.([]byte))
29 -
30 - // Error
31 - m, _ = NewMessage([]byte("-ohey\r\n"))
32 - assert.Equal(t, Err, m.Type)
33 - assert.Equal(t, []byte("ohey"), m.val.([]byte))
34 -
35 - // Empty error
36 - m, _ = NewMessage([]byte("-\r\n"))
37 - assert.Equal(t, Err, m.Type)
38 - assert.Equal(t, []byte(""), m.val.([]byte))
39 -
40 - // Int
41 - m, _ = NewMessage([]byte(":1024\r\n"))
42 - assert.Equal(t, Int, m.Type)
43 - assert.Equal(t, int64(1024), m.val.(int64))
44 -
45 - // Bulk string
46 - m, _ = NewMessage([]byte("$3\r\nfoo\r\n"))
47 - assert.Equal(t, BulkStr, m.Type)
48 - assert.Equal(t, []byte("foo"), m.val.([]byte))
49 -
50 - // Empty bulk string
51 - m, _ = NewMessage([]byte("$0\r\n\r\n"))
52 - assert.Equal(t, BulkStr, m.Type)
53 - assert.Equal(t, []byte(""), m.val.([]byte))
54 -
55 - // Nil bulk string
56 - m, _ = NewMessage([]byte("$-1\r\n"))
57 - assert.Equal(t, Nil, m.Type)
58 -
59 - // Array
60 - m, _ = NewMessage([]byte("*2\r\n+foo\r\n+bar\r\n"))
61 - assert.Equal(t, Array, m.Type)
62 - assert.Equal(t, 2, len(m.val.([]*Message)))
63 - assert.Equal(t, SimpleStr, m.val.([]*Message)[0].Type)
64 - assert.Equal(t, []byte("foo"), m.val.([]*Message)[0].val.([]byte))
65 - assert.Equal(t, SimpleStr, m.val.([]*Message)[1].Type)
66 - assert.Equal(t, []byte("bar"), m.val.([]*Message)[1].val.([]byte))
67 -
68 - // Empty array
69 - m, _ = NewMessage([]byte("*0\r\n"))
70 - assert.Equal(t, Array, m.Type)
71 - assert.Equal(t, 0, len(m.val.([]*Message)))
72 -
73 - // Nil Array
74 - m, _ = NewMessage([]byte("*-1\r\n"))
75 - assert.Equal(t, Nil, m.Type)
76 -
77 - // Embedded Array
78 - m, _ = NewMessage([]byte("*3\r\n+foo\r\n+bar\r\n*2\r\n+foo\r\n+bar\r\n"))
79 - assert.Equal(t, Array, m.Type)
80 - assert.Equal(t, 3, len(m.val.([]*Message)))
81 - assert.Equal(t, SimpleStr, m.val.([]*Message)[0].Type)
82 - assert.Equal(t, []byte("foo"), m.val.([]*Message)[0].val.([]byte))
83 - assert.Equal(t, SimpleStr, m.val.([]*Message)[1].Type)
84 - assert.Equal(t, []byte("bar"), m.val.([]*Message)[1].val.([]byte))
85 - m = m.val.([]*Message)[2]
86 - assert.Equal(t, 2, len(m.val.([]*Message)))
87 - assert.Equal(t, SimpleStr, m.val.([]*Message)[0].Type)
88 - assert.Equal(t, []byte("foo"), m.val.([]*Message)[0].val.([]byte))
89 - assert.Equal(t, SimpleStr, m.val.([]*Message)[1].Type)
90 - assert.Equal(t, []byte("bar"), m.val.([]*Message)[1].val.([]byte))
91 -
92 - // Test that two bulks in a row read correctly
93 - m, _ = NewMessage([]byte("*2\r\n$3\r\nfoo\r\n$3\r\nbar\r\n"))
94 - assert.Equal(t, Array, m.Type)
95 - assert.Equal(t, 2, len(m.val.([]*Message)))
96 - assert.Equal(t, BulkStr, m.val.([]*Message)[0].Type)
97 - assert.Equal(t, []byte("foo"), m.val.([]*Message)[0].val.([]byte))
98 - assert.Equal(t, BulkStr, m.val.([]*Message)[1].Type)
99 - assert.Equal(t, []byte("bar"), m.val.([]*Message)[1].val.([]byte))
100 -}
101 -
102 -type arbitraryTest struct {
103 - val interface{}
104 - expect []byte
105 -}
106 -
107 -var nilMessage, _ = NewMessage([]byte("$-1\r\n"))
108 -
109 -var arbitraryTests = []arbitraryTest{
110 - {[]byte("OHAI"), []byte("$4\r\nOHAI\r\n")},
111 - {"OHAI", []byte("$4\r\nOHAI\r\n")},
112 - {true, []byte("$1\r\n1\r\n")},
113 - {false, []byte("$1\r\n0\r\n")},
114 - {nil, []byte("$-1\r\n")},
115 - {80, []byte(":80\r\n")},
116 - {int64(-80), []byte(":-80\r\n")},
117 - {uint64(80), []byte(":80\r\n")},
118 - {float32(0.1234), []byte("$6\r\n0.1234\r\n")},
119 - {float64(0.1234), []byte("$6\r\n0.1234\r\n")},
120 - {errors.New("hi"), []byte("-hi\r\n")},
121 -
122 - {nilMessage, []byte("$-1\r\n")},
123 -
124 - {[]int{1, 2, 3}, []byte("*3\r\n:1\r\n:2\r\n:3\r\n")},
125 - {map[int]int{1: 2}, []byte("*2\r\n:1\r\n:2\r\n")},
126 -
127 - {NewSimpleString("OK"), []byte("+OK\r\n")},
128 -}
129 -
130 -var arbitraryAsStringTests = []arbitraryTest{
131 - {[]byte("OHAI"), []byte("$4\r\nOHAI\r\n")},
132 - {"OHAI", []byte("$4\r\nOHAI\r\n")},
133 - {true, []byte("$1\r\n1\r\n")},
134 - {false, []byte("$1\r\n0\r\n")},
135 - {nil, []byte("$0\r\n\r\n")},
136 - {80, []byte("$2\r\n80\r\n")},
137 - {int64(-80), []byte("$3\r\n-80\r\n")},
138 - {uint64(80), []byte("$2\r\n80\r\n")},
139 - {float32(0.1234), []byte("$6\r\n0.1234\r\n")},
140 - {float64(0.1234), []byte("$6\r\n0.1234\r\n")},
141 - {errors.New("hi"), []byte("$2\r\nhi\r\n")},
142 -
143 - {nilMessage, []byte("$-1\r\n")},
144 -
145 - {[]int{1, 2, 3}, []byte("*3\r\n$1\r\n1\r\n$1\r\n2\r\n$1\r\n3\r\n")},
146 - {map[int]int{1: 2}, []byte("*2\r\n$1\r\n1\r\n$1\r\n2\r\n")},
147 -
148 - {NewSimpleString("OK"), []byte("+OK\r\n")},
149 -}
150 -
151 -var arbitraryAsFlattenedStringsTests = []arbitraryTest{
152 - {
153 - []interface{}{"wat", map[string]interface{}{
154 - "foo": 1,
155 - }},
156 - []byte("*3\r\n$3\r\nwat\r\n$3\r\nfoo\r\n$1\r\n1\r\n"),
157 - },
158 -}
159 -
160 -func TestWriteArbitrary(t *T) {
161 - var err error
162 - buf := bytes.NewBuffer([]byte{})
163 - for _, test := range arbitraryTests {
164 - t.Logf("Checking test %v", test)
165 - buf.Reset()
166 - err = WriteArbitrary(buf, test.val)
167 - assert.Nil(t, err)
168 - assert.Equal(t, test.expect, buf.Bytes())
169 - }
170 -}
171 -
172 -func TestWriteArbitraryAsString(t *T) {
173 - var err error
174 - buf := bytes.NewBuffer([]byte{})
175 - for _, test := range arbitraryAsStringTests {
176 - t.Logf("Checking test %v", test)
177 - buf.Reset()
178 - err = WriteArbitraryAsString(buf, test.val)
179 - assert.Nil(t, err)
180 - assert.Equal(t, test.expect, buf.Bytes())
181 - }
182 -}
183 -
184 -func TestWriteArbitraryAsFlattenedStrings(t *T) {
185 - var err error
186 - buf := bytes.NewBuffer([]byte{})
187 - for _, test := range arbitraryAsFlattenedStringsTests {
188 - t.Logf("Checking test %v", test)
189 - buf.Reset()
190 - err = WriteArbitraryAsFlattenedStrings(buf, test.val)
191 - assert.Nil(t, err)
192 - assert.Equal(t, test.expect, buf.Bytes())
193 - }
194 -}
195 -
196 -func TestMessageWrite(t *T) {
197 - var err error
198 - var m *Message
199 - buf := bytes.NewBuffer([]byte{})
200 - for _, test := range arbitraryTests {
201 - t.Logf("Checking test; %v", test)
202 - buf.Reset()
203 - m, err = NewMessage(test.expect)
204 - assert.Nil(t, err)
205 - err = WriteMessage(buf, m)
206 - assert.Nil(t, err)
207 - assert.Equal(t, test.expect, buf.Bytes())
208 - }
209 -}
Godeps/_workspace/src/github.com/jbenet/go-datastore/Godeps/Godeps.json
+8 -4
@@ -17,6 +17,10 @@
17 "ImportPath": "github.com/codahale/metrics",
18 "Rev": "7d3beb1b480077e77c08a6f6c65ea969f6e91420"
19 },
20 + {
21 + "ImportPath": "github.com/dustin/randbo",
22 + "Rev": "7f1b564ca7242d22bcc6e2128beb90d9fa38b9f0"
23 + },
24 {
25 "ImportPath": "github.com/hashicorp/golang-lru",
26 "Rev": "4dfff096c4973178c8f35cf6dd1a732a0a139370"
@@ -29,10 +33,6 @@
33 "ImportPath": "github.com/jbenet/goprocess",
34 "Rev": "5b02f8d275a2dd882fb06f8bbdf74347795ff3b1"
35 },
32 - {
33 - "ImportPath": "github.com/satori/go.uuid",
34 - "Rev": "7c7f2020c4c9491594b85767967f4619c2fa75f9"
35 - },
36 {
37 "ImportPath": "github.com/mattbaird/elastigo/api",
38 "Rev": "041b88c1fcf6489a5721ede24378ce1253b9159d"
@@ -41,6 +41,10 @@
41 "ImportPath": "github.com/mattbaird/elastigo/core",
42 "Rev": "041b88c1fcf6489a5721ede24378ce1253b9159d"
43 },
44 + {
45 + "ImportPath": "github.com/satori/go.uuid",
46 + "Rev": "7c7f2020c4c9491594b85767967f4619c2fa75f9"
47 + },
48 {
49 "ImportPath": "github.com/syndtr/goleveldb/leveldb",
50 "Rev": "871eee0a7546bb7d1b2795142e29c4534abc49b3"
Godeps/_workspace/src/github.com/jbenet/go-datastore/basic_ds.go
+18 -1
@@ -63,6 +63,10 @@ func (d *MapDatastore) Query(q dsq.Query) (dsq.Results, error) {
63 return r, nil
64 }
65
66 +func (d *MapDatastore) Batch() (Batch, error) {
67 + return NewBasicBatch(d), nil
68 +}
69 +
70 // NullDatastore stores nothing, but conforms to the API.
71 // Useful to test with.
72 type NullDatastore struct {
@@ -98,6 +102,10 @@ func (d *NullDatastore) Query(q dsq.Query) (dsq.Results, error) {
102 return dsq.ResultsWithEntries(q, nil), nil
103 }
104
105 +func (d *NullDatastore) Batch() (Batch, error) {
106 + return NewBasicBatch(d), nil
107 +}
108 +
109 // LogDatastore logs all accesses through the datastore.
110 type LogDatastore struct {
111 Name string
@@ -112,7 +120,7 @@ type Shim interface {
120 }
121
122 // NewLogDatastore constructs a log datastore.
115 -func NewLogDatastore(ds Datastore, name string) Shim {
123 +func NewLogDatastore(ds Datastore, name string) *LogDatastore {
124 if len(name) < 1 {
125 name = "LogDatastore"
126 }
@@ -154,3 +162,12 @@ func (d *LogDatastore) Query(q dsq.Query) (dsq.Results, error) {
162 log.Printf("%s: Query\n", d.Name)
163 return d.child.Query(q)
164 }
165 +
166 +func (d *LogDatastore) Batch() (Batch, error) {
167 + log.Printf("%s: Batch\n", d.Name)
168 + bds, ok := d.child.(BatchingDatastore)
169 + if !ok {
170 + return nil, ErrBatchUnsupported
171 + }
172 + return bds.Batch()
173 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/batch.go new
+44
@@ -0,0 +1,44 @@
1 +package datastore
2 +
3 +// basicBatch implements the transaction interface for datastores who do
4 +// not have any sort of underlying transactional support
5 +type basicBatch struct {
6 + puts map[Key]interface{}
7 + deletes map[Key]struct{}
8 +
9 + target Datastore
10 +}
11 +
12 +func NewBasicBatch(ds Datastore) Batch {
13 + return &basicBatch{
14 + puts: make(map[Key]interface{}),
15 + deletes: make(map[Key]struct{}),
16 + target: ds,
17 + }
18 +}
19 +
20 +func (bt *basicBatch) Put(key Key, val interface{}) error {
21 + bt.puts[key] = val
22 + return nil
23 +}
24 +
25 +func (bt *basicBatch) Delete(key Key) error {
26 + bt.deletes[key] = struct{}{}
27 + return nil
28 +}
29 +
30 +func (bt *basicBatch) Commit() error {
31 + for k, val := range bt.puts {
32 + if err := bt.target.Put(k, val); err != nil {
33 + return err
34 + }
35 + }
36 +
37 + for k, _ := range bt.deletes {
38 + if err := bt.target.Delete(k); err != nil {
39 + return err
40 + }
41 + }
42 +
43 + return nil
44 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/callback/callback.go
+1 -1
@@ -10,7 +10,7 @@ type Datastore struct {
10 F func()
11 }
12
13 -func Wrap(ds ds.Datastore, f func()) ds.Datastore {
13 +func Wrap(ds ds.Datastore, f func()) *Datastore {
14 return &Datastore{ds, f}
15 }
16
Godeps/_workspace/src/github.com/jbenet/go-datastore/datastore.go
+16
@@ -69,6 +69,14 @@ type Datastore interface {
69 Query(q query.Query) (query.Results, error)
70 }
71
72 +type BatchingDatastore interface {
73 + Datastore
74 +
75 + Batch() (Batch, error)
76 +}
77 +
78 +var ErrBatchUnsupported = errors.New("this datastore does not support batching")
79 +
80 // ThreadSafeDatastore is an interface that all threadsafe datastore should
81 // implement to leverage type safety checks.
82 type ThreadSafeDatastore interface {
@@ -104,3 +112,11 @@ func GetBackedHas(ds Datastore, key Key) (bool, error) {
112 return false, err
113 }
114 }
115 +
116 +type Batch interface {
117 + Put(key Key, val interface{}) error
118 +
119 + Delete(key Key) error
120 +
121 + Commit() error
122 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/flatfs/flatfs.go
+134 -6
@@ -68,12 +68,8 @@ func (fs *Datastore) decode(file string) (key datastore.Key, ok bool) {
68 }
69
70 func (fs *Datastore) makePrefixDir(dir string) error {
71 - if err := os.Mkdir(dir, 0777); err != nil {
72 - // EEXIST is safe to ignore here, that just means the prefix
73 - // directory already existed.
74 - if !os.IsExist(err) {
75 - return err
76 - }
71 + if err := fs.makePrefixDirNoSync(dir); err != nil {
72 + return err
73 }
74
75 // In theory, if we create a new prefix dir and add a file to
@@ -86,6 +82,17 @@ func (fs *Datastore) makePrefixDir(dir string) error {
82 return nil
83 }
84
85 +func (fs *Datastore) makePrefixDirNoSync(dir string) error {
86 + if err := os.Mkdir(dir, 0777); err != nil {
87 + // EEXIST is safe to ignore here, that just means the prefix
88 + // directory already existed.
89 + if !os.IsExist(err) {
90 + return err
91 + }
92 + }
93 + return nil
94 +}
95 +
96 func (fs *Datastore) Put(key datastore.Key, value interface{}) error {
97 val, ok := value.([]byte)
98 if !ok {
@@ -137,6 +144,88 @@ func (fs *Datastore) Put(key datastore.Key, value interface{}) error {
144 return nil
145 }
146
147 +func (fs *Datastore) putMany(data map[datastore.Key]interface{}) error {
148 + var dirsToSync []string
149 + files := make(map[*os.File]string)
150 +
151 + for key, value := range data {
152 + val, ok := value.([]byte)
153 + if !ok {
154 + return datastore.ErrInvalidType
155 + }
156 + dir, path := fs.encode(key)
157 + if err := fs.makePrefixDirNoSync(dir); err != nil {
158 + return err
159 + }
160 + dirsToSync = append(dirsToSync, dir)
161 +
162 + tmp, err := ioutil.TempFile(dir, "put-")
163 + if err != nil {
164 + return err
165 + }
166 +
167 + if _, err := tmp.Write(val); err != nil {
168 + return err
169 + }
170 +
171 + files[tmp] = path
172 + }
173 +
174 + ops := make(map[*os.File]int)
175 +
176 + defer func() {
177 + for fi, _ := range files {
178 + val, _ := ops[fi]
179 + switch val {
180 + case 0:
181 + _ = fi.Close()
182 + fallthrough
183 + case 1:
184 + _ = os.Remove(fi.Name())
185 + }
186 + }
187 + }()
188 +
189 + // Now we sync everything
190 + // sync and close files
191 + for fi, _ := range files {
192 + if err := fi.Sync(); err != nil {
193 + return err
194 + }
195 +
196 + if err := fi.Close(); err != nil {
197 + return err
198 + }
199 +
200 + // signify closed
201 + ops[fi] = 1
202 + }
203 +
204 + // move files to their proper places
205 + for fi, path := range files {
206 + if err := osrename.Rename(fi.Name(), path); err != nil {
207 + return err
208 + }
209 +
210 + // signify removed
211 + ops[fi] = 2
212 + }
213 +
214 + // now sync the dirs for those files
215 + for _, dir := range dirsToSync {
216 + if err := syncDir(dir); err != nil {
217 + return err
218 + }
219 + }
220 +
221 + // sync top flatfs dir
222 + if err := syncDir(fs.path); err != nil {
223 + return err
224 + }
225 +
226 + return nil
227 +}
228 +
229 func (fs *Datastore) Get(key datastore.Key) (value interface{}, err error) {
230 _, path := fs.encode(key)
231 data, err := ioutil.ReadFile(path)
@@ -234,6 +323,45 @@ func (fs *Datastore) enumerateKeys(fi os.FileInfo, res []query.Entry) ([]query.E
323 return res, nil
324 }
325
326 +type flatfsBatch struct {
327 + puts map[datastore.Key]interface{}
328 + deletes map[datastore.Key]struct{}
329 +
330 + ds *Datastore
331 +}
332 +
333 +func (fs *Datastore) Batch() (datastore.Batch, error) {
334 + return &flatfsBatch{
335 + puts: make(map[datastore.Key]interface{}),
336 + deletes: make(map[datastore.Key]struct{}),
337 + ds: fs,
338 + }, nil
339 +}
340 +
341 +func (bt *flatfsBatch) Put(key datastore.Key, val interface{}) error {
342 + bt.puts[key] = val
343 + return nil
344 +}
345 +
346 +func (bt *flatfsBatch) Delete(key datastore.Key) error {
347 + bt.deletes[key] = struct{}{}
348 + return nil
349 +}
350 +
351 +func (bt *flatfsBatch) Commit() error {
352 + if err := bt.ds.putMany(bt.puts); err != nil {
353 + return err
354 + }
355 +
356 + for k, _ := range bt.deletes {
357 + if err := bt.ds.Delete(k); err != nil {
358 + return err
359 + }
360 + }
361 +
362 + return nil
363 +}
364 +
365 var _ datastore.ThreadSafeDatastore = (*Datastore)(nil)
366
367 func (*Datastore) IsThreadSafe() {}
Godeps/_workspace/src/github.com/jbenet/go-datastore/flatfs/flatfs_test.go
+98
@@ -1,15 +1,18 @@
1 package flatfs_test
2
3 import (
4 + "encoding/base32"
5 "io/ioutil"
6 "os"
7 "path/filepath"
8 "runtime"
9 "testing"
10
11 + rand "github.com/dustin/randbo"
12 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
13 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/flatfs"
14 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/query"
15 + dstest "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/test"
16 )
17
18 func tempdir(t testing.TB) (path string, cleanup func()) {
@@ -316,3 +319,98 @@ func TestQuerySimple(t *testing.T) {
319 t.Errorf("did not see wanted key %q in %+v", myKey, entries)
320 }
321 }
322 +
323 +func TestBatchPut(t *testing.T) {
324 + temp, cleanup := tempdir(t)
325 + defer cleanup()
326 +
327 + fs, err := flatfs.New(temp, 2)
328 + if err != nil {
329 + t.Fatalf("New fail: %v\n", err)
330 + }
331 +
332 + dstest.RunBatchTest(t, fs)
333 +}
334 +
335 +func TestBatchDelete(t *testing.T) {
336 + temp, cleanup := tempdir(t)
337 + defer cleanup()
338 +
339 + fs, err := flatfs.New(temp, 2)
340 + if err != nil {
341 + t.Fatalf("New fail: %v\n", err)
342 + }
343 +
344 + dstest.RunBatchDeleteTest(t, fs)
345 +}
346 +
347 +func BenchmarkConsecutivePut(b *testing.B) {
348 + r := rand.New()
349 + var blocks [][]byte
350 + var keys []datastore.Key
351 + for i := 0; i < b.N; i++ {
352 + blk := make([]byte, 256*1024)
353 + r.Read(blk)
354 + blocks = append(blocks, blk)
355 +
356 + key := base32.StdEncoding.EncodeToString(blk[:8])
357 + keys = append(keys, datastore.NewKey(key))
358 + }
359 + temp, cleanup := tempdir(b)
360 + defer cleanup()
361 +
362 + fs, err := flatfs.New(temp, 2)
363 + if err != nil {
364 + b.Fatalf("New fail: %v\n", err)
365 + }
366 +
367 + b.ResetTimer()
368 +
369 + for i := 0; i < b.N; i++ {
370 + err := fs.Put(keys[i], blocks[i])
371 + if err != nil {
372 + b.Fatal(err)
373 + }
374 + }
375 +}
376 +
377 +func BenchmarkBatchedPut(b *testing.B) {
378 + r := rand.New()
379 + var blocks [][]byte
380 + var keys []datastore.Key
381 + for i := 0; i < b.N; i++ {
382 + blk := make([]byte, 256*1024)
383 + r.Read(blk)
384 + blocks = append(blocks, blk)
385 +
386 + key := base32.StdEncoding.EncodeToString(blk[:8])
387 + keys = append(keys, datastore.NewKey(key))
388 + }
389 + temp, cleanup := tempdir(b)
390 + defer cleanup()
391 +
392 + fs, err := flatfs.New(temp, 2)
393 + if err != nil {
394 + b.Fatalf("New fail: %v\n", err)
395 + }
396 +
397 + b.ResetTimer()
398 +
399 + for i := 0; i < b.N; {
400 + batch, err := fs.Batch()
401 + if err != nil {
402 + b.Fatal(err)
403 + }
404 +
405 + for n := i; i-n < 512 && i < b.N; i++ {
406 + err := batch.Put(keys[i], blocks[i])
407 + if err != nil {
408 + b.Fatal(err)
409 + }
410 + }
411 + err = batch.Commit()
412 + if err != nil {
413 + b.Fatal(err)
414 + }
415 + }
416 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/key.go
+2 -1
@@ -4,8 +4,9 @@ import (
4 "path"
5 "strings"
6
7 - dsq "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/query"
7 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/satori/go.uuid"
8 +
9 + dsq "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/query"
10 )
11
12 /*
Godeps/_workspace/src/github.com/jbenet/go-datastore/keytransform/interface.go
+2
@@ -16,6 +16,8 @@ type KeyTransform interface {
16 type Datastore interface {
17 ds.Shim
18 KeyTransform
19 +
20 + Batch() (ds.Batch, error)
21 }
22
23 // Wrap wraps a given datastore with a KeyTransform function.
Godeps/_workspace/src/github.com/jbenet/go-datastore/keytransform/keytransform.go
+34
@@ -73,3 +73,37 @@ func (d *ktds) Query(q dsq.Query) (dsq.Results, error) {
73
74 return dsq.DerivedResults(qr, ch), nil
75 }
76 +
77 +func (d *ktds) Batch() (ds.Batch, error) {
78 + bds, ok := d.child.(ds.BatchingDatastore)
79 + if !ok {
80 + return nil, ds.ErrBatchUnsupported
81 + }
82 +
83 + childbatch, err := bds.Batch()
84 + if err != nil {
85 + return nil, err
86 + }
87 + return &transformBatch{
88 + dst: childbatch,
89 + f: d.ConvertKey,
90 + }, nil
91 +}
92 +
93 +type transformBatch struct {
94 + dst ds.Batch
95 +
96 + f KeyMapping
97 +}
98 +
99 +func (t *transformBatch) Put(key ds.Key, val interface{}) error {
100 + return t.dst.Put(t.f(key), val)
101 +}
102 +
103 +func (t *transformBatch) Delete(key ds.Key) error {
104 + return t.dst.Delete(t.f(key))
105 +}
106 +
107 +func (t *transformBatch) Commit() error {
108 + return t.dst.Commit()
109 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/measure/measure.go
+79
@@ -148,6 +148,85 @@ func (m *measure) Query(q query.Query) (query.Results, error) {
148 return res, err
149 }
150
151 +type measuredBatch struct {
152 + puts int
153 + deletes int
154 +
155 + putts datastore.Batch
156 + delts datastore.Batch
157 +
158 + m *measure
159 +}
160 +
161 +func (m *measure) Batch() (datastore.Batch, error) {
162 + bds, ok := m.backend.(datastore.BatchingDatastore)
163 + if !ok {
164 + return nil, datastore.ErrBatchUnsupported
165 + }
166 + pb, err := bds.Batch()
167 + if err != nil {
168 + return nil, err
169 + }
170 +
171 + db, err := bds.Batch()
172 + if err != nil {
173 + return nil, err
174 + }
175 +
176 + return &measuredBatch{
177 + putts: pb,
178 + delts: db,
179 +
180 + m: m,
181 + }, nil
182 +}
183 +
184 +func (mt *measuredBatch) Put(key datastore.Key, val interface{}) error {
185 + mt.puts++
186 + valb, ok := val.([]byte)
187 + if !ok {
188 + return datastore.ErrInvalidType
189 + }
190 + _ = mt.m.putSize.RecordValue(int64(len(valb)))
191 + return mt.putts.Put(key, val)
192 +}
193 +
194 +func (mt *measuredBatch) Delete(key datastore.Key) error {
195 + mt.deletes++
196 + return mt.delts.Delete(key)
197 +}
198 +
199 +func (mt *measuredBatch) Commit() error {
200 + err := logBatchCommit(mt.delts, mt.deletes, mt.m.deleteNum, mt.m.deleteErr, mt.m.deleteLatency)
201 + if err != nil {
202 + return err
203 + }
204 +
205 + err = logBatchCommit(mt.putts, mt.puts, mt.m.putNum, mt.m.putErr, mt.m.putLatency)
206 + if err != nil {
207 + return err
208 + }
209 +
210 + return nil
211 +}
212 +
213 +func logBatchCommit(b datastore.Batch, n int, num, errs metrics.Counter, lat *metrics.Histogram) error {
214 + if n > 0 {
215 + before := time.Now()
216 + err := b.Commit()
217 + took := int(time.Now().Sub(before)/time.Microsecond) / n
218 + num.AddN(uint64(n))
219 + for i := 0; i < n; i++ {
220 + _ = lat.RecordValue(int64(took))
221 + }
222 + if err != nil {
223 + errs.Add()
224 + return err
225 + }
226 + }
227 + return nil
228 +}
229 +
230 func (m *measure) Close() error {
231 m.putNum.Remove()
232 m.putErr.Remove()
Godeps/_workspace/src/github.com/jbenet/go-datastore/mount/mount.go
+59
@@ -114,3 +114,62 @@ func (d *Datastore) Query(q query.Query) (query.Results, error) {
114 r = query.ResultsReplaceQuery(r, q)
115 return r, nil
116 }
117 +
118 +type mountBatch struct {
119 + mounts map[string]datastore.Batch
120 +
121 + d *Datastore
122 +}
123 +
124 +func (d *Datastore) Batch() (datastore.Batch, error) {
125 + return &mountBatch{
126 + mounts: make(map[string]datastore.Batch),
127 + d: d,
128 + }, nil
129 +}
130 +
131 +func (mt *mountBatch) lookupBatch(key datastore.Key) (datastore.Batch, datastore.Key, error) {
132 + child, loc, rest := mt.d.lookup(key)
133 + t, ok := mt.mounts[loc.String()]
134 + if !ok {
135 + bds, ok := child.(datastore.BatchingDatastore)
136 + if !ok {
137 + return nil, datastore.NewKey(""), datastore.ErrBatchUnsupported
138 + }
139 + var err error
140 + t, err = bds.Batch()
141 + if err != nil {
142 + return nil, datastore.NewKey(""), err
143 + }
144 + mt.mounts[loc.String()] = t
145 + }
146 + return t, rest, nil
147 +}
148 +
149 +func (mt *mountBatch) Put(key datastore.Key, val interface{}) error {
150 + t, rest, err := mt.lookupBatch(key)
151 + if err != nil {
152 + return err
153 + }
154 +
155 + return t.Put(rest, val)
156 +}
157 +
158 +func (mt *mountBatch) Delete(key datastore.Key) error {
159 + t, rest, err := mt.lookupBatch(key)
160 + if err != nil {
161 + return err
162 + }
163 +
164 + return t.Delete(rest)
165 +}
166 +
167 +func (mt *mountBatch) Commit() error {
168 + for _, t := range mt.mounts {
169 + err := t.Commit()
170 + if err != nil {
171 + return err
172 + }
173 + }
174 + return nil
175 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/panic/panic.go
+31
@@ -66,3 +66,34 @@ func (d *datastore) Query(q dsq.Query) (dsq.Results, error) {
66 }
67 return r, nil
68 }
69 +
70 +type panicBatch struct {
71 + t ds.Batch
72 +}
73 +
74 +func (p *panicBatch) Put(key ds.Key, val interface{}) error {
75 + err := p.t.Put(key, val)
76 + if err != nil {
77 + fmt.Fprintf(os.Stdout, "panic datastore: %s", err)
78 + panic("panic datastore: transaction put failed")
79 + }
80 + return nil
81 +}
82 +
83 +func (p *panicBatch) Delete(key ds.Key) error {
84 + err := p.t.Delete(key)
85 + if err != nil {
86 + fmt.Fprintf(os.Stdout, "panic datastore: %s", err)
87 + panic("panic datastore: transaction delete failed")
88 + }
89 + return nil
90 +}
91 +
92 +func (p *panicBatch) Commit() error {
93 + err := p.t.Commit()
94 + if err != nil {
95 + fmt.Fprintf(os.Stdout, "panic datastore: %s", err)
96 + panic("panic datastore: transaction commit failed")
97 + }
98 + return nil
99 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/redis/redis.go renamed
+2 -1
@@ -6,7 +6,8 @@ import (
6 "sync"
7 "time"
8
9 - "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/fzzy/radix/redis"
9 + "github.com/fzzy/radix/redis"
10 +
11 datastore "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
12 query "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/query"
13 )
Godeps/_workspace/src/github.com/jbenet/go-datastore/redis/redis_test.go renamed
+8 -8
@@ -6,9 +6,9 @@ import (
6 "testing"
7 "time"
8
9 - "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/fzzy/radix/redis"
9 + "github.com/fzzy/radix/redis"
10 datastore "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
11 - "github.com/ipfs/go-ipfs/thirdparty/assert"
11 + dstest "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/test"
12 )
13
14 const RedisEnv = "REDIS_DATASTORE_TEST_HOST"
@@ -20,7 +20,7 @@ func TestPutGetBytes(t *testing.T) {
20 t.Fatal(err)
21 }
22 key, val := datastore.NewKey("foo"), []byte("bar")
23 - assert.Nil(ds.Put(key, val), t)
23 + dstest.Nil(ds.Put(key, val), t)
24 v, err := ds.Get(key)
25 if err != nil {
26 t.Fatal(err)
@@ -45,7 +45,7 @@ func TestHasBytes(t *testing.T) {
45 t.Fail()
46 }
47
48 - assert.Nil(ds.Put(key, val), t)
48 + dstest.Nil(ds.Put(key, val), t)
49 hasAfterPut, err := ds.Has(key)
50 if err != nil {
51 t.Fatal(err)
@@ -62,8 +62,8 @@ func TestDelete(t *testing.T) {
62 t.Fatal(err)
63 }
64 key, val := datastore.NewKey("foo"), []byte("bar")
65 - assert.Nil(ds.Put(key, val), t)
66 - assert.Nil(ds.Delete(key), t)
65 + dstest.Nil(ds.Put(key, val), t)
66 + dstest.Nil(ds.Delete(key), t)
67
68 hasAfterDelete, err := ds.Has(key)
69 if err != nil {
@@ -82,9 +82,9 @@ func TestExpiry(t *testing.T) {
82 t.Fatal(err)
83 }
84 key, val := datastore.NewKey("foo"), []byte("bar")
85 - assert.Nil(ds.Put(key, val), t)
85 + dstest.Nil(ds.Put(key, val), t)
86 time.Sleep(ttl + 1*time.Second)
87 - assert.Nil(ds.Delete(key), t)
87 + dstest.Nil(ds.Delete(key), t)
88
89 hasAfterExpiration, err := ds.Has(key)
90 if err != nil {
Godeps/_workspace/src/github.com/jbenet/go-datastore/sync/sync.go
+40
@@ -63,3 +63,43 @@ func (d *MutexDatastore) Query(q dsq.Query) (dsq.Results, error) {
63 defer d.RUnlock()
64 return d.child.Query(q)
65 }
66 +
67 +func (d *MutexDatastore) Batch() (ds.Batch, error) {
68 + d.RLock()
69 + defer d.RUnlock()
70 + bds, ok := d.child.(ds.BatchingDatastore)
71 + if !ok {
72 + return nil, ds.ErrBatchUnsupported
73 + }
74 +
75 + b, err := bds.Batch()
76 + if err != nil {
77 + return nil, err
78 + }
79 + return &syncBatch{
80 + batch: b,
81 + }, nil
82 +}
83 +
84 +type syncBatch struct {
85 + lk sync.Mutex
86 + batch ds.Batch
87 +}
88 +
89 +func (b *syncBatch) Put(key ds.Key, val interface{}) error {
90 + b.lk.Lock()
91 + defer b.lk.Unlock()
92 + return b.batch.Put(key, val)
93 +}
94 +
95 +func (b *syncBatch) Delete(key ds.Key) error {
96 + b.lk.Lock()
97 + defer b.lk.Unlock()
98 + return b.batch.Delete(key)
99 +}
100 +
101 +func (b *syncBatch) Commit() error {
102 + b.lk.Lock()
103 + defer b.lk.Unlock()
104 + return b.batch.Commit()
105 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/test/assert.go new
+25
@@ -0,0 +1,25 @@
1 +package dstest
2 +
3 +import "testing"
4 +
5 +func Nil(err error, t *testing.T, msgs ...string) {
6 + if err != nil {
7 + t.Fatal(msgs, "error:", err)
8 + }
9 +}
10 +
11 +func True(v bool, t *testing.T, msgs ...string) {
12 + if !v {
13 + t.Fatal(msgs)
14 + }
15 +}
16 +
17 +func False(v bool, t *testing.T, msgs ...string) {
18 + True(!v, t, msgs...)
19 +}
20 +
21 +func Err(err error, t *testing.T, msgs ...string) {
22 + if err == nil {
23 + t.Fatal(msgs, "error:", err)
24 + }
25 +}
Godeps/_workspace/src/github.com/jbenet/go-datastore/test/test_util.go new
+99
@@ -0,0 +1,99 @@
1 +package dstest
2 +
3 +import (
4 + "bytes"
5 + "encoding/base32"
6 + "testing"
7 +
8 + rand "github.com/dustin/randbo"
9 + dstore "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
10 +)
11 +
12 +func RunBatchTest(t *testing.T, ds dstore.BatchingDatastore) {
13 + batch, err := ds.Batch()
14 + if err != nil {
15 + t.Fatal(err)
16 + }
17 +
18 + r := rand.New()
19 + var blocks [][]byte
20 + var keys []dstore.Key
21 + for i := 0; i < 20; i++ {
22 + blk := make([]byte, 256*1024)
23 + r.Read(blk)
24 + blocks = append(blocks, blk)
25 +
26 + key := dstore.NewKey(base32.StdEncoding.EncodeToString(blk[:8]))
27 + keys = append(keys, key)
28 +
29 + err := batch.Put(key, blk)
30 + if err != nil {
31 + t.Fatal(err)
32 + }
33 + }
34 +
35 + // Ensure they are not in the datastore before comitting
36 + for _, k := range keys {
37 + _, err := ds.Get(k)
38 + if err == nil {
39 + t.Fatal("should not have found this block")
40 + }
41 + }
42 +
43 + // commit, write them to the datastore
44 + err = batch.Commit()
45 + if err != nil {
46 + t.Fatal(err)
47 + }
48 +
49 + for i, k := range keys {
50 + blk, err := ds.Get(k)
51 + if err != nil {
52 + t.Fatal(err)
53 + }
54 +
55 + if !bytes.Equal(blk.([]byte), blocks[i]) {
56 + t.Fatal("blocks not correct!")
57 + }
58 + }
59 +}
60 +
61 +func RunBatchDeleteTest(t *testing.T, ds dstore.BatchingDatastore) {
62 + r := rand.New()
63 + var keys []dstore.Key
64 + for i := 0; i < 20; i++ {
65 + blk := make([]byte, 16)
66 + r.Read(blk)
67 +
68 + key := dstore.NewKey(base32.StdEncoding.EncodeToString(blk[:8]))
69 + keys = append(keys, key)
70 +
71 + err := ds.Put(key, blk)
72 + if err != nil {
73 + t.Fatal(err)
74 + }
75 + }
76 +
77 + batch, err := ds.Batch()
78 + if err != nil {
79 + t.Fatal(err)
80 + }
81 +
82 + for _, k := range keys {
83 + err := batch.Delete(k)
84 + if err != nil {
85 + t.Fatal(err)
86 + }
87 + }
88 + err = batch.Commit()
89 + if err != nil {
90 + t.Fatal(err)
91 + }
92 +
93 + for _, k := range keys {
94 + _, err := ds.Get(k)
95 + if err == nil {
96 + t.Fatal("shouldnt have found block")
97 + }
98 + }
99 +}
blocks/blockstore/blockstore.go
+22 -1
@@ -30,6 +30,7 @@ type Blockstore interface {
30 Has(key.Key) (bool, error)
31 Get(key.Key) (*blocks.Block, error)
32 Put(*blocks.Block) error
33 + PutMany([]*blocks.Block) error
34
35 AllKeysChan(ctx context.Context) (<-chan key.Key, error)
36 }
@@ -42,7 +43,7 @@ func NewBlockstore(d ds.ThreadSafeDatastore) Blockstore {
43 }
44
45 type blockstore struct {
45 - datastore ds.Datastore
46 + datastore ds.BatchingDatastore
47 // cant be ThreadSafeDatastore cause namespace.Datastore doesnt support it.
48 // we do check it on `NewBlockstore` though.
49 }
@@ -74,6 +75,26 @@ func (bs *blockstore) Put(block *blocks.Block) error {
75 return bs.datastore.Put(k, block.Data)
76 }
77
78 +func (bs *blockstore) PutMany(blocks []*blocks.Block) error {
79 + t, err := bs.datastore.Batch()
80 + if err != nil {
81 + return err
82 + }
83 + for _, b := range blocks {
84 + k := b.Key().DsKey()
85 + exists, err := bs.datastore.Has(k)
86 + if err == nil && exists {
87 + continue
88 + }
89 +
90 + err = t.Put(k, b.Data)
91 + if err != nil {
92 + return err
93 + }
94 + }
95 + return t.Commit()
96 +}
97 +
98 func (bs *blockstore) Has(k key.Key) (bool, error) {
99 return bs.datastore.Has(k.DsKey())
100 }
blocks/blockstore/blockstore_test.go
+4
@@ -266,3 +266,7 @@ func (c *queryTestDS) Query(q dsq.Query) (dsq.Results, error) {
266 }
267 return c.ds.Query(q)
268 }
269 +
270 +func (c *queryTestDS) Batch() (ds.Batch, error) {
271 + return ds.NewBasicBatch(c), nil
272 +}
blocks/blockstore/write_cache.go
+10
@@ -45,6 +45,16 @@ func (w *writecache) Put(b *blocks.Block) error {
45 return w.blockstore.Put(b)
46 }
47
48 +func (w *writecache) PutMany(bs []*blocks.Block) error {
49 + var good []*blocks.Block
50 + for _, b := range bs {
51 + if _, ok := w.cache.Get(b.Key()); !ok {
52 + good = append(good, b)
53 + }
54 + }
55 + return w.blockstore.PutMany(good)
56 +}
57 +
58 func (w *writecache) AllKeysChan(ctx context.Context) (<-chan key.Key, error) {
59 return w.blockstore.AllKeysChan(ctx)
60 }
blocks/blockstore/write_cache_test.go
+4
@@ -88,3 +88,7 @@ func (c *callbackDatastore) Query(q dsq.Query) (dsq.Results, error) {
88 c.f()
89 return c.ds.Query(q)
90 }
91 +
92 +func (c *callbackDatastore) Batch() (ds.Batch, error) {
93 + return ds.NewBasicBatch(c), nil
94 +}
blockservice/blockservice.go
+16
@@ -77,6 +77,22 @@ func (s *BlockService) AddBlock(b *blocks.Block) (key.Key, error) {
77 return k, nil
78 }
79
80 +func (s *BlockService) AddBlocks(bs []*blocks.Block) ([]key.Key, error) {
81 + err := s.Blockstore.PutMany(bs)
82 + if err != nil {
83 + return nil, err
84 + }
85 +
86 + var ks []key.Key
87 + for _, b := range bs {
88 + if err := s.worker.HasBlock(b); err != nil {
89 + return nil, errors.New("blockservice is closed")
90 + }
91 + ks = append(ks, b.Key())
92 + }
93 + return ks, nil
94 +}
95 +
96 // GetBlock retrieves a particular block from the service,
97 // Getting it from the datastore using the key (hash).
98 func (s *BlockService) GetBlock(ctx context.Context, k key.Key) (*blocks.Block, error) {
importer/balanced/builder.go
+11 -1
@@ -31,7 +31,17 @@ func BalancedLayout(db *h.DagBuilderHelper) (*dag.Node, error) {
31 root = h.NewUnixfsNode()
32 }
33
34 - return db.Add(root)
34 + out, err := db.Add(root)
35 + if err != nil {
36 + return nil, err
37 + }
38 +
39 + err = db.Close()
40 + if err != nil {
41 + return nil, err
42 + }
43 +
44 + return out, nil
45 }
46
47 // fillNodeRec will fill the given node with data from the dagBuilders input
importer/helpers/dagbuilder.go
+7
@@ -22,6 +22,8 @@ type DagBuilderHelper struct {
22 nextData []byte // the next item to return.
23 maxlinks int
24 ncb NodeCB
25 +
26 + batch *dag.Batch
27 }
28
29 type DagBuilderParams struct {
@@ -48,6 +50,7 @@ func (dbp *DagBuilderParams) New(in <-chan []byte) *DagBuilderHelper {
50 in: in,
51 maxlinks: dbp.Maxlinks,
52 ncb: ncb,
53 + batch: dbp.Dagserv.Batch(),
54 }
55 }
56
@@ -156,3 +159,7 @@ func (db *DagBuilderHelper) Add(node *UnixfsNode) (*dag.Node, error) {
159 func (db *DagBuilderHelper) Maxlinks() int {
160 return db.maxlinks
161 }
162 +
163 +func (db *DagBuilderHelper) Close() error {
164 + return db.batch.Commit()
165 +}
importer/helpers/helpers.go
+1 -1
@@ -107,7 +107,7 @@ func (n *UnixfsNode) AddChild(child *UnixfsNode, db *DagBuilderHelper) error {
107 return err
108 }
109
110 - _, err = db.dserv.Add(childnode)
110 + _, err = db.batch.Add(childnode)
111 if err != nil {
112 return err
113 }
importer/trickle/trickledag.go
+21 -2
@@ -36,7 +36,17 @@ func TrickleLayout(db *h.DagBuilderHelper) (*dag.Node, error) {
36 }
37 }
38
39 - return db.Add(root)
39 + out, err := db.Add(root)
40 + if err != nil {
41 + return nil, err
42 + }
43 +
44 + err = db.Close()
45 + if err != nil {
46 + return nil, err
47 + }
48 +
49 + return out, nil
50 }
51
52 func fillTrickleRec(db *h.DagBuilderHelper, node *h.UnixfsNode, depth int) error {
@@ -64,7 +74,16 @@ func fillTrickleRec(db *h.DagBuilderHelper, node *h.UnixfsNode, depth int) error
74 }
75
76 // TrickleAppend appends the data in `db` to the dag, using the Trickledag format
67 -func TrickleAppend(base *dag.Node, db *h.DagBuilderHelper) (*dag.Node, error) {
77 +func TrickleAppend(base *dag.Node, db *h.DagBuilderHelper) (out *dag.Node, err_out error) {
78 + defer func() {
79 + if err_out == nil {
80 + err := db.Close()
81 + if err != nil {
82 + err_out = err
83 + }
84 + }
85 + }()
86 +
87 // Convert to unixfs node for working with easily
88 ufsn, err := h.NewUnixfsNodeFromDag(base)
89 if err != nil {
merkledag/merkledag.go
+44
@@ -26,6 +26,8 @@ type DAGService interface {
26 // nodes of the passed in node.
27 GetDAG(context.Context, *Node) []NodeGetter
28 GetNodes(context.Context, []key.Key) []NodeGetter
29 +
30 + Batch() *Batch
31 }
32
33 func NewDAGService(bs *bserv.BlockService) DAGService {
@@ -62,6 +64,10 @@ func (n *dagService) Add(nd *Node) (key.Key, error) {
64 return n.Blocks.AddBlock(b)
65 }
66
67 +func (n *dagService) Batch() *Batch {
68 + return &Batch{ds: n, MaxSize: 8 * 1024 * 1024}
69 +}
70 +
71 // AddRecursive adds the given node and all child nodes to the BlockService
72 func (n *dagService) AddRecursive(nd *Node) error {
73 _, err := n.Add(nd)
@@ -269,3 +275,41 @@ func (np *nodePromise) Get(ctx context.Context) (*Node, error) {
275 }
276 return np.cache, nil
277 }
278 +
279 +type Batch struct {
280 + ds *dagService
281 +
282 + blocks []*blocks.Block
283 + size int
284 + MaxSize int
285 +}
286 +
287 +func (t *Batch) Add(nd *Node) (key.Key, error) {
288 + d, err := nd.Encoded(false)
289 + if err != nil {
290 + return "", err
291 + }
292 +
293 + b := new(blocks.Block)
294 + b.Data = d
295 + b.Multihash, err = nd.Multihash()
296 + if err != nil {
297 + return "", err
298 + }
299 +
300 + k := key.Key(b.Multihash)
301 +
302 + t.blocks = append(t.blocks, b)
303 + t.size += len(b.Data)
304 + if t.size > t.MaxSize {
305 + return k, t.Commit()
306 + }
307 + return k, nil
308 +}
309 +
310 +func (t *Batch) Commit() error {
311 + _, err := t.ds.Blocks.AddBlocks(t.blocks)
312 + t.blocks = nil
313 + t.size = 0
314 + return err
315 +}
util/datastore2/datastore_closer.go
+11
@@ -9,6 +9,8 @@ import (
9 type ThreadSafeDatastoreCloser interface {
10 datastore.ThreadSafeDatastore
11 io.Closer
12 +
13 + Batch() (datastore.Batch, error)
14 }
15
16 func CloserWrap(ds datastore.ThreadSafeDatastore) ThreadSafeDatastoreCloser {
@@ -22,3 +24,12 @@ type datastoreCloserWrapper struct {
24 func (w *datastoreCloserWrapper) Close() error {
25 return nil // no-op
26 }
27 +
28 +func (w *datastoreCloserWrapper) Batch() (datastore.Batch, error) {
29 + bds, ok := w.ThreadSafeDatastore.(datastore.BatchingDatastore)
30 + if !ok {
31 + return nil, datastore.ErrBatchUnsupported
32 + }
33 +
34 + return bds.Batch()
35 +}
util/datastore2/delayed.go
+4
@@ -41,4 +41,8 @@ func (dds *delayed) Query(q dsq.Query) (dsq.Results, error) {
41 return dds.ds.Query(q)
42 }
43
44 +func (dds *delayed) Batch() (ds.Batch, error) {
45 + return ds.NewBasicBatch(dds), nil
46 +}
47 +
48 var _ ds.Datastore = &delayed{}
util/datastore2/threadsafe.go
+1 -1
@@ -7,7 +7,7 @@ import (
7 // ClaimThreadSafe claims that a Datastore is threadsafe, even when
8 // it's type does not guarantee this. Use carefully.
9 type ClaimThreadSafe struct {
10 - datastore.Datastore
10 + datastore.BatchingDatastore
11 }
12
13 var _ datastore.ThreadSafeDatastore = ClaimThreadSafe{}