Check readable data after ondata handler is assigned. (#1074)

* Check readable data after ondata handler is assigned.

If signaling data is arrived before ondata handler is assigned, it will
not fire ondata again since OnCanRead is not called when readable queue
is not empty.

* Check readable data in ondata setter.

Author: jianjunz

source/agent/addons/quic/QuicTransportStream.cc

@@ -58,6 +58,7 @@ QuicTransportStream::~QuicTransportStream()
     }
     m_stream->SetVisitor(nullptr);
     delete[] m_buffer;
+    m_onDataCallback.Reset();
 }
 
 void QuicTransportStream::OnCanRead()
@@ -123,6 +124,7 @@ NAN_MODULE_INIT(QuicTransportStream::init)
     Nan::SetPrototypeMethod(tpl, "readTrackId", readTrackId);
     Nan::SetPrototypeMethod(tpl, "addDestination", addDestination);
     Nan::SetAccessor(instanceTpl, Nan::New("isMedia").ToLocalChecked(), isMediaGetter, isMediaSetter);
+    Nan::SetAccessor(instanceTpl, Nan::New("ondata").ToLocalChecked(), onDataGetter, onDataSetter);
 
     s_constructor.Reset(Nan::GetFunction(tpl).ToLocalChecked());
     Nan::Set(target, Nan::New("QuicTransportStream").ToLocalChecked(), Nan::GetFunction(tpl).ToLocalChecked());
@@ -203,6 +205,13 @@ NAN_METHOD(QuicTransportStream::readTrackId){
     obj->ReadTrackId();
 }
 
+void QuicTransportStream::CheckReadableData()
+{
+    if (m_stream->ReadableBytes() > 0) {
+        SignalOnData();
+    }
+}
+
 NAN_GETTER(QuicTransportStream::isMediaGetter){
     QuicTransportStream* obj = Nan::ObjectWrap::Unwrap<QuicTransportStream>(info.Holder());
     info.GetReturnValue().Set(Nan::New(obj->m_isMedia));
@@ -214,6 +223,24 @@ NAN_SETTER(QuicTransportStream::isMediaSetter)
     obj->m_isMedia = value->ToBoolean(Nan::GetCurrentContext()).ToLocalChecked()->Value();
 }
 
+NAN_GETTER(QuicTransportStream::onDataGetter)
+{
+    QuicTransportStream* obj = Nan::ObjectWrap::Unwrap<QuicTransportStream>(info.Holder());
+    info.GetReturnValue().Set(Nan::New(obj->m_onDataCallback));
+}
+
+NAN_SETTER(QuicTransportStream::onDataSetter)
+{
+    QuicTransportStream* obj = Nan::ObjectWrap::Unwrap<QuicTransportStream>(info.Holder());
+    if (value->IsFunction()) {
+        Nan::Persistent<v8::Value> persistentCallback(value);
+        obj->m_onDataCallback.Reset(persistentCallback);
+        obj->CheckReadableData();
+    } else {
+        obj->m_onDataCallback.Reset();
+    }
+}
+
 v8::Local<v8::Object> QuicTransportStream::newInstance(owt::quic::WebTransportStreamInterface* stream)
 {
     Local<Object> streamObject = Nan::NewInstance(Nan::New(QuicTransportStream::s_constructor)).ToLocalChecked();
@@ -273,6 +300,7 @@ NAUV_WORK_CB(QuicTransportStream::onData)
     if (obj == nullptr) {
         return;
     }
+    // TODO: Check m_onDataCallback instead of reading ondata.
     Nan::MaybeLocal<v8::Value> onEvent = Nan::Get(obj->handle(), Nan::New<v8::String>("ondata").ToLocalChecked());
     if (!onEvent.IsEmpty()) {
         v8::Local<v8::Value> onEventLocal = onEvent.ToLocalChecked();

source/agent/addons/quic/QuicTransportStream.h

@@ -42,6 +42,8 @@ class QuicTransportStream : public owt_base::FrameSource, public owt_base::Frame
 
     static NAN_GETTER(isMediaGetter);
     static NAN_SETTER(isMediaSetter);
+    static NAN_GETTER(onDataGetter);
+    static NAN_SETTER(onDataSetter);
 
     static NAUV_WORK_CB(onContentSessionId);
     static NAUV_WORK_CB(onTrackId);
@@ -78,6 +80,8 @@ class QuicTransportStream : public owt_base::FrameSource, public owt_base::Frame
     void ReadTrackId();
     void SignalOnData();
     void ReallocateBuffer(size_t size);
+    // Check whether there is readable data. If so, fire ondata event.
+    void CheckReadableData();
     void AddedDestination();
     void RemovedDestination();
 
@@ -96,6 +100,7 @@ class QuicTransportStream : public owt_base::FrameSource, public owt_base::Frame
 
     // Indicates whether this is a media stream. If this is not a media stream, it can only be piped to another QUIC agent.
     bool m_isMedia;
+    Nan::Persistent<v8::Value> m_onDataCallback;
 
     size_t m_readingFrameSize;
     size_t m_frameSizeOffset;

source/agent/quic/webtransport/quicTransportServer.js

@@ -151,7 +151,7 @@ module.exports = class QuicTransportServer extends EventEmitter {
             }
             return;
           }
-        }
+        };
       }
     }
   }