PeerConnectionManager.h

Go to the documentation of this file.

// ----------------------------------------------------------------------------
// -                        Open3D: www.open3d.org                            -
// ----------------------------------------------------------------------------
// Copyright (c) 2018-2024 www.open3d.org
// SPDX-License-Identifier: MIT
// ----------------------------------------------------------------------------
// ----------------------------------------------------------------------------
// Contains source code from
// https://github.com/mpromonet/webrtc-streamer
//
// This software is in the public domain, furnished "as is", without technical
// support, and with no warranty, express or implied, as to its usefulness for
// any purpose.
// ----------------------------------------------------------------------------
//
// This is a private header. It shall be hidden from Open3D's public API. Do not
// put this in Open3D.h.in.
 
#pragma once
 
#include <api/field_trials.h>
#include <api/peer_connection_interface.h>
#include <api/scoped_refptr.h>
#include <api/stats/rtc_stats.h>
#include <rtc_base/ref_counted_object.h>
#include <rtc_base/strings/json.h>
 
#include <atomic>
#include <condition_variable>
#include <future>
#include <mutex>
#include <regex>
#include <string>
#include <thread>
#include <unordered_map>
 
#include "open3d/visualization/webrtc_server/BitmapTrackSource.h"
#include "open3d/visualization/webrtc_server/HttpServerRequestHandler.h"
#include "open3d/visualization/webrtc_server/WebRTCWindowSystem.h"
 
namespace webrtc {
class Thread;
}
 
namespace open3d {
namespace visualization {
namespace webrtc_server {
 
class PeerConnectionManager {
    class VideoSink : public webrtc::VideoSinkInterface<webrtc::VideoFrame> {
    public:
        VideoSink(webrtc::VideoTrackInterface* track) : track_(track) {
            track_->AddOrUpdateSink(this, webrtc::VideoSinkWants());
        }
        virtual ~VideoSink() { track_->RemoveSink(this); }
 
        // VideoSinkInterface implementation
        virtual void OnFrame(const webrtc::VideoFrame& video_frame) {
            webrtc::scoped_refptr<webrtc::I420BufferInterface> buffer(
                    video_frame.video_frame_buffer()->ToI420());
            utility::LogDebug("[{}] frame: {}x{}", OPEN3D_FUNCTION,
                              buffer->height(), buffer->width());
        }
 
    protected:
        webrtc::scoped_refptr<webrtc::VideoTrackInterface> track_;
    };
 
    class SetSessionDescriptionObserver
        : public webrtc::SetSessionDescriptionObserver {
    public:
        static SetSessionDescriptionObserver* Create(
                webrtc::PeerConnectionInterface* pc,
                std::promise<const webrtc::SessionDescriptionInterface*>&
                        promise) {
            return new webrtc::RefCountedObject<SetSessionDescriptionObserver>(
                    pc, promise);
        }
        virtual void OnSuccess() {
            std::string sdp;
            if (pc_->local_description()) {
                promise_.set_value(pc_->local_description());
                pc_->local_description()->ToString(&sdp);
            } else if (pc_->remote_description()) {
                promise_.set_value(pc_->remote_description());
                pc_->remote_description()->ToString(&sdp);
            }
        }
        virtual void OnFailure(webrtc::RTCError error) {
            utility::LogWarning("{}", error.message());
            promise_.set_value(nullptr);
        }
 
    protected:
        SetSessionDescriptionObserver(
                webrtc::PeerConnectionInterface* pc,
                std::promise<const webrtc::SessionDescriptionInterface*>&
                        promise)
            : pc_(pc), promise_(promise) {};
 
    private:
        webrtc::PeerConnectionInterface* pc_;
        std::promise<const webrtc::SessionDescriptionInterface*>& promise_;
    };
 
    class CreateSessionDescriptionObserver
        : public webrtc::CreateSessionDescriptionObserver {
    public:
        static CreateSessionDescriptionObserver* Create(
                webrtc::PeerConnectionInterface* pc,
                std::promise<const webrtc::SessionDescriptionInterface*>&
                        promise) {
            return new webrtc::RefCountedObject<
                    CreateSessionDescriptionObserver>(pc, promise);
        }
        virtual void OnSuccess(webrtc::SessionDescriptionInterface* desc) {
            std::string sdp;
            desc->ToString(&sdp);
            pc_->SetLocalDescription(
                    SetSessionDescriptionObserver::Create(pc_, promise_), desc);
        }
        virtual void OnFailure(webrtc::RTCError error) {
            utility::LogWarning("{}", error.message());
            promise_.set_value(nullptr);
        }
 
    protected:
        CreateSessionDescriptionObserver(
                webrtc::PeerConnectionInterface* pc,
                std::promise<const webrtc::SessionDescriptionInterface*>&
                        promise)
            : pc_(pc), promise_(promise) {};
 
    private:
        webrtc::PeerConnectionInterface* pc_;
        std::promise<const webrtc::SessionDescriptionInterface*>& promise_;
    };
 
    class PeerConnectionStatsCollectorCallback
        : public webrtc::RTCStatsCollectorCallback {
    public:
        PeerConnectionStatsCollectorCallback() {}
        void clearReport() { report_.clear(); }
        Json::Value getReport() { return report_; }
 
    protected:
        virtual void OnStatsDelivered(
                const webrtc::scoped_refptr<const webrtc::RTCStatsReport>&
                        report) {
            for (const webrtc::RTCStats& stats : *report) {
                Json::Value stats_members;
                for (const webrtc::Attribute& attribute : stats.Attributes()) {
                    stats_members[attribute.name()] = attribute.ToString();
                }
                report_[stats.id()] = stats_members;
            }
        }
 
        Json::Value report_;
    };
 
    class DataChannelObserver : public webrtc::DataChannelObserver {
    public:
        DataChannelObserver(PeerConnectionManager* peer_connection_manager,
                            webrtc::scoped_refptr<webrtc::DataChannelInterface>
                                    data_channel,
                            const std::string& peerid)
            : peer_connection_manager_(peer_connection_manager),
              data_channel_(data_channel),
              peerid_(peerid) {
            data_channel_->RegisterObserver(this);
        }
        virtual ~DataChannelObserver() { data_channel_->UnregisterObserver(); }
 
        // DataChannelObserver interface
        virtual void OnStateChange() {
            // Useful to know when the data channel is established.
            const std::string label = data_channel_->label();
            const std::string state =
                    webrtc::DataChannelInterface::DataStateString(
                            data_channel_->state());
            utility::LogDebug(
                    "DataChannelObserver::OnStateChange label: {}, state: {}, "
                    "peerid: {}",
                    label, state, peerid_);
            std::string msg(label + " " + state);
            webrtc::DataBuffer buffer(msg);
            data_channel_->Send(buffer);
            // ClientDataChannel is established after ServerDataChannel. Once
            // ClientDataChannel is established, we need to send initial frames
            // to the client such that the video is not empty. Afterwards,
            // video frames will only be sent when the GUI redraws.
            if (label == "ClientDataChannel" && state == "open") {
                {
                    std::lock_guard<std::mutex> mutex_lock(
                            peer_connection_manager_
                                    ->peerid_data_channel_mutex_);
                    peer_connection_manager_->peerid_data_channel_ready_.insert(
                            peerid_);
                }
                peer_connection_manager_->SendInitFramesToPeer(peerid_);
            }
            if (label == "ClientDataChannel" &&
                (state == "closed" || state == "closing")) {
                std::lock_guard<std::mutex> mutex_lock(
                        peer_connection_manager_->peerid_data_channel_mutex_);
                peer_connection_manager_->peerid_data_channel_ready_.erase(
                        peerid_);
            }
        }
        virtual void OnMessage(const webrtc::DataBuffer& buffer) {
            std::string msg((const char*)buffer.data.data(),
                            buffer.data.size());
            utility::LogDebug("DataChannelObserver::OnMessage: {}, msg: {}.",
                              data_channel_->label(), msg);
            std::string reply =
                    WebRTCWindowSystem::GetInstance()->OnDataChannelMessage(
                            msg);
            if (!reply.empty()) {
                webrtc::DataBuffer buffer(reply);
                data_channel_->Send(buffer);
            }
        }
 
    protected:
        PeerConnectionManager* peer_connection_manager_;
        webrtc::scoped_refptr<webrtc::DataChannelInterface> data_channel_;
        const std::string peerid_;
    };
 
    class PeerConnectionObserver : public webrtc::PeerConnectionObserver {
    public:
        PeerConnectionObserver(PeerConnectionManager* peer_connection_manager,
                               const std::string& peerid);
 
        void Initialize(webrtc::scoped_refptr<webrtc::PeerConnectionInterface>
                                peer_connection);
 
        virtual ~PeerConnectionObserver() {
            delete local_channel_;
            delete remote_channel_;
            if (pc_.get()) {
                // warning: pc->close call OnIceConnectionChange
                deleting_ = true;
                pc_->Close();
            }
        }
 
        Json::Value GetIceCandidateList() { return ice_candidate_list_; }
 
        Json::Value GetStats() {
            stats_callback_->clearReport();
            pc_->GetStats(stats_callback_.get());
            int count = 10;
            while ((stats_callback_->getReport().empty()) && (--count > 0)) {
                std::this_thread::sleep_for(std::chrono::milliseconds(1000));
            }
            return Json::Value(stats_callback_->getReport());
        };
 
        webrtc::scoped_refptr<webrtc::PeerConnectionInterface>
        GetPeerConnection() {
            return pc_;
        };
 
        // PeerConnectionObserver interface
        virtual void OnAddStream(
                webrtc::scoped_refptr<webrtc::MediaStreamInterface> stream) {
            utility::LogDebug("[{}] GetVideoTracks().size(): {}.",
                              OPEN3D_FUNCTION, stream->GetVideoTracks().size());
            webrtc::VideoTrackVector videoTracks = stream->GetVideoTracks();
            if (videoTracks.size() > 0) {
                video_sink_.reset(new VideoSink(videoTracks.at(0).get()));
            }
        }
        virtual void OnRemoveStream(
                webrtc::scoped_refptr<webrtc::MediaStreamInterface> stream) {
            video_sink_.reset();
        }
        virtual void OnDataChannel(
                webrtc::scoped_refptr<webrtc::DataChannelInterface> channel) {
            utility::LogDebug(
                    "PeerConnectionObserver::OnDataChannel peerid: {}",
                    peerid_);
            remote_channel_ = new DataChannelObserver(peer_connection_manager_,
                                                      channel, peerid_);
        }
        virtual void OnRenegotiationNeeded() {
            std::lock_guard<std::mutex> mutex_lock(
                    peer_connection_manager_->peerid_data_channel_mutex_);
            peer_connection_manager_->peerid_data_channel_ready_.erase(peerid_);
            utility::LogDebug(
                    "PeerConnectionObserver::OnRenegotiationNeeded peerid: {}",
                    peerid_);
        }
        virtual void OnIceCandidate(
                const webrtc::IceCandidateInterface* candidate);
 
        virtual void OnSignalingChange(
                webrtc::PeerConnectionInterface::SignalingState state) {
            utility::LogDebug("state: {}, peerid: {}", state, peerid_);
        }
        virtual void OnIceConnectionChange(
                webrtc::PeerConnectionInterface::IceConnectionState state) {
            if ((state ==
                 webrtc::PeerConnectionInterface::kIceConnectionFailed) ||
                (state ==
                 webrtc::PeerConnectionInterface::kIceConnectionClosed)) {
                ice_candidate_list_.clear();
                if (!deleting_) {
                    std::thread([this]() {
                        peer_connection_manager_->HangUp(peerid_);
                    }).detach();
                }
            }
        }
 
        virtual void OnIceGatheringChange(
                webrtc::PeerConnectionInterface::IceGatheringState) {}
 
    private:
        PeerConnectionManager* peer_connection_manager_;
        const std::string peerid_;
        webrtc::scoped_refptr<webrtc::PeerConnectionInterface> pc_;
        DataChannelObserver* local_channel_;
        DataChannelObserver* remote_channel_;
        Json::Value ice_candidate_list_;
        webrtc::scoped_refptr<PeerConnectionStatsCollectorCallback>
                stats_callback_;
        std::unique_ptr<VideoSink> video_sink_;
        bool deleting_;
    };
 
public:
    PeerConnectionManager(const std::list<std::string>& ice_server_list,
                          const Json::Value& config,
                          const std::string& publish_filter,
                          const std::string& webrtc_udp_port_range);
    virtual ~PeerConnectionManager();
 
    bool InitializePeerConnection();
    const std::map<std::string, HttpServerRequestHandler::HttpFunction>
    GetHttpApi();
 
    const Json::Value GetIceCandidateList(const std::string& peerid);
    const Json::Value AddIceCandidate(const std::string& peerid,
                                      const Json::Value& json_message);
    const Json::Value GetMediaList();
    const Json::Value HangUp(const std::string& peerid);
    const Json::Value Call(const std::string& peerid,
                           const std::string& window_uid,
                           const std::string& options,
                           const Json::Value& json_message);
    const Json::Value GetIceServers();
 
    void SendInitFramesToPeer(const std::string& peerid);
 
    void CloseWindowConnections(const std::string& window_uid);
 
    void OnFrame(const std::string& window_uid,
                 const std::shared_ptr<core::Tensor>& im);
 
protected:
    webrtc::scoped_refptr<BitmapTrackSourceInterface> GetVideoTrackSource(
            const std::string& window_uid);
    PeerConnectionObserver* CreatePeerConnection(const std::string& peerid);
    bool AddStreams(webrtc::PeerConnectionInterface* peer_connection,
                    const std::string& window_uid,
                    const std::string& options);
    webrtc::scoped_refptr<BitmapTrackSourceInterface> CreateVideoSource(
            const std::string& window_uid,
            const std::map<std::string, std::string>& opts);
    bool WindowStillUsed(const std::string& window_uid);
    webrtc::scoped_refptr<webrtc::PeerConnectionInterface> GetPeerConnection(
            const std::string& peerid);
 
protected:
    std::unique_ptr<webrtc::FieldTrials> field_trials_;
    webrtc::scoped_refptr<webrtc::PeerConnectionFactoryInterface>
            peer_connection_factory_;
 
    // Each peer has exactly one connection.
    std::unordered_map<std::string, PeerConnectionObserver*>
            peerid_to_connection_;
    std::mutex peerid_to_connection_mutex_;
    // Set of peerids with data channel ready for communication
    std::unordered_set<std::string> peerid_data_channel_ready_;
    std::mutex peerid_data_channel_mutex_;
 
    // Each Window has exactly one TrackSource.
    std::unordered_map<std::string,
                       webrtc::scoped_refptr<BitmapTrackSourceInterface>>
            window_uid_to_track_source_;
    std::mutex window_uid_to_track_source_mutex_;
 
    // Each Window can be connected to zero, one or more peers.
    std::unordered_map<std::string, std::set<std::string>>
            window_uid_to_peerids_;
    std::unordered_map<std::string, std::string> peerid_to_window_uid_;
    // Shared by window_uid_to_peerids_ and peerid_to_window_uid_.
    std::mutex window_uid_to_peerids_mutex_;
 
    std::list<std::string> ice_server_list_;
    const Json::Value config_;
    const std::regex publish_filter_;
    std::map<std::string, HttpServerRequestHandler::HttpFunction> func_;
    std::string webrtc_port_range_;
 
    // Async encoder thread: decouples the render thread from the blocking
    // libyuv + WebRTC encode path. OnFrame() posts the latest frame per window;
    // the thread drains the map and posts video_track_source->OnFrame() to the
    // WebRTC worker thread.
    std::unordered_map<std::string, std::shared_ptr<core::Tensor>>
            pending_frames_;
    std::mutex pending_frames_mutex_;
    std::condition_variable pending_frames_cv_;
    std::atomic<bool> encoder_running_{false};
    std::thread encoder_thread_;
    // WebRTC worker thread (PeerConnectionFactoryDependencies::worker_thread).
    webrtc::Thread* webrtc_worker_thread_ = nullptr;
 
    void EncoderThreadLoop();
};
 
}  // namespace webrtc_server
}  // namespace visualization
}  // namespace open3d
 
namespace fmt {
 
template <>
struct formatter<webrtc::PeerConnectionInterface::SignalingState> {
    template <typename FormatContext>
    auto format(const webrtc::PeerConnectionInterface::SignalingState& state,
                FormatContext& ctx) const -> decltype(ctx.out()) {
        using namespace webrtc;
        const char* text = nullptr;
        switch (state) {
            case PeerConnectionInterface::SignalingState::kStable:
                text = "kStable";
                break;
            case PeerConnectionInterface::SignalingState::kHaveLocalOffer:
                text = "kHaveLocalOffer";
                break;
            case PeerConnectionInterface::SignalingState::kHaveLocalPrAnswer:
                text = "kHaveLocalPrAnswer";
                break;
            case PeerConnectionInterface::SignalingState::kHaveRemoteOffer:
                text = "kHaveRemoteOffer";
                break;
            case PeerConnectionInterface::SignalingState::kHaveRemotePrAnswer:
                text = "kHaveRemotePrAnswer";
                break;
            case PeerConnectionInterface::SignalingState::kClosed:
                text = "kClosed";
                break;
            default:
                text = "unknown";
        }
        return fmt::format_to(ctx.out(), "{}", text);
    }
 
    template <typename ParseContext>
    constexpr auto parse(ParseContext& ctx) -> decltype(ctx.begin()) {
        return ctx.begin();
    }
};
 
}  // namespace fmt