modules/kvs: add kvs_get_dirat, kvs_get_symlinkat

Add kvs_get_dirat() and kvs_get_symlinkat(), which are
like kvs_get_dir() and kvs_get_symlink() but lookup 'key'
starting at 'treeobj'.

These functions are sufficient to be able to walk a KVS
snapshot recursively.

src/modules/kvs/kvs.h

@@ -51,6 +51,10 @@ int kvs_get_treeobj (flux_t h, const char *key, char **treeobj);
  */
 int kvs_getat (flux_t h, const char *treeobj,
                const char *key, char **json_str);
+int kvs_get_dirat (flux_t h, const char *treeobj,
+                   const char *key, kvsdir_t **dirp);
+int kvs_get_symlinkat (flux_t h, const char *treeobj,
+                               const char *key, char **val);
 
 /* kvs_watch* is like kvs_get* except the registered callback is called
  * to set the value.  It will be called immediately to set the initial

src/modules/kvs/libkvs.c

@@ -55,6 +55,7 @@
 
 struct kvsdir_struct {
     flux_t handle;
+    char *treeobj;
     char *key;
     json_object *o;
     int count;
@@ -209,17 +210,22 @@ void kvsdir_incref (kvsdir_t *dir)
 void kvsdir_destroy (kvsdir_t *dir)
 {
     if (--dir->usecount == 0) {
+        if (dir->treeobj)
+            free (dir->treeobj);
         free (dir->key);
         json_object_put (dir->o);
         free (dir);
     }
 }
 
-static kvsdir_t *kvsdir_alloc (flux_t handle, const char *key, json_object *o)
+static kvsdir_t *kvsdir_alloc (flux_t handle, const char *treeobj,
+                               const char *key, json_object *o)
 {
     kvsdir_t *dir = xzmalloc (sizeof (*dir));
 
     dir->handle = handle;
+    if (treeobj)
+        dir->treeobj = xstrdup (treeobj);
     dir->key = xstrdup (key);
     dir->o = o;
     json_object_get (dir->o);
@@ -270,7 +276,7 @@ kvsitr_t *kvsitr_create (kvsdir_t *dir)
 
 void kvsitr_destroy (kvsitr_t *itr)
 {
-    free (itr); 
+    free (itr);
 }
 
 const char *kvsitr_next (kvsitr_t *itr)
@@ -403,6 +409,55 @@ int kvs_getat (flux_t h, const char *treeobj,
     return -1;
 }
 
+int kvs_get_dirat (flux_t h, const char *treeobj,
+                   const char *key, kvsdir_t **dir)
+{
+    JSON v = NULL;
+    JSON dirent = NULL;
+    int rc = -1;
+
+    if (!treeobj || !key || !dir || !(dirent = Jfromstr (treeobj))
+                         || dirent_validate (dirent) < 0) {
+        errno = EINVAL;
+        goto done;
+    }
+    if (getobj (h, dirent, key, KVS_PROTO_READDIR, &v) < 0)
+        goto done;
+    *dir = kvsdir_alloc (h, treeobj, key, v);
+    rc = 0;
+done:
+    Jput (v);
+    Jput (dirent);
+    return rc;
+}
+
+int kvs_get_symlinkat (flux_t h, const char *treeobj,
+                       const char *key, char **val)
+{
+    JSON v = NULL;
+    JSON dirent = NULL;
+    int rc = -1;
+
+    if (!treeobj || !key || !(dirent = Jfromstr (treeobj))
+                         || dirent_validate (dirent) < 0) {
+        errno = EINVAL;
+        goto done;
+    }
+    if (getobj (h, dirent, key, KVS_PROTO_READLINK, &v) < 0)
+        goto done;
+    if (json_object_get_type (v) != json_type_string) {
+        errno = EPROTO;
+        goto done;
+    }
+    if (val)
+        *val = xstrdup (json_object_get_string (v));
+    rc = 0;
+done:
+    Jput (v);
+    Jput (dirent);
+    return rc;
+}
+
 /* deprecated */
 int kvs_get_obj (flux_t h, const char *key, JSON *val)
 {
@@ -441,7 +496,7 @@ int kvs_get_dir (flux_t h, kvsdir_t **dir, const char *fmt, ...)
     }
     if (kp_rget_dec (out, &v) < 0)
         goto done;
-    *dir = kvsdir_alloc (h, k, v);
+    *dir = kvsdir_alloc (h, NULL, k, v);
     rc = 0;
 done:
     Jput (in);
@@ -703,7 +758,7 @@ static int dispatch_watch (flux_t h, kvs_watcher_t *wp, json_object *val)
         }
         case WATCH_DIR: {
             kvs_set_dir_f set = wp->set;
-            kvsdir_t *dir = val ? kvsdir_alloc (h, wp->key, val) : NULL;
+            kvsdir_t *dir = val ? kvsdir_alloc (h, NULL, wp->key, val) : NULL;
             rc = set (wp->key, dir, wp->arg, errnum);
             if (dir)
                 kvsdir_destroy (dir);
@@ -902,7 +957,7 @@ int kvs_watch_once_dir (flux_t h, kvsdir_t **dirp, const char *fmt, ...)
     }
     if (*dirp)
         kvsdir_destroy (*dirp);
-    *dirp = kvsdir_alloc (h, key, val);
+    *dirp = kvsdir_alloc (h, NULL, key, val);
     rc = 0;
 done:
     if (val)