Merge pull request #19 from gierschv/feat-omap-iterator

ioctx: ListOmapValues & GetAllOmapValues added

rados/ioctx.go

@@ -380,64 +380,117 @@ func (ioctx *IOContext) SetOmap(oid string, pairs map[string][]byte) error {
         }
 }
 
+// OmapListFunc is the type of the function called for each omap key
+// visited by ListOmapValues
+type OmapListFunc func(key string, value []byte)
+
+// Iterate on a set of keys and their values from an omap
+// `startAfter`: iterate only on the keys after this specified one
+// `filterPrefix`: iterate only on the keys beginning with this prefix
+// `maxReturn`: iterate no more than `maxReturn` key/value pairs
+// `listFn`: the function called at each iteration
+func (ioctx *IOContext) ListOmapValues(oid string, startAfter string, filterPrefix string, maxReturn int64, listFn OmapListFunc) error {
+	c_oid := C.CString(oid)
+	c_start_after := C.CString(startAfter)
+	c_filter_prefix := C.CString(filterPrefix)
+	c_max_return := C.uint64_t(maxReturn)
+
+	defer C.free(unsafe.Pointer(c_oid))
+	defer C.free(unsafe.Pointer(c_start_after))
+	defer C.free(unsafe.Pointer(c_filter_prefix))
+
+	op := C.rados_create_read_op()
+
+	var c_iter C.rados_omap_iter_t
+	var c_prval C.int
+	C.rados_read_op_omap_get_vals(
+		op,
+		c_start_after,
+		c_filter_prefix,
+		c_max_return,
+		&c_iter,
+		&c_prval,
+	)
+
+	ret := C.rados_read_op_operate(op, ioctx.ioctx, c_oid, 0)
+
+	if int(c_prval) != 0 {
+		return RadosError(int(c_prval))
+	} else if int(ret) != 0 {
+		return RadosError(int(ret))
+	}
+
+	for {
+		var c_key *C.char
+		var c_val *C.char
+		var c_len C.size_t
+
+		ret = C.rados_omap_get_next(c_iter, &c_key, &c_val, &c_len)
+
+		if int(ret) != 0 {
+			return RadosError(int(ret))
+		}
+
+		if c_key == nil {
+			break
+		}
+
+		listFn(C.GoString(c_key), C.GoBytes(unsafe.Pointer(c_val), C.int(c_len)))
+	}
+
+	C.rados_omap_get_end(c_iter)
+	C.rados_release_read_op(op)
+
+	return nil
+}
+
 // Fetch a set of keys and their values from an omap and returns then as a map
 // `startAfter`: retrieve only the keys after this specified one
 // `filterPrefix`: retrieve only the keys beginning with this prefix
 // `maxReturn`: retrieve no more than `maxReturn` key/value pairs
 func (ioctx *IOContext) GetOmapValues(oid string, startAfter string, filterPrefix string, maxReturn int64) (map[string][]byte, error) {
-        c_oid := C.CString(oid)
-        c_start_after := C.CString(startAfter)
-        c_filter_prefix := C.CString(filterPrefix)
-        c_max_return := C.uint64_t(maxReturn)
+	omap := map[string][]byte{}
 
-        defer C.free(unsafe.Pointer(c_oid))
-        defer C.free(unsafe.Pointer(c_start_after))
-        defer C.free(unsafe.Pointer(c_filter_prefix))
+	err := ioctx.ListOmapValues(
+		oid, startAfter, filterPrefix, maxReturn,
+		func(key string, value []byte) {
+			omap[key] = value
+		},
+	)
 
-        op := C.rados_create_read_op()
+	return omap, err
+}
 
-        var c_iter C.rados_omap_iter_t
-        var c_prval C.int
-        C.rados_read_op_omap_get_vals(
-                op,
-                c_start_after,
-                c_filter_prefix,
-                c_max_return,
-                &c_iter,
-                &c_prval,
-        )
+// Fetch all the keys and their values from an omap and returns then as a map
+// `startAfter`: retrieve only the keys after this specified one
+// `filterPrefix`: retrieve only the keys beginning with this prefix
+// `iteratorSize`: internal number of keys to fetch during a read operation
+func (ioctx *IOContext) GetAllOmapValues(oid string, startAfter string, filterPrefix string, iteratorSize int64) (map[string][]byte, error) {
+	omap := map[string][]byte{}
+	omapSize := 0
 
-        ret := C.rados_read_op_operate(op, ioctx.ioctx, c_oid, 0)
-        omap := map[string][]byte{}
+	for {
+		err := ioctx.ListOmapValues(
+			oid, startAfter, filterPrefix, iteratorSize,
+			func (key string, value []byte) {
+				omap[key] = value
+				startAfter = key
+			},
+		)
 
-        if int(c_prval) != 0 {
-                return omap, RadosError(int(c_prval))
-        } else if int(ret) != 0 {
-                return omap, RadosError(int(ret))
-        }
+		if err != nil {
+			return omap, err
+		}
 
-        for {
-                var c_key *C.char
-                var c_val *C.char
-                var c_len C.size_t
+		// End of omap
+		if len(omap) == omapSize {
+			break
+		}
 
-                ret = C.rados_omap_get_next(c_iter, &c_key, &c_val, &c_len)
+		omapSize = len(omap)
+	}
 
-                if int(ret) != 0 {
-                        return omap, RadosError(int(ret))
-                }
-
-                if c_key == nil {
-                        break
-                }
-
-                omap[C.GoString(c_key)] = C.GoBytes(unsafe.Pointer(c_val), C.int(c_len))
-        }
-
-        C.rados_omap_get_end(c_iter)
-        C.rados_release_read_op(op)
-
-        return omap, nil
+	return omap, nil
 }
 
 // Remove the specified `keys` from the omap `oid`

rados/rados_test.go

@@ -605,11 +605,34 @@ func TestReadWriteOmap(t *testing.T) {
 	err = pool.SetOmap("obj", orig)
 	assert.NoError(t, err)
 
-	// Get
+	// List
+	remaining := map[string][]byte{}
+	for k, v := range orig {
+		remaining[k] = v
+	}
+
+	err = pool.ListOmapValues("obj", "", "", 4, func(key string, value []byte) {
+		assert.Equal(t, remaining[key], value)
+		delete(remaining, key)
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, 0, len(remaining))
+
+	// Get (with a fixed number of keys)
 	fetched, err := pool.GetOmapValues("obj", "",  "", 4)
 	assert.NoError(t, err)
 	assert.Equal(t, orig, fetched)
 
+	// Get All (with an iterator size bigger than the map size)
+	fetched, err = pool.GetAllOmapValues("obj", "",  "", 100)
+	assert.NoError(t, err)
+	assert.Equal(t, orig, fetched)
+
+	// Get All (with an iterator size smaller than the map size)
+	fetched, err = pool.GetAllOmapValues("obj", "",  "", 1)
+	assert.NoError(t, err)
+	assert.Equal(t, orig, fetched)
+
 	// Remove
 	err = pool.RmOmapKeys("obj", []string{"key1", "prefixed-key3"})
 	assert.NoError(t, err)