Add wasm APIs for RTP depacketizing.

jianjunz · jianjunz · commit 0f82778ab494 · 2021-10-13T14:36:14.000+08:00
diff --git a/talk/owt/BUILD.gn b/talk/owt/BUILD.gn
@@ -340,6 +340,8 @@ if (is_wasm) {
       "//third_party/webrtc/call",
       "//third_party/webrtc/modules/rtp_rtcp:rtp_rtcp",
       "//third_party/webrtc/rtc_base:rtc_json",
+      "//third_party/webrtc/rtc_base:rtc_task_queue_stdlib",
+      "//third_party/webrtc/api/rtc_event_log:rtc_event_log_factory",
     ]
     sources = [
       "sdk/wasm/binding.h",
diff --git a/talk/owt/sdk/wasm/binding.h b/talk/owt/sdk/wasm/binding.h
@@ -13,9 +13,16 @@ namespace owt {
 namespace wasm {
 
 EMSCRIPTEN_BINDINGS(Owt) {
-  emscripten::class_<MediaSession>("MediaSession").constructor<>();
+  emscripten::class_<MediaSession>("MediaSession")
+      .constructor<>()
+      .function("createRtpVideoReceiver",
+                &MediaSession::CreateRtpVideoReceiver);
   emscripten::class_<RtpVideoReceiver>("RtpVideoReceiver")
-      .function("onRtpPacket", &RtpVideoReceiver::OnRtpPacket);
+      .smart_ptr<std::shared_ptr<RtpVideoReceiver>>("RtpVideoReceiver")
+      .function("onRtpPacket", &RtpVideoReceiver::OnRtpPacket,
+                emscripten::allow_raw_pointers())
+      .function("setCompleteFrameCallback",
+                &RtpVideoReceiver::SetCompleteFrameCallback);
 }
 
 }  // namespace wasm
diff --git a/talk/owt/sdk/wasm/gn/BUILD.gn b/talk/owt/sdk/wasm/gn/BUILD.gn
@@ -65,23 +65,23 @@ template("gcc_like_toolchain") {
 
     tool("cc") {
       depfile = "{{output}}.d"
-      command = "$cc_wrapper $cc -MMD -MF $depfile {{defines}} {{include_dirs}} {{cflags}} {{cflags_c}} -DEMSCRIPTEN_HAS_UNBOUND_TYPE_NAMES=0 --bind -fno-stack-protector ${external_cflags} -c {{source}} -o {{output}}"
+      command = "$cc_wrapper $cc -MMD -MF $depfile {{defines}} {{include_dirs}} {{cflags}} {{cflags_c}} -DEMSCRIPTEN_HAS_UNBOUND_TYPE_NAMES=1 --bind -fno-stack-protector ${external_cflags} -c {{source}} -o {{output}}"
       depsformat = "gcc"
       outputs = [ "$object_subdir/{{source_name_part}}.o" ]
       description = "compile {{source}}"
     }
 
     tool("cxx") {
       depfile = "{{output}}.d"
-      command = "$cc_wrapper $cxx -MMD -MF $depfile {{defines}} {{include_dirs}} {{cflags}} {{cflags_cc}} -DEMSCRIPTEN_HAS_UNBOUND_TYPE_NAMES=0 --bind -fno-stack-protector ${external_cflags} ${external_cxxflags} -c {{source}} -o {{output}}"
+      command = "$cc_wrapper $cxx -MMD -MF $depfile {{defines}} {{include_dirs}} {{cflags}} {{cflags_cc}} -DEMSCRIPTEN_HAS_UNBOUND_TYPE_NAMES=1 --bind -fno-stack-protector ${external_cflags} ${external_cxxflags} -c {{source}} -o {{output}}"
       depsformat = "gcc"
       outputs = [ "$object_subdir/{{source_name_part}}.o" ]
       description = "compile {{source}}"
     }
 
     tool("asm") {
       depfile = "{{output}}.d"
-      command = "$cc_wrapper $cc -MMD -MF $depfile {{defines}} {{include_dirs}} {{asmflags}} -DEMSCRIPTEN_HAS_UNBOUND_TYPE_NAMES=0 --bind -fno-stack-protector -c {{source}} -o {{output}}"
+      command = "$cc_wrapper $cc -MMD -MF $depfile {{defines}} {{include_dirs}} {{asmflags}} -DEMSCRIPTEN_HAS_UNBOUND_TYPE_NAMES=1 --bind -fno-stack-protector -c {{source}} -o {{output}}"
       depsformat = "gcc"
       outputs = [ "$object_subdir/{{source_name_part}}.o" ]
       description = "assemble {{source}}"
@@ -142,7 +142,7 @@ template("gcc_like_toolchain") {
     tool("link") {
       unstripped_exe =
           "{{root_out_dir}}/{{target_output_name}}{{output_extension}}"
-      command = "$cc_wrapper $cxx $ld_arg ${external_ldflags} {{inputs}} {{solibs}} {{libs}} -o $unstripped_exe"
+      command = "$cc_wrapper $cxx $ld_arg {{ldflags}} ${external_ldflags} {{inputs}} {{solibs}} {{libs}} -o $unstripped_exe"
       outputs = [ unstripped_exe ]
       description = "link $unstripped_exe"
       if (strip != "") {
@@ -174,7 +174,7 @@ gcc_like_toolchain("wasm") {
   cpu = host_cpu
   os = host_os
   ar = "$emsdk_dir/emscripten/emar --em-config $em_config"
-  cc = "$emsdk_dir/emscripten/emcc --em-config $em_config"
-  cxx = "$emsdk_dir/emscripten/em++ --em-config $em_config  --bind -s USE_PTHREADS=1"
+  cc = "$emsdk_dir/emscripten/emcc --em-config $em_config -pthread"
+  cxx = "$emsdk_dir/emscripten/em++ --em-config $em_config --bind -pthread"
   strip = ""
 }
diff --git a/talk/owt/sdk/wasm/gn/wasm.gni b/talk/owt/sdk/wasm/gn/wasm.gni
@@ -38,26 +38,26 @@ template("wasm_lib") {
   assert(invoker.name + "_wasm" == target_name)
   _lib_name = invoker.name
   if (is_wasm) {
-    _exports = "['ccall', 'callMain', 'addFunction', 'FS']"
+    _exports = "['writeArrayToMemory']"
     _target_ldflags = [
       "-s",
       "WASM=1",
-      "-s",
-      "DISABLE_EXCEPTION_CATCHING=1",
-      "-s",
-      "NO_DYNAMIC_EXECUTION=1",
+      # "-s",
+      # "DISABLE_EXCEPTION_CATCHING=1",
       "-s",
       "INITIAL_MEMORY=33554432",
       "-s",
       "ALLOW_TABLE_GROWTH=1",
-      "-s",
-      "WASM_ASYNC_COMPILATION=0",
+      # "-s",
+      # "WASM_ASYNC_COMPILATION=0",
       "-s",
       "USE_PTHREADS=1",
       "-s",
+      "PROXY_TO_PTHREAD=1",
+      "-s",
       "LLD_REPORT_UNDEFINED",
       "-s",
-      "EXTRA_EXPORTED_RUNTIME_METHODS=" + _exports,
+      "EXPORTED_RUNTIME_METHODS=" + _exports,
 
       # This forces the MEMFS filesystem library to always use typed arrays
       # instead of building strings/arrays when appending to a file. This allows
@@ -66,30 +66,37 @@ template("wasm_lib") {
       "MEMFS_APPEND_TO_TYPED_ARRAYS=1",
 
       # Reduces global namespace pollution.
-      "-s",
-      "MODULARIZE=1",
+      # "-s",
+      # "MODULARIZE=1",
 
-      "-fno-stack-protector",
+      #"-fno-stack-protector",
 
       # This is to prevent that two different wasm modules end up generating
       # JS that overrides the same global variable (var Module = ...)
-      "-s",
-      "EXPORT_NAME=${target_name}",
+      # "-s",
+      # "EXPORT_NAME=${target_name}",
 
       "-lworkerfs.js",  # For FS.filesystems.WORKERFS
 
       "--bind",
+
+      "-s",
+      "PTHREAD_POOL_SIZE=2",
+
+      "-pthread",
+      
+      "--source-map-base",
     ]
     if (is_debug) {
       _target_ldflags += [
-        "-s",
-        "ASSERTIONS=2",
-        "-s",
-        "SAFE_HEAP=1",
-        "-s",
-        "STACK_OVERFLOW_CHECK=1",
-        "-gsource-map",
-        "-O0",
+        # "-s",
+        # "ASSERTIONS=2",
+        # "-s",
+        # "SAFE_HEAP=1",
+        # "-s",
+        # "STACK_OVERFLOW_CHECK=1",
+        # "-gsource-map",
+        # "-O0",
       ]
     } else {
       _target_ldflags += [
@@ -108,7 +115,6 @@ template("wasm_lib") {
     }
 
     _target_cflags = [
-      "-fno-stack-protector",
       "-pthread",
     ]
     if (defined(invoker.cflags)) {
diff --git a/talk/owt/sdk/wasm/media_session.cc b/talk/owt/sdk/wasm/media_session.cc
@@ -3,30 +3,45 @@
 // SPDX-License-Identifier: Apache-2.0
 
 #include "talk/owt/sdk/wasm/media_session.h"
+#include "third_party/webrtc/api/rtc_event_log/rtc_event_log_factory.h"
+#include "third_party/webrtc/api/transport/field_trial_based_config.h"
+#include "third_party/webrtc/rtc_base/task_queue_stdlib.h"
 
 namespace owt {
 namespace wasm {
 MediaSession::MediaSession()
-    : call_(std::unique_ptr<webrtc::Call>(
-          webrtc::Call::Create(webrtc::Call::Config(nullptr)))),
-      video_receiver_(nullptr) {}
+    : task_queue_factory_(webrtc::CreateTaskQueueStdlibFactory()),
+      event_log_factory_(std::make_unique<webrtc::RtcEventLogFactory>(
+          task_queue_factory_.get())),
+      event_log_(event_log_factory_->CreateRtcEventLog(
+          webrtc::RtcEventLog::EncodingType::NewFormat)),
+      web_transport_session_(std::make_unique<WebTransportSession>()),
+      call_(nullptr),
+      receive_statistics_(
+          webrtc::ReceiveStatistics::Create(webrtc::Clock::GetRealTimeClock())),
+      receiver_process_thread_(nullptr),
+      video_receiver_(nullptr) {
+  RTC_LOG(LS_ERROR) << "Ctor of Media Session. " << __EMSCRIPTEN_PTHREADS__;
+  rtc::ThreadManager::Instance()->WrapCurrentThread();
+  receiver_process_thread_=webrtc::ProcessThread::Create("ReceiverProcessThread");
+  webrtc::Call::Config config(event_log_.get());
+  config.task_queue_factory = task_queue_factory_.get();
+  webrtc::FieldTrialBasedConfig trials;
+  config.trials = &trials;
+  call_ = std::unique_ptr<webrtc::Call>(webrtc::Call::Create(config));
+}
 
 MediaSession::~MediaSession() {}
 
-RtpVideoReceiver* MediaSession::CreateRtpVideoReceiver() {
+std::shared_ptr<RtpVideoReceiver> MediaSession::CreateRtpVideoReceiver() {
   webrtc::VideoReceiveStream::Config config(web_transport_session_.get());
-  webrtc::TaskQueueBase* task_queue = webrtc::TaskQueueBase::Current();
-  if (!task_queue) {
-    task_queue = rtc::ThreadManager::Instance()->CurrentThread();
-  }
-  std::unique_ptr<webrtc::RtpVideoStreamReceiver2> receiver =
-      std::make_unique<webrtc::RtpVideoStreamReceiver2>(
-          task_queue, webrtc::Clock::GetRealTimeClock(),
-          web_transport_session_.get(), nullptr, nullptr, &config, nullptr,
-          nullptr, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr,
-          nullptr);
-  video_receiver_ = std::make_unique<RtpVideoReceiver>(std::move(receiver));
-  return video_receiver_.get();
+  config.rtp.local_ssrc = 1;
+  // TODO: Get a message queue for receiver.
+  video_receiver_ = std::make_shared<RtpVideoReceiver>(
+      web_transport_session_.get(), &config, receive_statistics_.get(),
+      receiver_process_thread_.get(), web_transport_session_.get());
+  return video_receiver_;
 }
+
 }  // namespace wasm
 }  // namespace owt
diff --git a/talk/owt/sdk/wasm/media_session.h b/talk/owt/sdk/wasm/media_session.h
@@ -8,6 +8,7 @@
 #include "call/call.h"
 #include "talk/owt/sdk/wasm/rtp_video_receiver.h"
 #include "talk/owt/sdk/wasm/web_transport_session.h"
+#include "third_party/webrtc/api/rtc_event_log/rtc_event_log_factory_interface.h"
 
 namespace owt {
 namespace wasm {
@@ -16,12 +17,17 @@ class MediaSession {
  public:
   explicit MediaSession();
   virtual ~MediaSession();
-  RtpVideoReceiver* CreateRtpVideoReceiver();
+  std::shared_ptr<RtpVideoReceiver> CreateRtpVideoReceiver();
 
  private:
+  std::unique_ptr<webrtc::TaskQueueFactory> task_queue_factory_;
+  std::unique_ptr<webrtc::RtcEventLogFactoryInterface> event_log_factory_;
+  std::unique_ptr<webrtc::RtcEventLog> event_log_;
   std::unique_ptr<WebTransportSession> web_transport_session_;
   std::unique_ptr<webrtc::Call> call_;
-  std::unique_ptr<RtpVideoReceiver> video_receiver_;
+  std::unique_ptr<webrtc::ReceiveStatistics> receive_statistics_;
+  std::unique_ptr<webrtc::ProcessThread> receiver_process_thread_;
+  std::shared_ptr<RtpVideoReceiver> video_receiver_;
 };
 
 }  // namespace wasm
diff --git a/talk/owt/sdk/wasm/rtp_video_receiver.cc b/talk/owt/sdk/wasm/rtp_video_receiver.cc
@@ -8,16 +8,50 @@
 
 namespace owt {
 namespace wasm {
+
 RtpVideoReceiver::RtpVideoReceiver(
-    std::unique_ptr<webrtc::RtpVideoStreamReceiver2> receiver)
-    : receiver_(std::move(receiver)) {}
+    webrtc::Transport* transport,
+    const webrtc::VideoReceiveStream::Config* config,
+    webrtc::ReceiveStatistics* rtp_receive_statistics,
+    webrtc::ProcessThread* process_thread,
+    webrtc::NackSender* nack_sender)
+    : receiver_(nullptr), complete_frame_callback_(emscripten::val::null()) {
+  receiver_ = std::make_unique<webrtc::RtpVideoStreamReceiver2>(
+      nullptr, webrtc::Clock::GetRealTimeClock(), transport, nullptr, nullptr,
+      config, rtp_receive_statistics, nullptr, nullptr, process_thread,
+      nack_sender, nullptr, this, nullptr, nullptr);
+  webrtc::VideoCodec codec;
+  codec.codecType = webrtc::VideoCodecType::kVideoCodecH264;
+  // TODO: Assign payload value dynamically according to server side
+  // configuration.
+  receiver_->AddReceiveCodec(127, codec, std::map<std::string, std::string>(),
+                             false);
+  receiver_->StartReceive();
+}
 
-bool RtpVideoReceiver::OnRtpPacket(std::vector<uint8_t> packet) {
-  RTC_LOG(LS_INFO) << "On RTP packet, size " << packet.size();
+bool RtpVideoReceiver::OnRtpPacket(uintptr_t packet_ptr, size_t packet_size) {
+  const uint8_t* packet = reinterpret_cast<uint8_t*>(packet_ptr);
   // TODO: Create `rtp_packet_received` from `packet`.
   webrtc::RtpPacketReceived rtp_packet_received;
+  rtp_packet_received.Parse(packet, packet_size);
   receiver_->OnRtpPacket(rtp_packet_received);
   return false;
 }
+
+void RtpVideoReceiver::OnCompleteFrame(
+    std::unique_ptr<webrtc::video_coding::EncodedFrame> frame) {
+  if(complete_frame_callback_.isNull()){
+    RTC_LOG(LS_WARNING) << "No callback registered for complete frames.";
+    return;
+  }
+  emscripten::val buffer(emscripten::typed_memory_view(
+      frame->EncodedImage().size(), frame->EncodedImage().data()));
+  complete_frame_callback_(buffer);
+}
+
+void RtpVideoReceiver::SetCompleteFrameCallback(emscripten::val callback) {
+  complete_frame_callback_ = callback;
+}
+
 }  // namespace wasm
 }  // namespace owt
diff --git a/talk/owt/sdk/wasm/rtp_video_receiver.h b/talk/owt/sdk/wasm/rtp_video_receiver.h
@@ -5,19 +5,31 @@
 #ifndef OWT_WASM_RTPVIDEORECEIVER_H_
 #define OWT_WASM_RTPVIDEORECEIVER_H_
 
+#include <emscripten/bind.h>
 #include "third_party/webrtc/video/rtp_video_stream_receiver2.h"
 
 namespace owt {
 namespace wasm {
-class RtpVideoReceiver {
+class RtpVideoReceiver
+    : public webrtc::RtpVideoStreamReceiver2::OnCompleteFrameCallback {
  public:
-  explicit RtpVideoReceiver(
-      std::unique_ptr<webrtc::RtpVideoStreamReceiver2> receiver);
+  explicit RtpVideoReceiver(webrtc::Transport* transport,
+                            const webrtc::VideoReceiveStream::Config* config,
+                            webrtc::ReceiveStatistics* rtp_receive_statistics,
+                            webrtc::ProcessThread* process_thread,
+                            webrtc::NackSender* nack_sender);
   virtual ~RtpVideoReceiver() {}
-  virtual bool OnRtpPacket(std::vector<uint8_t> packet);
+  virtual bool OnRtpPacket(uintptr_t packet_ptr, size_t packet_size);
+
+  // Overrides webrtc::RtpVideoStreamReceiver2::OnCompleteFrameCallback.
+  void OnCompleteFrame(
+      std::unique_ptr<webrtc::video_coding::EncodedFrame> frame) override;
+
+  void SetCompleteFrameCallback(emscripten::val callback);
 
  private:
   std::unique_ptr<webrtc::RtpVideoStreamReceiver2> receiver_;
+  emscripten::val complete_frame_callback_;
 };
 
 }  // namespace wasm
diff --git a/talk/owt/sdk/wasm/web_transport_session.cc b/talk/owt/sdk/wasm/web_transport_session.cc
@@ -6,7 +6,7 @@
 
 namespace owt {
 namespace wasm {
-  
+
 bool WebTransportSession::SendRtp(const uint8_t* packet,
                                   size_t length,
                                   const webrtc::PacketOptions& options) {
@@ -16,5 +16,9 @@ bool WebTransportSession::SendRtcp(const uint8_t* packet, size_t length) {
   return false;
 }
 
+void WebTransportSession::SendNack(
+    const std::vector<uint16_t>& sequence_numbers,
+    bool buffering_allowed) {}
+
 }  // namespace wasm
 }  // namespace owt
diff --git a/talk/owt/sdk/wasm/web_transport_session.h b/talk/owt/sdk/wasm/web_transport_session.h
@@ -6,10 +6,13 @@
 #define OWT_WASM_WEBTRANSPORTSESSION_H_
 
 #include "third_party/webrtc/api/call/transport.h"
+#include "third_party/webrtc/modules/include/module_common_types.h"
 
 namespace owt {
 namespace wasm {
-class WebTransportSession : public webrtc::Transport {
+
+class WebTransportSession : public webrtc::Transport,
+                            public webrtc::NackSender {
  public:
   explicit WebTransportSession() {}
   virtual ~WebTransportSession() {}
@@ -19,7 +22,12 @@ class WebTransportSession : public webrtc::Transport {
                size_t length,
                const webrtc::PacketOptions& options) override;
   bool SendRtcp(const uint8_t* packet, size_t length) override;
+
+  // Overrides webrtc::NackSender.
+  void SendNack(const std::vector<uint16_t>& sequence_numbers,
+                bool buffering_allowed) override;
 };
+
 }  // namespace wasm
 }  // namespace owt
 

Original file line number	Diff line number	Diff line change
`@@ -340,6 +340,8 @@ if (is_wasm) {`
`340`	`340`	`"//third_party/webrtc/call",`
`341`	`341`	`"//third_party/webrtc/modules/rtp_rtcp:rtp_rtcp",`
`342`	`342`	`"//third_party/webrtc/rtc_base:rtc_json",`
	`343`	`+ "//third_party/webrtc/rtc_base:rtc_task_queue_stdlib",`
	`344`	`+ "//third_party/webrtc/api/rtc_event_log:rtc_event_log_factory",`
`343`	`345`	`]`
`344`	`346`	`sources = [`
`345`	`347`	`"sdk/wasm/binding.h",`