Almost finish blob stuff + do some todos in DataQueue (nodejs#20)

* Almost finish up blob work

* Finish some todos in DataQueue

src/dataqueue/queue.cc

@@ -41,6 +41,8 @@ using v8::Value;
 using v8::String;
 using v8::Global;
 using v8::Function;
+using v8::Int32;
+using v8::Uint32;
 
 namespace {
 // ============================================================================
@@ -959,8 +961,8 @@ class StreamEntry final : public EntryBase {
  // Creating the callback failed for whatever reason. The error will propagate
  // thank to the callback scope, but let's end the reader and fail this read.
  ended_ = true;
- std::move(next)(bob::STATUS_EOS, nullptr, 0, [](size_t) {});
- return bob::STATUS_EOS;
+ std::move(next)(bob::STATUS_FAILED, nullptr, 0, [](size_t) {});
+ return bob::STATUS_FAILED;
  }
 
  Local<Value> argv[] = { callback };
@@ -969,10 +971,9 @@ class StreamEntry final : public EntryBase {
  if (!pull->Call(isolate->GetCurrentContext(), wrap, arraysize(argv), argv).ToLocal(&ret)) {
  // The call failed for whatever reason. The error will propagate thanks to the
  // callback scope, but let's end the reader and fail this read.
- // TODO(@jasnell, @flakey5): Bob streams need a proper error status...
  ended_ = true;
- std::move(next)(bob::STATUS_EOS, nullptr, 0, [](size_t) {});
- return bob::STATUS_EOS;
+ std::move(next)(bob::STATUS_FAILED, nullptr, 0, [](size_t) {});
+ return bob::STATUS_FAILED;
  }
 
  return bob::STATUS_WAIT;
@@ -1386,10 +1387,13 @@ class FdEntry final : public EntryBase {
  CHECK(args.IsConstructCall());
  Environment* env = Environment::GetCurrent(args);
 
- // TODO(dataqueue): Get these from the arguments
- int fd = 0;
- size_t start = 0;
- size_t end = 0;
+ CHECK(args[0]->IsInt32());
+ CHECK(args[1]->IsUint32());
+ CHECK(args[2]->IsUint32());
+
+ int fd = args[0].As<Int32>()->Value();
+ size_t start = args[1].As<Uint32>()->Value();
+ size_t end = args[1].As<Uint32>()->Value();
 
  new Wrap(env, args.This(), fd, start, Just(end));
  }

src/node_blob.cc

@@ -71,7 +71,8 @@ bool Blob::HasInstance(Environment* env, v8::Local<v8::Value> object) {
  return GetConstructorTemplate(env)->HasInstance(object);
 }
 
-BaseObjectPtr<Blob> Blob::Create(Environment* env, std::shared_ptr<DataQueue> data_queue) {
+BaseObjectPtr<Blob> Blob::Create(
+ Environment* env, std::shared_ptr<DataQueue> data_queue) {
  HandleScope scope(env->isolate());
 
  Local<Function> ctor;
@@ -90,9 +91,8 @@ void Blob::New(const FunctionCallbackInfo<Value>& args) {
  CHECK(args[0]->IsArray()); // sources
  CHECK(args[1]->IsUint32()); // length
 
- // TODO(@flakey5): revisit when DataQueue is complete
+ // TODO(@flakey5): delete
  //std::vector<BlobEntry> entries;
-
  //size_t length = args[1].As<Uint32>()->Value();
  //size_t len = 0;
  //Local<Array> ary = args[0].As<Array>();
@@ -119,10 +119,32 @@ void Blob::New(const FunctionCallbackInfo<Value>& args) {
  //}
  //CHECK_EQ(length, len);
 
- // TODO(@flakey5): get dataqueue from js
- std::shared_ptr<DataQueue> data_queue = nullptr;
+ Local<Array> array = args[0].As<Array>();
+ size_t length = args[1].As<Uint32>()->Value();
+ std::vector<std::unique_ptr<DataQueue::Entry>> entries(length);
+
+ for (size_t i = 0; i < array->Length(); i++) {
+ Local<Value> entry;
+ if (!array->Get(env->context(), i).ToLocal(&entry)) {
+ return;
+ }
+
+ // TODO(@flakey5): check for different entry types
+ CHECK(entry->IsArrayBufferView() || Blob::HasInstance(env, entry));
+ if (entry->IsArrayBufferView()) {
+ Local<ArrayBufferView> view = entry.As<ArrayBufferView>();
+ CHECK_EQ(view->ByteOffset(), 0);
 
- BaseObjectPtr<Blob> blob = Create(env, data_queue);
+ entries[i] = DataQueue::CreateInMemoryEntryFromView(view);
+ } else {
+ Blob* blob;
+ ASSIGN_OR_RETURN_UNWRAP(&blob, entry);
+
+ entries[i] = DataQueue::CreateDataQueueEntry(blob->data_queue_);
+ }
+ }
+
+ BaseObjectPtr<Blob> blob = Create(env, DataQueue::CreateIdempotent(entries));
  if (blob)
  args.GetReturnValue().Set(blob->object());
 }
@@ -150,16 +172,37 @@ void Blob::ToSlice(const FunctionCallbackInfo<Value>& args) {
 }
 
 void Blob::MemoryInfo(MemoryTracker* tracker) const {
- tracker->TrackFieldWithSize("store", length());
+ tracker->TrackField("data_queue_", data_queue_);
 }
 
 MaybeLocal<Value> Blob::GetArrayBuffer(Environment* env) {
- // TODO(@flakey5): figure out how this'll work
  EscapableHandleScope scope(env->isolate());
  size_t len = length();
  std::shared_ptr<BackingStore> store =
  ArrayBuffer::NewBackingStore(env->isolate(), len);
 
+ if (len > 0) {
+ std::unique_ptr<DataQueue::Reader> reader = this->data_queue_->getReader();
+ DataQueue::Vec vec;
+ size_t offset = 0;
+ while (reader->Pull(
+ [&store, &offset](int,
+ const DataQueue::Vec* vec,
+ size_t count,
+ bob::Done done) {
+ for (size_t i = 0; i < count; i++) {
+ memcpy(
+ static_cast<uintptr_t*>(store->Data()) + offset,
+ vec[i].base,
+ vec[i].len);
+ offset += vec[i].len;
+ }
+ },
+ bob::Options::OPTIONS_NONE,
+ &vec,
+ len) != bob::Status::STATUS_EOS);
+ }
+
  /*if (len > 0) {
  unsigned char* dest = static_cast<unsigned char*>(store->Data());
  size_t total = 0;
@@ -281,7 +324,7 @@ void Blob::GetDataObject(const v8::FunctionCallbackInfo<v8::Value>& args) {
  }
 }
 
-// TODO(@flakey5): revisit when DataQueue is complete
+// TODO(@flakey5): delete
 //FixedSizeBlobCopyJob::FixedSizeBlobCopyJob(
 // Environment* env,
 // Local<Object> object,
@@ -471,7 +514,7 @@ void Blob::RegisterExternalReferences(ExternalReferenceRegistry* registry) {
  registry->Register(Blob::GetDataObject);
  registry->Register(Blob::RevokeDataObject);
 
- FixedSizeBlobCopyJob::RegisterExternalReferences(registry);
+ //FixedSizeBlobCopyJob::RegisterExternalReferences(registry);
 }
 
 } // namespace node

src/node_blob.h

@@ -85,7 +85,7 @@ class Blob : public BaseObject {
  std::shared_ptr<DataQueue> data_queue_;
 };
 
-// TODO(@flakey5): revisit when DataQueue is complete
+// TODO(@flakey5): delete
 //class FixedSizeBlobCopyJob : public AsyncWrap, public ThreadPoolWork {
 // public:
 // enum class Mode {

src/node_bob.h

@@ -10,6 +10,10 @@ constexpr size_t kMaxCountHint = 16;
 
 // Negative status codes indicate error conditions.
 enum Status : int {
+ // Indicates that there was an error while pulling.
+ // Should be treated similar to STATUS_EOS
+ STATUS_FAILED = -2,
+
  // Indicates that an attempt was made to pull after end.
  STATUS_EOS = -1,