feat: push streaming to sfu through websocket signaling

Author: TzuHuanTai

src/args.h

@@ -21,6 +21,7 @@ struct Args {
     bool use_libcamera = false;
     bool use_mqtt = false;
     bool use_whep = false;
+    bool use_websocket = false;
     bool fixed_resolution = false;
     uint32_t format = V4L2_PIX_FMT_MJPEG;
     std::string v4l2_format = "mjpeg";
@@ -40,6 +41,16 @@ struct Args {
 
     // http signaling
     uint16_t http_port = 8080;
+
+    // websocket signaling
+    int ws_port = 8080;
+    std::string ws_host = "192.168.4.21";
+    std::string ws_token =
+        "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
+        "eyJleHAiOjE3NzQ3NDU2MzksImlzcyI6IkFQSVduUVRzNHRtVVp2QSIsIm5iZiI6MTc0MzIwOTYzOSwic3ViIjoiZj"
+        "JkNzRiOTUtMmYxNi00ODRiLTg3NjctYThjNWY3NzFlZWY2IiwidmlkZW8iOnsiY2FuUHVibGlzaCI6dHJ1ZSwiY2Fu"
+        "UHVibGlzaERhdGEiOnRydWUsImNhblN1YnNjcmliZSI6ZmFsc2UsInJvb20iOiJkZXZpY2UtMSIsInJvb21Kb2luIj"
+        "p0cnVlfX0.o7e-gjkqfMpeDjATwaWWLKUWPa8RlaoIOcuw3p8FxQk";
 };
 
 #endif // ARGS_H_

src/conductor.cpp

@@ -79,17 +79,15 @@ void Conductor::AddTracks(rtc::scoped_refptr<webrtc::PeerConnectionInterface> pe
         return;
     }
 
-    std::string stream_id = "test_stream_id";
-
     if (audio_track_) {
-        auto audio_res = peer_connection->AddTrack(audio_track_, {stream_id});
+        auto audio_res = peer_connection->AddTrack(audio_track_, {args.uid});
         if (!audio_res.ok()) {
             ERROR_PRINT("Failed to add audio track, %s", audio_res.error().message());
         }
     }
 
     if (video_track_) {
-        auto video_res = peer_connection->AddTrack(video_track_, {stream_id});
+        auto video_res = peer_connection->AddTrack(video_track_, {args.uid});
         if (!video_res.ok()) {
             ERROR_PRINT("Failed to add video track, %s", video_res.error().message());
         }
@@ -101,8 +99,7 @@ void Conductor::AddTracks(rtc::scoped_refptr<webrtc::PeerConnectionInterface> pe
     }
 }
 
-rtc::scoped_refptr<RtcPeer> Conductor::CreatePeerConnection(PeerConfig peer_config) {
-    webrtc::PeerConnectionInterface::RTCConfiguration config;
+rtc::scoped_refptr<RtcPeer> Conductor::CreatePeerConnection(PeerConfig config) {
     config.sdp_semantics = webrtc::SdpSemantics::kUnifiedPlan;
     webrtc::PeerConnectionInterface::IceServer server;
     server.uri = args.stun_url;
@@ -116,8 +113,8 @@ rtc::scoped_refptr<RtcPeer> Conductor::CreatePeerConnection(PeerConfig peer_conf
         config.servers.push_back(turn_server);
     }
 
-    peer_config.timeout = args.peer_timeout;
-    auto peer = RtcPeer::Create(std::move(peer_config));
+    config.timeout = args.peer_timeout;
+    auto peer = RtcPeer::Create(std::move(config));
     auto result = peer_connection_factory_->CreatePeerConnectionOrError(
         config, webrtc::PeerConnectionDependencies(peer.get()));
 
@@ -127,6 +124,11 @@ rtc::scoped_refptr<RtcPeer> Conductor::CreatePeerConnection(PeerConfig peer_conf
     }
 
     peer->SetPeer(result.MoveValue());
+
+    if (!config.is_publisher) {
+        return peer;
+    }
+
     peer->CreateDataChannel();
     peer->OnSnapshot([this](std::shared_ptr<DataChannelSubject> datachannel, std::string msg) {
         OnSnapshot(datachannel, msg);

src/main.cpp

@@ -8,6 +8,7 @@
 #include "recorder/recorder_manager.h"
 #include "signaling/http_service.h"
 #include "signaling/mqtt_service.h"
+#include "signaling/websocket_service.h"
 
 int main(int argc, char *argv[]) {
     Args args;
@@ -33,6 +34,10 @@ int main(int argc, char *argv[]) {
         services.push_back(HttpService::Create(args, conductor, ioc_));
     }
 
+    if (args.use_websocket) {
+        services.push_back(WebsocketService::Create(args, conductor, ioc_));
+    }
+
     if (args.use_mqtt) {
         services.push_back(MqttService::Create(args, conductor));
     }

src/parser.cpp

@@ -59,6 +59,11 @@ void Parser::ParseArgs(int argc, char *argv[], Args &args) {
         ("mqtt_password", bpo::value<std::string>()->default_value(args.mqtt_password),
             "Mqtt server password")
         ("http_port", bpo::value<uint16_t>()->default_value(args.http_port), "Http server port")
+        ("ws_host", bpo::value<std::string>()->default_value(args.ws_host),
+            "Websocket server host")
+        ("ws_port", bpo::value<int>()->default_value(args.ws_port), "Websocket server port")
+        ("ws_token", bpo::value<std::string>()->default_value(args.ws_token),
+            "Websocket server token")
         ("record_path", bpo::value<std::string>()->default_value(args.record_path),
             "The path to save the recording video files. The recorder won't start if it's empty")
         ("hw_accel", bpo::bool_switch()->default_value(args.hw_accel),
@@ -67,6 +72,8 @@ void Parser::ParseArgs(int argc, char *argv[], Args &args) {
             "Use mqtt to exchange sdp and ice candidates")
         ("use_whep", bpo::bool_switch()->default_value(args.use_whep),
             "Use whep to exchange sdp and ice candidates")
+        ("use_websocket", bpo::bool_switch()->default_value(args.use_websocket),
+            "Use websocket to exchange sdp and ice candidates")
         ("v4l2_format", bpo::value<std::string>()->default_value(args.v4l2_format),
             "Set v4l2 camera capture format to `i420`, `mjpeg`, `h264`. The `h264` can pass "
             "packets into mp4 without encoding to reduce cpu usage."
@@ -106,13 +113,17 @@ void Parser::ParseArgs(int argc, char *argv[], Args &args) {
     SetIfExists(vm, "mqtt_username", args.mqtt_username);
     SetIfExists(vm, "mqtt_password", args.mqtt_password);
     SetIfExists(vm, "http_port", args.http_port);
+    SetIfExists(vm, "ws_host", args.ws_host);
+    SetIfExists(vm, "ws_port", args.ws_port);
+    SetIfExists(vm, "ws_token", args.ws_token);
     SetIfExists(vm, "record_path", args.record_path);
 
     args.fixed_resolution = vm["fixed_resolution"].as<bool>();
     args.no_audio = vm["no_audio"].as<bool>();
     args.hw_accel = vm["hw_accel"].as<bool>();
     args.use_mqtt = vm["use_mqtt"].as<bool>();
     args.use_whep = vm["use_whep"].as<bool>();
+    args.use_websocket = vm["use_websocket"].as<bool>();
 
     if (!args.stun_url.empty() && args.stun_url.substr(0, 4) != "stun") {
         std::cout << "Stun url should not be empty and start with \"stun:\"" << std::endl;

src/rtc_peer.cpp

@@ -18,6 +18,14 @@ RtcPeer::~RtcPeer() {
     DEBUG_PRINT("peer connection (%s) was destroyed!", id_.c_str());
 }
 
+void RtcPeer::CreateOffer() {
+    if (signaling_state_ == webrtc::PeerConnectionInterface::SignalingState::kHaveLocalOffer) {
+        return;
+    }
+
+    peer_connection_->CreateOffer(this, webrtc::PeerConnectionInterface::RTCOfferAnswerOptions());
+}
+
 void RtcPeer::Terminate() {
     is_connected_.store(false);
     is_complete_.store(true);
@@ -105,6 +113,10 @@ void RtcPeer::OnCameraOption(OnCommand func) {
 }
 
 void RtcPeer::SubscribeCommandChannel(CommandType type, OnCommand func) {
+    if (!data_channel_subject_) {
+        ERROR_PRINT("Data channel is not created!");
+        return;
+    }
     auto observer = data_channel_subject_->AsObservable(type);
     observer->Subscribe([this, func](std::string message) {
         if (!message.empty()) {
@@ -114,6 +126,7 @@ void RtcPeer::SubscribeCommandChannel(CommandType type, OnCommand func) {
 }
 
 void RtcPeer::OnSignalingChange(webrtc::PeerConnectionInterface::SignalingState new_state) {
+    signaling_state_ = new_state;
     auto state = webrtc::PeerConnectionInterface::AsString(new_state);
     DEBUG_PRINT("OnSignalingChange => %s", std::string(state).c_str());
     if (new_state == webrtc::PeerConnectionInterface::SignalingState::kHaveRemoteOffer) {

src/rtc_peer.h

@@ -12,8 +12,9 @@
 #include "common/logging.h"
 #include "data_channel_subject.h"
 
-struct PeerConfig {
+struct PeerConfig : public webrtc::PeerConnectionInterface::RTCConfiguration {
     int timeout = 10;
+    bool is_publisher = true;
     bool has_candidates_in_sdp = false;
 };
 
@@ -82,6 +83,7 @@ class RtcPeer : public webrtc::PeerConnectionObserver,
 
     RtcPeer(PeerConfig config);
     ~RtcPeer();
+    void CreateOffer();
     void Terminate();
     bool IsConnected() const;
     std::string GetId() const;
@@ -129,6 +131,7 @@ class RtcPeer : public webrtc::PeerConnectionObserver,
 
     std::string modified_sdp_;
     webrtc::SdpParseError *modified_desc_error_;
+    webrtc::PeerConnectionInterface::SignalingState signaling_state_;
     std::unique_ptr<webrtc::SessionDescriptionInterface> modified_desc_;
 
     std::shared_ptr<DataChannelSubject> data_channel_subject_;

src/signaling/CMakeLists.txt

@@ -8,6 +8,7 @@ endif()
 add_library(${PROJECT_NAME}
     mqtt_service.cpp
     http_service.cpp
+    websocket_service.cpp
 )
 
 target_link_libraries(${PROJECT_NAME} PUBLIC ${MOSQUITTO_LIBS})

src/signaling/signaling_service.h

@@ -8,7 +8,7 @@
 class SignalingService {
   public:
     SignalingService(std::shared_ptr<Conductor> conductor, bool has_candidates_in_sdp = false)
-        : conductor_(conductor),
+        : conductor(conductor),
           has_candidates_in_sdp_(has_candidates_in_sdp) {}
 
     void Start() {
@@ -20,11 +20,10 @@ class SignalingService {
         Connect();
     }
 
-    rtc::scoped_refptr<RtcPeer> CreatePeer() {
-        PeerConfig config;
+    rtc::scoped_refptr<RtcPeer> CreatePeer(PeerConfig config = PeerConfig{}) {
         config.has_candidates_in_sdp = has_candidates_in_sdp_;
 
-        auto peer = conductor_->CreatePeerConnection(std::move(config));
+        auto peer = conductor->CreatePeerConnection(std::move(config));
         peer_map_[peer->GetId()] = peer;
         return peer;
     }
@@ -58,10 +57,11 @@ class SignalingService {
     virtual void Connect() = 0;
     virtual void Disconnect() = 0;
 
+    std::shared_ptr<Conductor> conductor;
+
   private:
     bool has_candidates_in_sdp_;
     std::unique_ptr<Worker> worker_;
-    std::shared_ptr<Conductor> conductor_;
     std::unordered_map<std::string, rtc::scoped_refptr<RtcPeer>> peer_map_;
 };