webrtc 信号槽实现分析

Posted 2021-03-16 woder

前言

从网上看到的另一句话，webrtc中的信号槽像是变种的观察者模式，当有信号发送的时候，信号发送者通过遍历自己的所有connect的信号槽，然后执行他们的回调；起初会有为什么需要信号槽 信号槽与普通的函数回调有什么区别的疑问, 个人认为信号槽是一种面向对象的回调，当回调涉及到对象，需要考虑到执行对象的生命周期，信号槽这种变种的观察者模式，在接收信号的对象析构导致信号槽析构的时候，会将注册的信号删除(就像是删除注册了的观察者)，很好的将信号和其对象的生命周期进行了绑定管理。

正文

一个例子

从网上拿了XXDK使用webrtc信号槽的例子，如下:

/*************************************************************************
    > File Name: main.cpp
    > Author: 
    > Email: [e-mail] 
    > Created Time: Fri 31 Aug 2018 09:38:20 AM CST
 ************************************************************************/

#include<iostream>
#include"sigslot.h"
#include<unistd.h>
#include<stdio.h>

using namespace std;
using namespace sigslot;

class Sender 
{
public:
    sigslot::signal2<std::string, int> m_pfnsigDanger;

    void Panic() {
        static int nVal = 0;
        char szVal[20] = {0};
        snprintf(szVal, 20, "help--%d", nVal);
        m_pfnsigDanger(szVal, nVal++); //发射信号
    }
};

class Receiver: public sigslot::has_slots<>
{
public: 
    void OnDanger(std::string strMsg, int nVal) {
        std::cout << "ccc. Receiver::OnDanger(std::string, int)" << std::endl;
        std::cout << strMsg.c_str() << "==> " << nVal << std::endl;
    }
};

int main()
{
    Sender sender;
    Receiver recever;

    sender.m_pfnsigDanger.connect(&recever, &Receiver::OnDanger);//将信号与信号槽相连

    while (1) {
        std::cout << std::endl;
        std::cout << "==================================> in while..." << std::endl;
        sender.Panic();//打算发射信号
        sleep(2);
    }

    return 0;
}

首先定义了一个sender和一个receiver，sender作为信号的发送者，类内包含了一个信号m_pfnsigDanger， recever作为信号的接收者，继承了has_slots这个类, 提供了信号槽用于给信号绑定。使用m_pfnsigDanger.connect(), 将信号连接到receiver的信号槽中去，并设置回调事件的触发回调；

在这个例子中，涉及到的类如下图所示:

对于Sender, 其内部拥有一个signal_with_thread_policy类型的信号对象，signal_with_thread_policy提供模板以生成不同入参类型回调函数的信号作用，并对发送者提供一些简单的使用接口，如发送信号和连接信号；signal_with_thread_policy继承于_signal_base，_signal_base是真正存储连接信号的类，并提供了更多对于信号的控制和处理；_signal_base继承自_signal_base_interface，_signal_base_interface提供了外部（如信号槽）管理使用信号的接口；
对于Reserver,其继承了has_slots，就拥有了一个信号槽，其作用是对连接上来的信号进行管理，当信号槽析构的时候，对应的信号也就不复存在；信号槽继承了has_slots_interface，用于外部对该信号槽的一些管理操作

信号连接过程和发射过程

一个基本的信号连接和发射的过程如下图所示

接下来仔细分析connect过程；

template <class mt_policy, typename... Args>
class signal_with_thread_policy : public _signal_base<mt_policy> {
 void connect(desttype* pclass, void (desttype::*pmemfun)(Args...)) {    
        lock_block<mt_policy> lock(this);
        this->m_connected_slots.push_back(_opaque_connection(pclass, pmemfun));
        pclass->signal_connect(static_cast<_signal_base_interface*>(this));//将该信号添加到信号槽中
    }
}

首先根据传入的对象指针pclass和信号触发的成员回调函数pmemfun生成opaque_connection，opaque_connection的作用就是保存对象的指针以及回调的成员函数的指针，然后提供接口去调用对象的成员函数，如下：

class _opaque_connection {
private:
    typedef void (*emit_t)(const _opaque_connection*);

    template <typename FromT, typename ToT>
    union union_caster {
        FromT from;
        ToT to;
    };

    emit_t pemit;                // 回调的成员函数指针
    has_slots_interface* pdest;  // 对象指针
    unsigned char pmethod[16];

public:
    template <typename DestT, typename... Args>
    _opaque_connection(DestT* pd, void (DestT::*pm)(Args...)) : pdest(pd) {//构造函数，保存对象指针和模板实例化后的成员函数指针
        typedef void (DestT::*pm_t)(Args...); 
        static_assert(sizeof(pm_t) <= sizeof(pmethod),
                      "Size of slot function pointer too large.");

        std::memcpy(pmethod, &pm, sizeof(pm_t)); 

        typedef void (*em_t)(const _opaque_connection* self, Args...);
        union_caster<em_t, emit_t> caster2; 
        caster2.from = &_opaque_connection::emitter<DestT, Args...>; //模板实例化Dest类型的emitter函数地址
        pemit = caster2.to;                                         //将该实例化的函数地址强转成emit_t类型先保存起来，需要使用的使用，在通过caster 按照相反的方式转回来
    }
    ...

    //发射函数
    template <typename... Args>
    void emit(Args... args) const {
        std::cout << "999. _opaque_connection::emit(Args... args)" << std::endl;
        typedef void (*em_t)(const _opaque_connection*, Args...);
        union_caster<emit_t, em_t> caster;
        caster.from = pemit;         //获取模板实例化的emitter（）函数地址，强转，调用
        (caster.to)(this, args...); //调用函数
    }
private:

    template <typename DestT, typename... Args>
    static void emitter(const _opaque_connection* self, Args... args) {
        typedef void (DestT::*pm_t)(Args...);
        pm_t pm;
        std::memcpy(&pm, self->pmethod, sizeof(pm_t)); ///< 成员函数指针拷贝到pm_t 
     
        (static_cast<DestT*>(self->pdest)->*(pm))(args...); 
    }
};

_opaque_connection(DestT* pd, void (DestT::*pm)(Args...))将传入的对象指针和函数指针进行保存，在该构造函数中有一个设计的细节，注意观察_opaque_connection并不是一个模板类，使用模板的是其构造函数，因为库的作者想通过构造函数的参数进行类型推导而不是显示实例化模板类型去使用这个类，当涉及到回调函数和对象参数类型的保存，需要保存的模板实例化的东西，下次调用的时候才能找到，为此使用了union_caster去将模板实例化的函数强转保存；
提供调用保存下来的成员函数回调的接口便是 void emit(Args... args)

在生成完opaque_connection之后, 会将其push_back到m_connected_slots， m_connected_slots是_signal_base中的一个list, 用于保存所有的opaque_connection，以便调用的时候找到；

接下来receiver调用signal_connect将信号放入信号槽中，以便于信号槽析构的时候能够将该信号删除

template <class mt_policy, typename... Args>
class signal_with_thread_policy : public _signal_base<mt_policy> {
 void connect(desttype* pclass, void (desttype::*pmemfun)(Args...)) {    
        lock_block<mt_policy> lock(this);
        this->m_connected_slots.push_back(_opaque_connection(pclass, pmemfun));
=》     pclass->signal_connect(static_cast<_signal_base_interface*>(this));//将该信号添加到信号槽中
    }
}

pclass->signal_connect()，调用的是has_slots_interface::signal_connect(),而signal_connect()调用是has_slots::do_signal_connect(),这一切都在has_slots构造的时候关联好了，这些操作把信号记录信号槽的一个集合中，以便于信号槽析构的时候析构会删除信号

class has_slots_interface {
    ...
    typedef void (*signal_connect_t)(has_slots_interface* self,
                                   _signal_base_interface* sender);
    const signal_connect_t m_signal_connect;       
protected:
    has_slots_interface(signal_connect_t conn,
                        signal_disconnect_t disc,
                        disconnect_all_t disc_all)
      : m_signal_connect(conn),                        //设置connect
        m_signal_disconnect(disc),
        m_disconnect_all(disc_all) {}

public:
    void signal_connect(_signal_base_interface* sender) {//调用connect
        m_signal_connect(this, sender);  
    }
    ....
};


template <class mt_policy = SIGSLOT_DEFAULT_MT_POLICY>
class has_slots : public has_slots_interface, public mt_policy {
private:
    typedef std::set<_signal_base_interface*> sender_set;         
    typedef sender_set::const_iterator const_iterator;

public:
    has_slots()
        : has_slots_interface(&has_slots::do_signal_connect,     //设定has_slots_interface::m_signal_connect为do_signal_connect()
                              &has_slots::do_signal_disconnect,
                              &has_slots::do_disconnect_all) {}

    ....
private:
    static void do_signal_connect(has_slots_interface* p,
                                  _signal_base_interface* sender) {
        has_slots* const self = static_cast<has_slots*>(p);
        lock_block<mt_policy> lock(self);
        self->m_senders.insert(sender);                          //将此signal插入slot的signal集合中
    }
};

接下来，再看看信号发射过程
在Sender中执行了panic()，信号通过重载的()去调用signal_with_thread_policy::emit()函数，这个函数去遍历所有记录的opaque_connection, 执行opaque_connection中保存的回调

    void signal_with_thread_policy:: emit(Args... args) {
        lock_block<mt_policy> lock(this);
        this->m_current_iterator = this->m_connected_slots.begin();
     
        while (this->m_current_iterator != this->m_connected_slots.end()) { //遍历所有记录的 opaque_connection
            _opaque_connection const& conn = *this->m_current_iterator;   
            conn.emit<Args...>(args...);                                    //执行其中保存的回调 
        }
    }

其中opaque_connection::emit()使用到了上面提到的caster(但其实这种联合体的用法貌似有问题的by way of accessing an inactive member of the union lead to undefied behavior)，将保存的实例化的模板函数指针进行强转回来进行调用

template <typename... Args>
void emit(Args... args) const {
    typedef void (*em_t)(const _opaque_connection*, Args...);
       
    union_caster<emit_t, em_t> caster;
    caster.from = pemit;     
    (caster.to)(this, args...); //强转函数指针，调用回调
}