quic: add streams implementation

nodejs · May 29, 2023 · 4bdb924 · 4bdb924
1 parent a907721
commit 4bdb924
Show file tree

Hide file tree

Showing 6 changed files with 1,188 additions and 42 deletions.
diff --git a/src/dataqueue/queue.cc b/src/dataqueue/queue.cc
@@ -176,8 +176,7 @@ class DataQueueImpl final : public DataQueue,
 
   void NotifyBackpressure(size_t amount) {
     if (idempotent_) return;
-    for (auto listener : backpressure_listeners_)
-      listener->EntryRead(amount);
+    for (auto listener : backpressure_listeners_) listener->EntryRead(amount);
   }
 
   bool has_backpressure_listeners() const {
@@ -457,8 +456,8 @@ class NonIdempotentDataQueueReader final
             return;
           }
 
-          // If there is a backpressure listener, lets report on how much data was
-          // actually read.
+          // If there is a backpressure listener, lets report on how much data
+          // was actually read.
           if (data_queue_->has_backpressure_listeners()) {
             // How much did we actually read?
             size_t read = 0;

diff --git a/src/quic/application.cc b/src/quic/application.cc
@@ -1,3 +1,4 @@
+#include "node_bob.h"
 #include "uv.h"
 #if HAVE_OPENSSL && NODE_OPENSSL_HAS_QUIC
 
@@ -79,7 +80,7 @@ void Session::Application::AcknowledgeStreamData(Stream* stream,
 
 void Session::Application::BlockStream(int64_t id) {
   auto stream = session().FindStream(id);
-  if (stream) stream->Blocked();
+  if (stream) stream->EmitBlocked();
 }
 
 bool Session::Application::CanAddHeader(size_t current_count,
@@ -233,7 +234,7 @@ void Session::Application::SendPendingData() {
           // and no more outbound data can be sent.
           CHECK_LE(ndatalen, 0);
           auto stream = session_->FindStream(stream_data.id);
-          if (stream) stream->End();
+          if (stream) stream->EndWritable();
           continue;
         }
         case NGTCP2_ERR_WRITE_MORE: {
@@ -360,10 +361,8 @@ class DefaultApplication final : public Session::Application {
                              stream_data->data,
                              arraysize(stream_data->data),
                              kMaxVectorCount);
-      switch (ret) {
-        case bob::Status::STATUS_EOS:
-          stream_data->fin = 1;
-          break;
+      if (ret == bob::Status::STATUS_EOS) {
+        stream_data->fin = 1;
       }
     } else {
       stream_data->fin = 1;

diff --git a/src/quic/bindingdata.cc b/src/quic/bindingdata.cc
@@ -204,8 +204,12 @@ CallbackScopeBase::CallbackScopeBase(Environment* env)
     : env(env), context_scope(env->context()), try_catch(env->isolate()) {}
 
 CallbackScopeBase::~CallbackScopeBase() {
-  if (try_catch.HasCaught() && !try_catch.HasTerminated()) {
-    errors::TriggerUncaughtException(env->isolate(), try_catch);
+  if (try_catch.HasCaught()) {
+    if (!try_catch.HasTerminated() && env->can_call_into_js()) {
+      errors::TriggerUncaughtException(env->isolate(), try_catch);
+    } else {
+      try_catch.ReThrow();
+    }
   }
 }
 

diff --git a/src/quic/bindingdata.h b/src/quic/bindingdata.h
@@ -103,7 +103,6 @@ constexpr size_t kMaxVectorCount = 16;
   V(session_version_negotiation, SessionVersionNegotiation)                    \
   V(session_path_validation, SessionPathValidation)                            \
   V(stream_close, StreamClose)                                                 \
-  V(stream_error, StreamError)                                                 \
   V(stream_created, StreamCreated)                                             \
   V(stream_reset, StreamReset)                                                 \
   V(stream_headers, StreamHeaders)                                             \
@@ -304,6 +303,8 @@ struct CallbackScopeBase {
   ~CallbackScopeBase();
 };
 
+// Maintains a strong reference to BaseObject type ptr to keep it alive during
+// a MakeCallback during which it might be destroyed.
 template <typename T>
 struct CallbackScope final : public CallbackScopeBase {
   BaseObjectPtr<T> ref;