http2: refactor read mechanism

lib/internal/http2/core.js

@@ -279,16 +279,15 @@ function submitRstStream(code) {
 // point, close them. If there is an open fd for file send, close that also.
 // At this point the underlying node::http2:Http2Stream handle is no
 // longer usable so destroy it also.
-function onStreamClose(code, hasData) {
+function onStreamClose(code) {
   const stream = this[kOwner];
   if (stream.destroyed)
     return;
 
   const state = stream[kState];
 
   debug(`Http2Stream ${stream[kID]} [Http2Session ` +
-        `${sessionName(stream[kSession][kType])}]: closed with code ${code}` +
-        ` [has data? ${hasData}]`);
+        `${sessionName(stream[kSession][kType])}]: closed with code ${code}`);
 
   if (!stream.closed) {
     // Clear timeout and remove timeout listeners
@@ -306,21 +305,22 @@ function onStreamClose(code, hasData) {
 
   if (state.fd !== undefined)
     tryClose(state.fd);
-  stream[kMaybeDestroy](null, code, hasData);
+  stream.push(null);
+  stream[kMaybeDestroy](null, code);
 }
 
 // Receives a chunk of data for a given stream and forwards it on
 // to the Http2Stream Duplex for processing.
-function onStreamRead(nread, buf, handle) {
-  const stream = handle[kOwner];
+function onStreamRead(nread, buf) {
+  const stream = this[kOwner];
   if (nread >= 0 && !stream.destroyed) {
     debug(`Http2Stream ${stream[kID]} [Http2Session ` +
           `${sessionName(stream[kSession][kType])}]: receiving data chunk ` +
           `of size ${nread}`);
     stream[kUpdateTimer]();
     if (!stream.push(buf)) {
       if (!stream.destroyed)  // we have to check a second time
-        handle.readStop();
+        this.readStop();
     }
     return;
   }
@@ -1431,13 +1431,8 @@ function streamOnResume() {
 }
 
 function streamOnPause() {
-  // if (!this.destroyed && !this.pending)
-  //   this[kHandle].readStop();
-}
-
-function handleFlushData(self) {
   if (!this.destroyed && !this.pending)
-    this[kHandle].flushData();
+    this[kHandle].readStop();
 }
 
 // If the writable side of the Http2Stream is still open, emit the
@@ -1686,11 +1681,10 @@ class Http2Stream extends Duplex {
       this.push(null);
       return;
     }
-    const flushfn = handleFlushData.bind(this);
     if (!this.pending) {
-      flushfn();
+      streamOnResume.call(this);
     } else {
-      this.once('ready', flushfn);
+      this.once('ready', streamOnResume);
     }
   }
 
@@ -1831,10 +1825,10 @@ class Http2Stream extends Duplex {
 
   // The Http2Stream can be destroyed if it has closed and if the readable
   // side has received the final chunk.
-  [kMaybeDestroy](error, code = NGHTTP2_NO_ERROR, hasData = true) {
+  [kMaybeDestroy](error, code = NGHTTP2_NO_ERROR) {
     if (error == null) {
       if (code === NGHTTP2_NO_ERROR &&
-          ((!this._readableState.ended && hasData) ||
+          (!this._readableState.ended ||
           !this._writableState.ended ||
           this._writableState.pendingcb > 0 ||
           !this.closed)) {

src/node_buffer.cc

@@ -303,15 +303,14 @@ MaybeLocal<Object> New(Environment* env, size_t length) {
         data,
         length,
         ArrayBufferCreationMode::kInternalized);
-  Local<Uint8Array> ui = Uint8Array::New(ab, 0, length);
-  Maybe<bool> mb =
-      ui->SetPrototype(env->context(), env->buffer_prototype_object());
-  if (mb.FromMaybe(false))
-    return scope.Escape(ui);
+  MaybeLocal<Uint8Array> ui = Buffer::New(env, ab, 0, length);
 
-  // Object failed to be created. Clean up resources.
-  free(data);
-  return Local<Object>();
+  if (ui.IsEmpty()) {
+    // Object failed to be created. Clean up resources.
+    free(data);
+  }
+
+  return scope.Escape(ui.FromMaybe(Local<Uint8Array>()));
 }
 
 
@@ -349,15 +348,14 @@ MaybeLocal<Object> Copy(Environment* env, const char* data, size_t length) {
         new_data,
         length,
         ArrayBufferCreationMode::kInternalized);
-  Local<Uint8Array> ui = Uint8Array::New(ab, 0, length);
-  Maybe<bool> mb =
-      ui->SetPrototype(env->context(), env->buffer_prototype_object());
-  if (mb.FromMaybe(false))
-    return scope.Escape(ui);
+  MaybeLocal<Uint8Array> ui = Buffer::New(env, ab, 0, length);
 
-  // Object failed to be created. Clean up resources.
-  free(new_data);
-  return Local<Object>();
+  if (ui.IsEmpty()) {
+    // Object failed to be created. Clean up resources.
+    free(new_data);
+  }
+
+  return scope.Escape(ui.FromMaybe(Local<Uint8Array>()));
 }
 
 
@@ -392,15 +390,14 @@ MaybeLocal<Object> New(Environment* env,
   // correct.
   if (data == nullptr)
     ab->Neuter();
-  Local<Uint8Array> ui = Uint8Array::New(ab, 0, length);
-  Maybe<bool> mb =
-      ui->SetPrototype(env->context(), env->buffer_prototype_object());
+  MaybeLocal<Uint8Array> ui = Buffer::New(env, ab, 0, length);
 
-  if (!mb.FromMaybe(false))
+  if (ui.IsEmpty()) {
     return Local<Object>();
+  }
 
   CallbackInfo::New(env->isolate(), ab, callback, data, hint);
-  return scope.Escape(ui);
+  return scope.Escape(ui.ToLocalChecked());
 }
 
 
@@ -415,8 +412,6 @@ MaybeLocal<Object> New(Isolate* isolate, char* data, size_t length) {
 
 
 MaybeLocal<Object> New(Environment* env, char* data, size_t length) {
-  EscapableHandleScope scope(env->isolate());
-
   if (length > 0) {
     CHECK_NE(data, nullptr);
     CHECK(length <= kMaxLength);
@@ -427,12 +422,7 @@ MaybeLocal<Object> New(Environment* env, char* data, size_t length) {
                        data,
                        length,
                        ArrayBufferCreationMode::kInternalized);
-  Local<Uint8Array> ui = Uint8Array::New(ab, 0, length);
-  Maybe<bool> mb =
-      ui->SetPrototype(env->context(), env->buffer_prototype_object());
-  if (mb.FromMaybe(false))
-    return scope.Escape(ui);
-  return Local<Object>();
+  return Buffer::New(env, ab, 0, length).FromMaybe(Local<Object>());
 }
 
 namespace {