quic: Change Blob Reader to coalesce buffers

Author: martenrichter

src/node_blob.cc

@@ -350,6 +350,9 @@ void Blob::Reader::Pull(const FunctionCallbackInfo<Value>& args) {
     BaseObjectPtr<Blob::Reader> reader;
     Global<Function> callback;
     Environment* env;
+    std::vector<DataQueue::Vec> vecs;
+    std::list<bob::Done> dones;
+    bool innext = false;
   };
   // TODO(@jasnell): A unique_ptr is likely better here but making this a unique
   // pointer that is passed into the lambda causes the std::move(next) below to
@@ -375,22 +378,61 @@ void Blob::Reader::Pull(const FunctionCallbackInfo<Value>& args) {
     }
 
     if (count > 0) {
+      impl->vecs.insert(
+        impl->vecs.end(),
+        vecs,
+        vecs + count);
+      impl->dones.push_back(std::move(doneCb));
+    }
+    if (impl->innext) {
+      CHECK(status != bob::STATUS_WAIT);
+      return;
+    }
+    if (status == bob::STATUS_CONTINUE) {
+      // We pull some more,
+      // but it must be sync as we
+      // merge it together
+      impl->innext = true;
+      while (status == bob::STATUS_CONTINUE) {
+        auto snext = [impl](int status,
+                            const DataQueue::Vec* vecs,
+                            size_t count,
+                            bob::Done doneCb) {
+          if (count > 0) {
+            impl->vecs.insert(impl->vecs.end(), vecs, vecs + count);
+            impl->dones.push_back(std::move(doneCb));
+          }
+        };
+        status = impl->reader->inner_->Pull(std::move(snext),
+                                            node::bob::OPTIONS_SYNC,
+                                            nullptr,
+                                            0);
+      }
+    }
+    // otherwise we commit and call
+    if (impl->vecs.size() > 0) {
       // Copy the returns vectors into a single ArrayBuffer.
       size_t total = 0;
-      for (size_t n = 0; n < count; n++) total += vecs[n].len;
+      const size_t vecs_size = impl->vecs.size();
+      const DataQueue::Vec* ivecs = &(*impl->vecs.begin());
+      for (size_t n = 0; n < vecs_size; n++) total += ivecs[n].len;
 
       std::shared_ptr<BackingStore> store = ArrayBuffer::NewBackingStore(
           env->isolate(),
           total,
           BackingStoreInitializationMode::kUninitialized);
       auto ptr = static_cast<uint8_t*>(store->Data());
-      for (size_t n = 0; n < count; n++) {
-        std::copy(vecs[n].base, vecs[n].base + vecs[n].len, ptr);
-        ptr += vecs[n].len;
+      for (size_t n = 0; n < vecs_size; n++) {
+        std::copy(ivecs[n].base, ivecs[n].base + ivecs[n].len, ptr);
+        ptr += ivecs[n].len;
       }
       // Since we copied the data buffers, signal that we're done with them.
-      std::move(doneCb)(0);
-      Local<Value> argv[2] = {Uint32::New(env->isolate(), status),
+      std::for_each(impl->dones.begin(),
+                    impl->dones.end(),
+                    [](bob::Done& done) {
+        std::move(done)(0);
+      });
+      Local<Value> argv[2] = {Uint32::New(env->isolate(), bob::STATUS_CONTINUE),
                               ArrayBuffer::New(env->isolate(), store)};
       impl->reader->MakeCallback(fn, arraysize(argv), argv);
       return;