libutil: add an async nar indexer

Change-Id: I4c22364c94e2bea1f18b835d9514c4c65a530c80

lix/libutil/archive.cc

@@ -752,28 +752,18 @@ Generator<Entry> parse(Source & source)
 namespace nar_index {
 
 namespace {
-struct NarSource : Source
+struct NarPositioner
 {
-    Source & source;
     uint64_t pos = 0;
-
-    NarSource(Source & source) : source(source) {}
-
-    size_t read(char * data, size_t len) override
-    {
-        auto n = source.read(data, len);
-        pos += n;
-        return n;
-    }
 };
 
 struct Indexer : NARParseVisitor
 {
-    NarSource & source;
+    NarPositioner & source;
     Directory & parent;
 
 public:
-    Indexer(NarSource & source, Directory & parent) : source(source), parent(parent) {}
+    Indexer(NarPositioner & source, Directory & parent) : source(source), parent(parent) {}
 
     box_ptr<NARParseVisitor> createDirectory(const std::string & name) override
     {
@@ -803,6 +793,20 @@ public:
 
 Entry create(Source & source)
 {
+    struct NarSource : Source, NarPositioner
+    {
+        Source & source;
+
+        NarSource(Source & source) : source(source) {}
+
+        size_t read(char * data, size_t len) override
+        {
+            auto n = source.read(data, len);
+            pos += n;
+            return n;
+        }
+    };
+
     Directory root;
     NarSource wrapper{source};
     Indexer index{wrapper, root};
@@ -810,6 +814,33 @@ Entry create(Source & source)
     return root.contents.at("");
 }
 
+kj::Promise<Result<Entry>> create(AsyncInputStream & source)
+try {
+    struct NarSource : AsyncInputStream, NarPositioner
+    {
+        AsyncInputStream & source;
+
+        NarSource(AsyncInputStream & source) : source(source) {}
+
+        kj::Promise<Result<size_t>> read(void * data, size_t len) override
+        try {
+            auto n = TRY_AWAIT(source.read(data, len));
+            pos += n;
+            co_return n;
+        } catch (...) {
+            co_return result::current_exception();
+        }
+    };
+
+    Directory root;
+    NarSource wrapper{source};
+    Indexer index{wrapper, root};
+    TRY_AWAIT(parseDump(index, wrapper));
+    co_return root.contents.at("");
+} catch (...) {
+    co_return result::current_exception();
+}
+
 }
 
 static void restore(NARParseVisitor & sink, nar::Entry entry, const Path & path)

lix/libutil/archive.hh

@@ -200,6 +200,7 @@ struct Directory
 };
 
 Entry create(Source & source);
+kj::Promise<Result<Entry>> create(AsyncInputStream & source);
 
 }
 

tests/unit/libutil/archive.cc

@@ -364,6 +364,26 @@ TEST_P(NarTest, index)
     }
 }
 
+TEST_P(NarTest, indexAsync)
+{
+    auto & [raw, entriesF] = GetParam();
+    AsyncStringInputStream source(raw);
+
+    kj::EventLoop el;
+    kj::WaitScope ws{el};
+    auto entries = entriesF();
+    auto indexed = fromIndex(raw, nar_index::create(source).wait(ws).value());
+    while (true) {
+        auto e = entries.next();
+        auto p = indexed.next();
+        ASSERT_EQ(e.has_value(), p.has_value());
+        if (!e) {
+            break;
+        }
+        assert_eq(*e, *p);
+    }
+}
+
 INSTANTIATE_TEST_SUITE_P(
     ,
     NarTest,