一分钟快速搭建 rtmpd 服务器: https://blog.csdn.net/freeabc/article/details/102880984

软件下载地址: http://www.qiyicc.com/download/rtmpd.rar

github 地址：https://github.com/superconvert/smart_rtmpd

-----------------------------------------------------------------------------------------------------------------------------------------

webrtc 中有关 socket 运行机制以及 stun 收发过程 及 Candidates 生成流程分析

我写文章一般是两个思路：
1. 下一步要调用什么对象的方法
2.  这一步的对象，怎么关联到下一步的对象的流程分析
这一步的流程主要阐述怎么关联下一步的对象的流程分析，当然这一步做了什么具体的工作，不能
详细展示，否则，太庞大了，需要各位朋友针对重点的部分，自己揣摩了。

//*******************************************************************************************
//
// webrtc 内部很多创建 socket 的地方，这个需要调用类厂 BasicPacketSocketFactory , 下面
// 这一小段就是分析 BasicPacketSocketFactory 的创建，以及内部管理的 socket 的部分流程
//
//*******************************************************************************************

AsyncPacketSocket* BasicPacketSocketFactory::CreateUdpSocket(const SocketAddress& address,    uint16_t min_port,    uint16_t max_port) {   // 参见下面的 SocketDispatcher   AsyncSocket* socket =   socket_factory()->CreateAsyncSocket(address.family(), SOCK_DGRAM);    if (!socket) {        return NULL;    }    // 这个 BindSocket 最终会调用系统的 bind    if (BindSocket(socket, address, min_port, max_port) < 0) {        RTC_LOG(LS_ERROR) << "UDP bind failed with error " << socket->GetError();        delete socket;        return NULL;    }    // 这个里面绑定了读和写事件到 AsyncUDPSocket::OnReadEvent , AsyncUDPSocket::OnWriteEvent    return new AsyncUDPSocket(socket);}

    1. 创建 BasicPacketSocketFactory
    ./pc/peer_connection_factory.cc
    BasicPacketSocketFactory 是 PeerConnectionFactory::Initialize() 中创建的

default_socket_factory_.reset(new rtc::BasicPacketSocketFactory(network_thread_));

    2.
    ./sdk/android/src/jni/pc/peer_connection_factory.cc
    而 network_thread_ 则是 接口 CreatePeerConnectionFactoryForJava 里的
    std::unique_ptr<rtc::Thread> network_thread = rtc::Thread::CreateWithSocketServer();    
    其实就是这个

std::unique_ptr<Thread> Thread::CreateWithSocketServer() {return std::unique_ptr<Thread>(new Thread(SocketServer::CreateDefault()));}

    其实就是创建了 PhysicalSocketServer

std::unique_ptr<SocketServer> SocketServer::CreateDefault() {#if defined(__native_client__)        return std::unique_ptr<SocketServer>(new rtc::NullSocketServer);    #else        return std::unique_ptr<SocketServer>(new rtc::PhysicalSocketServer);    #endif}

   Thread 继承于 class Thread : public MessageQueue, public webrtc::TaskQueueBase ，我们看出 Thread 拥有消息队列的对象
    构造函数 Thread(SocketServer* ss)把 ss 赋值给基类 MessageQueue，基类接口通过 socketserver 返回这个对象，后续只要
    调用接口 socketserver 的，就是返回这个 socket 对象。

SocketServer* MessageQueue::socketserver() {return ss_;}

    上面的 BasicPacketSocketFactory::CreateUdpSocket 里的，这句话 socket_factory()->CreateAsyncSocket 其实就是调用 
    ./rtc_base/physical_socket_server.cc

AsyncSocket* PhysicalSocketServer::CreateAsyncSocket(int family, int type) {      SocketDispatcher* dispatcher = new SocketDispatcher(this);    // 这个里面通过 PhysicalSocket::Create 创建一个套接字    if (dispatcher->Create(family, type)) {return dispatcher;    } else {        delete dispatcher;        return nullptr;    }}

//******************************************************************************
//
// 下面这段是讲述 socket 怎么接收数据的，和上述流程没任何关系
//
//******************************************************************************

    上述流程中，有一个这个函数调用，

std::unique_ptr<Thread> Thread::CreateWithSocketServer() {return std::unique_ptr<Thread>(new Thread(SocketServer::CreateDefault()));}

    创建一个带线程的 socket    这个线程的 Run 如下：

void Thread::Run() {ProcessMessages(kForever);}

    // 这个里面不断的 Get 最新的 message 进行处理

bool Thread::ProcessMessages(int cmsLoop) {// Using ProcessMessages with a custom clock for testing and a time greater    // than 0 doesn't work, since it's not guaranteed to advance the custom    // clock's time, and may get stuck in an infinite loop.    RTC_DCHECK(GetClockForTesting() == nullptr || cmsLoop == 0 ||        cmsLoop == kForever);    int64_t msEnd = (kForever == cmsLoop) ? 0 : TimeAfter(cmsLoop);    int cmsNext = cmsLoop;    while (true) {  #if defined(WEBRTC_MAC)      ScopedAutoReleasePool pool;  #endif      Message msg;      if (!Get(&msg, cmsNext))          return !IsQuitting();      Dispatch(&msg);      if (cmsLoop != kForever) {          cmsNext = static_cast<int>(TimeUntil(msEnd));          if (cmsNext < 0)            return true;      }    }}

   // 其实就是基类的 MessageQueue 的接口

bool MessageQueue::Get(Message* pmsg, int cmsWait, bool process_io)// 看到这个 ss_ 了吗，就是 SocketServer::CreateDefault() 也就是 PhysicalSocketServer::Wait 接口    if (!ss_->Wait(static_cast<int>(cmsNext), process_io))

    这个地方监听所有的 socket 操作，三个版本的都有 win, linux，随便找一个分析
    ./rtc_base/physical_socket_server.cc

bool PhysicalSocketServer::Wait(int cmsWait, bool process_io)return WaitEpoll(cmsWait, signal_wakeup_);bool PhysicalSocketServer::WaitEpoll(int cmsWait)ProcessEvents(pdispatcher, readable, writable, check_error);static void ProcessEvents(Dispatcher* dispatcher, bool readable, bool writable, bool check_error) // 这里就是 SocketDispatcher -    dispatcher->OnEvent(ff, errcode);void SocketDispatcher::OnEvent(uint32_t ff, int err)    // 如果是读，这里假设是 UDP     SignalReadEvent(this);

    ./rtc_base/async_udp_socket.cc

void AsyncUDPSocket::OnReadEvent(AsyncSocket* socket)SignalReadPacket(this, buf_, static_cast<size_t>(len), remote_addr,        (timestamp > -1 ? timestamp : TimeMicros()));

    ./p2p/base/stun_port.cc 

void UDPPort::OnReadPacket(rtc::AsyncPacketSocket* socket,const char* data,    size_t size,    const rtc::SocketAddress& remote_addr,    const int64_t& packet_time_us) {        RTC_DCHECK(socket == socket_);    RTC_DCHECK(!remote_addr.IsUnresolvedIP());    // Look for a response from the STUN server.    // Even if the response doesn't match one of our outstanding requests, we    // will eat it because it might be a response to a retransmitted packet, and    // we already cleared the request when we got the first response.            if (server_addresses_.find(remote_addr) != server_addresses_.end()) {        // 这是 stun 阶段接收包        requests_.CheckResponse(data, size);        return;    }    // 这是建立链接后接收包，参考 webrtc 的视频数据接收过程    if (Connection* conn = GetConnection(remote_addr)) {        conn->OnReadPacket(data, size, packet_time_us);    } else {        Port::OnReadPacket(data, size, remote_addr, PROTO_UDP);    }}

//******************************************************************************
//
// 下面就分析了有关 webrtc stun 流程的部分
//
//******************************************************************************
    
1. 从这里开始分析，这个的调用参考 createPeerConnection 流程
JsepTransportController::MaybeStartGathering

2. 这个 ice_transport 就是 P2PTransportChannel 对象
dtls->ice_transport()->MaybeStartGathering();

3. 第一次创建流程
./p2p/base/p2p_transport_channel.cc

P2PTransportChannel::MaybeStartGathering     //------------------------------------------------------------    // 这个就是创建一个 PortAllocatorSession 并把信号挂接 P2PTransportChannel    //------------------------------------------------------------    AddAllocatorSession(allocator_->CreateSession(          transport_name(), component(), ice_parameters_.ufrag,          ice_parameters_.pwd));    // 进行 PortAllocatorSession 接口的调用    allocator_sessions_.back()->StartGettingPorts();

 3.1
    这个 allocator_ 来自下面的函数调用，我们看出 就是 JsepTransportController 的成员 port_allocator_
    ./pc/jsep_transport_controller.cc

rtc::scoped_refptr<webrtc::IceTransportInterface>JsepTransportController::CreateIceTransport(const std::string& transport_name, bool rtcp) {int component = rtcp ? cricket::ICE_CANDIDATE_COMPONENT_RTCP :cricket::ICE_CANDIDATE_COMPONENT_RTP;        IceTransportInit init;    init.set_port_allocator(port_allocator_);    init.set_async_resolver_factory(async_resolver_factory_);    init.set_event_log(config_.event_log);    return config_.ice_transport_factory->CreateIceTransport(transport_name, component, std::move(init));}

./api/ice_transport_factory.cc

rtc::scoped_refptr<IceTransportInterface> CreateIceTransport(IceTransportInit init) {return new rtc::RefCountedObject<IceTransportWithTransportChannel>(        std::make_unique<cricket::P2PTransportChannel>(        "", 0, init.port_allocator(), init.async_resolver_factory(), init.event_log()));}

    3.2
    我们跟踪一下 port_allocator_ 是在 JsepTransportController 初始化过程中传递过来的，我们分析 JsepTransportController 
    初始化，发现其实就是来自 PeerConnection 的 port_allocator_ 对象
    
    ./pc/peer_connection.cc

bool PeerConnection::Initialize(const PeerConnectionInterface::RTCConfiguration& configuration,        PeerConnectionDependencies dependencies)            // 传递过来的。。。。。。    port_allocator_ = std::move(dependencies.allocator);    ... ...    // 赋值给 JsepTransportController    transport_controller_.reset(new JsepTransportController(        signaling_thread(), network_thread(), port_allocator_.get(),        async_resolver_factory_.get(), config));

    3.3
    我们分析 PeerConnection 的初始化过程中, port_allocator_ 的产生过程
    ./pc/peer_connection_factory.cc

rtc::scoped_refptr<PeerConnectionInterface>PeerConnectionFactory::CreatePeerConnection(const PeerConnectionInterface::RTCConfiguration& configuration,    PeerConnectionDependencies dependencies)            if (!dependencies.allocator) {        rtc::PacketSocketFactory* packet_socket_factory;        if (dependencies.packet_socket_factory)            packet_socket_factory = dependencies.packet_socket_factory.get();        else            // 这个就是 BasicPacketSocketFactory 参见上面的分析            packet_socket_factory = default_socket_factory_.get();        network_thread_->Invoke<void>(RTC_FROM_HERE, [this, &configuration,            &dependencies,            &packet_socket_factory]() {            //------------------------------------------------------            // 这个就是我们要追踪的 port_allocator_ !!!!!!!!!!!!            //------------------------------------------------------            dependencies.allocator = std::make_unique<cricket::BasicPortAllocator>(                default_network_manager_.get(), packet_socket_factory, configuration.turn_customizer);        });    }            rtc::scoped_refptr<PeerConnection> pc(new rtc::RefCountedObject<PeerConnection>(this, std::move(event_log),        std::move(call)));    ActionsBeforeInitializeForTesting(pc);    if (!pc->Initialize(configuration, std::move(dependencies))) {        return nullptr;    }

    上述函数被下面这个调用，我们发现这个里面 dependencies.allocator 为空，因此 port_allocator_ 是在上面的步骤中分配的
    ./sdk/android/src/jni/pc/peer_connection_factory.cc

static jlong JNI_PeerConnectionFactory_CreatePeerConnection(JNIEnv* jni,    jlong factory,    const JavaParamRef<jobject>& j_rtc_config,    const JavaParamRef<jobject>& j_constraints,    jlong observer_p,    const JavaParamRef<jobject>& j_sslCertificateVerifier)            PeerConnectionDependencies peer_connection_dependencies(observer.get());    if (!j_sslCertificateVerifier.is_null()) {        peer_connection_dependencies.tls_cert_verifier = std::make_unique<SSLCertificateVerifierWrapper>(        jni, j_sslCertificateVerifier);    }    rtc::scoped_refptr<PeerConnectionInterface> pc =        PeerConnectionFactoryFromJava(factory)->CreatePeerConnection(rtc_config,         std::move(peer_connection_dependencies));

 3.4 我们继续分析 BasicPortAllocator 的接口 CreateSession
    ./p2p/base/port_allocator.cc

std::unique_ptr<PortAllocatorSession> PortAllocator::CreateSession(const std::string& content_name,    int component,    const std::string& ice_ufrag,    const std::string& ice_pwd) {    CheckRunOnValidThreadAndInitialized();    auto session = std::unique_ptr<PortAllocatorSession>(        CreateSessionInternal(content_name, component, ice_ufrag, ice_pwd));    session->SetCandidateFilter(candidate_filter());    return session;}

   ./p2p/client/basic_port_allocator.cc    

PortAllocatorSession* BasicPortAllocator::CreateSessionInternal(const std::string& content_name,int component, const std::string& ice_ufrag, const std::string& ice_pwd) {    CheckRunOnValidThreadAndInitialized();    PortAllocatorSession* session = new BasicPortAllocatorSession(this, content_name, component, ice_ufrag, ice_pwd);    session->SignalIceRegathering.connect(this, &BasicPortAllocator::OnIceRegathering);    return session;}

4.
./p2p/client/basic_port_allocator.cc

void BasicPortAllocatorSession::StartGettingPorts() {  RTC_DCHECK_RUN_ON(network_thread_);    state_ = SessionState::GATHERING;  if (!socket_factory_) {      owned_socket_factory_.reset(          new rtc::BasicPacketSocketFactory(network_thread_));      socket_factory_ = owned_socket_factory_.get();    }  network_thread_->Post(RTC_FROM_HERE, this, MSG_CONFIG_START);    RTC_LOG(LS_INFO) << "Start getting ports with turn_port_prune_policy "        << turn_port_prune_policy_;}

5.

void BasicPortAllocatorSession::OnMessage(rtc::Message* message) {  switch (message->message_id) {    case MSG_CONFIG_START:      GetPortConfigurations();      break;    case MSG_CONFIG_READY:      OnConfigReady(static_cast<PortConfiguration*>(message->pdata));      break;    case MSG_ALLOCATE:      OnAllocate();      break;    case MSG_SEQUENCEOBJECTS_CREATED:      OnAllocationSequenceObjectsCreated();      break;    case MSG_CONFIG_STOP:      OnConfigStop();      break;    default:      RTC_NOTREACHED();  }}

void BasicPortAllocatorSession::GetPortConfigurations() {  RTC_DCHECK_RUN_ON(network_thread_);  PortConfiguration* config =      new PortConfiguration(allocator_->stun_servers(), username(), password());  for (const RelayServerConfig& turn_server : allocator_->turn_servers()) {    config->AddRelay(turn_server);  }  ConfigReady(config);}

void BasicPortAllocatorSession::ConfigReady(PortConfiguration* config) {  RTC_DCHECK_RUN_ON(network_thread_);  network_thread_->Post(RTC_FROM_HERE, this, MSG_CONFIG_READY, config);}

6. 

void BasicPortAllocatorSession::OnConfigReady(PortConfiguration* config) {  RTC_DCHECK_RUN_ON(network_thread_);  if (config) {    configs_.push_back(config);  }  AllocatePorts();}

void BasicPortAllocatorSession::AllocatePorts() {  RTC_DCHECK_RUN_ON(network_thread_);  network_thread_->Post(RTC_FROM_HERE, this, MSG_ALLOCATE);}

7.

void BasicPortAllocatorSession::OnAllocate() {  RTC_DCHECK_RUN_ON(network_thread_);  if (network_manager_started_ && !IsStopped()) {    bool disable_equivalent_phases = true;    DoAllocate(disable_equivalent_phases);  }  allocation_started_ = true;}

void BasicPortAllocatorSession::DoAllocate(bool disable_equivalent)    AllocationSequence* sequence = new AllocationSequence(this, networks[i], config, sequence_flags);    sequence->SignalPortAllocationComplete.connect(this, &BasicPortAllocatorSession::OnPortAllocationComplete);    sequence->Init();    sequence->Start();    sequences_.push_back(sequence);    done_signal_needed = true;        network_thread_->Post(RTC_FROM_HERE, this, MSG_SEQUENCEOBJECTS_CREATED);

    7.1 sequence->Init() 这个里面创建了一个 UDP 套接字，并绑定读取接口
    ./p2p/client/basic_port_allocator.cc

    void AllocationSequence::Init() {        if (IsFlagSet(PORTALLOCATOR_ENABLE_SHARED_SOCKET)) {            udp_socket_.reset(session_->socket_factory()->CreateUdpSocket(            rtc::SocketAddress(network_->GetBestIP(), 0),            session_->allocator()->min_port(), session_->allocator()->max_port()));            if (udp_socket_) {                udp_socket_->SignalReadPacket.connect(this,  &AllocationSequence::OnReadPacket);            }            // Continuing if |udp_socket_| is NULL, as local TCP and RelayPort using TCP            // are next available options to setup a communication channel.        }    }

   7.2 sequence->Start()
        session_->network_thread()->Post(RTC_FROM_HERE, this, MSG_ALLOCATION_PHASE);
    
    7.3

    void AllocationSequence::OnMessage(rtc::Message* msg) {        const char* const PHASE_NAMES[kNumPhases] = {"Udp", "Relay", "Tcp"};        // Perform all of the phases in the current step.        RTC_LOG(LS_INFO) << network_->ToString() << ": Allocation Phase=" << PHASE_NAMES[phase_];        switch (phase_) {            case PHASE_UDP:            CreateUDPPorts();            CreateStunPorts();            break;        case PHASE_RELAY:            CreateRelayPorts();            break;        case PHASE_TCP:            CreateTCPPorts();            state_ = kCompleted;            break;        default:            RTC_NOTREACHED();        }        if (state() == kRunning) {            ++phase_;            session_->network_thread()->PostDelayed(RTC_FROM_HERE,                session_->allocator()->step_delay(),                this, MSG_ALLOCATION_PHASE);        } else {            // If all phases in AllocationSequence are completed, no allocation            // steps needed further. Canceling  pending signal.            session_->network_thread()->Clear(this, MSG_ALLOCATION_PHASE);            SignalPortAllocationComplete(this);        }    }

  7.4

    void AllocationSequence::CreateUDPPorts()        // 把上述创建的 udp_socket_         port = UDPPort::Create(        session_->network_thread(), session_->socket_factory(), network_,        session_->allocator()->min_port(), session_->allocator()->max_port(),        session_->username(), session_->password(),        session_->allocator()->origin(), emit_local_candidate_for_anyaddress,        session_->allocator()->stun_candidate_keepalive_interval());            //+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++        // 参见下面的 AddAllocatedPort 主要是 OnCandidateReady, OnPortComplete        //+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++        session_->AddAllocatedPort(port.release(), this, true);

        7.4.1
        ./p2p/base/stun_port.h

        static std::unique_ptr<UDPPort> Create(            rtc::Thread* thread, rtc::PacketSocketFactory* factory,            rtc::Network* network, rtc::AsyncPacketSocket* socket,            const std::string& username, const std::string& password,            const std::string& origin, bool emit_local_for_anyaddress,            absl::optional<int> stun_keepalive_interval) {            // Using `new` to access a non-public constructor.            auto port = absl::WrapUnique(new UDPPort(thread, factory, network, socket,                username, password, origin, emit_local_for_anyaddress));            port->set_stun_keepalive_delay(stun_keepalive_interval);            if (!port->Init()) {                return nullptr;            }            return port;        }

    
        7.4.2

        bool UDPPort::Init()            stun_keepalive_lifetime_ = GetStunKeepaliveLifetime();            if (!SharedSocket()) {                RTC_DCHECK(socket_ == nullptr);                //---------------------------------------------------------------------------------                // 这里的 socket_factory 其实就是上面的 BasicPacketSocketFactory 的接口，创建 socket 并绑定                //---------------------------------------------------------------------------------                socket_ = socket_factory()->CreateUdpSocket(                    rtc::SocketAddress(Network()->GetBestIP(), 0), min_port(), max_port());                if (!socket_) {                    RTC_LOG(LS_WARNING) << ToString() << ": UDP socket creation failed";                    return false;                }                socket_->SignalReadPacket.connect(this, &UDPPort::OnReadPacket);            }            socket_->SignalSentPacket.connect(this, &UDPPort::OnSentPacket);            socket_->SignalReadyToSend.connect(this, &UDPPort::OnReadyToSend);            socket_->SignalAddressReady.connect(this, &UDPPort::OnLocalAddressReady);            requests_.SignalSendPacket.connect(this, &UDPPort::OnSendPacket);

    到这一步，我们看到 socket 接收数据 pipeline (  参见下面流程 7.7.1 -->  7.7.7 ) 已经建立
        
    7.5

    void BasicPortAllocatorSession::AddAllocatedPort(Port* port,          AllocationSequence* seq, bool prepare_address) {        if (!port)            return;        port->set_content_name(content_name());        port->set_component(component());        port->set_generation(generation());        if (allocator_->proxy().type != rtc::PROXY_NONE)            port->set_proxy(allocator_->user_agent(), allocator_->proxy());        port->set_send_retransmit_count_attribute(            (flags() & PORTALLOCATOR_ENABLE_STUN_RETRANSMIT_ATTRIBUTE) != 0);        PortData data(port, seq);        ports_.push_back(data);        port->SignalCandidateReady.connect(            this, &BasicPortAllocatorSession::OnCandidateReady);        port->SignalCandidateError.connect(            this, &BasicPortAllocatorSession::OnCandidateError);        port->SignalPortComplete.connect(this,            &BasicPortAllocatorSession::OnPortComplete);        port->SignalDestroyed.connect(this,            &BasicPortAllocatorSession::OnPortDestroyed);        port->SignalPortError.connect(this, &BasicPortAllocatorSession::OnPortError);            RTC_LOG(LS_INFO) << port->ToString() << ": Added port to allocator";        if (prepare_address)            port->PrepareAddress();    }

    7.6

    void UDPPort::PrepareAddress() {        if (socket_->GetState() == rtc::AsyncPacketSocket::STATE_BOUND) {            OnLocalAddressReady(socket_, socket_->GetLocalAddress());        }    }        void UDPPort::OnLocalAddressReady(rtc::AsyncPacketSocket* socket,                                  const rtc::SocketAddress& address) {        // When adapter enumeration is disabled and binding to the any address, the        // default local address will be issued as a candidate instead if        // |emit_local_for_anyaddress| is true. This is to allow connectivity for        // applications which absolutely requires a HOST candidate.        rtc::SocketAddress addr = address;        // If MaybeSetDefaultLocalAddress fails, we keep the "any" IP so that at        // least the port is listening.        MaybeSetDefaultLocalAddress(&addr);        AddAddress(addr, addr, rtc::SocketAddress(), UDP_PROTOCOL_NAME, "", "",                 LOCAL_PORT_TYPE, ICE_TYPE_PREFERENCE_HOST, 0, "", false);        MaybePrepareStunCandidate();    }        void UDPPort::MaybePrepareStunCandidate() {        // Sending binding request to the STUN server if address is available to        // prepare STUN candidate.        if (!server_addresses_.empty()) {            SendStunBindingRequests();        } else {            // Port is done allocating candidates.            MaybeSetPortCompleteOrError();        }    }

   7.7 这个地方发送 stun 的绑定命令到 stun 服务器

    void UDPPort::SendStunBindingRequests() {        // We will keep pinging the stun server to make sure our NAT pin-hole stays        // open until the deadline (specified in SendStunBindingRequest).        RTC_DCHECK(requests_.empty());        for (ServerAddresses::const_iterator it = server_addresses_.begin();            it != server_addresses_.end(); ++it) {            SendStunBindingRequest(*it);        }    }    void UDPPort::SendStunBindingRequest(const rtc::SocketAddress& stun_addr) {         if (stun_addr.IsUnresolvedIP()) {            ResolveStunAddress(stun_addr);         } else if (socket_->GetState() == rtc::AsyncPacketSocket::STATE_BOUND) {            // Check if |server_addr_| is compatible with the port's ip.            if (IsCompatibleAddress(stun_addr)) {                  // 发送 stun 绑定命令                  requests_.Send(new StunBindingRequest(this, stun_addr, rtc::TimeMillis()));            } else {                  // Since we can't send stun messages to the server, we should mark this                  // port ready.                  const char* reason = "STUN server address is incompatible.";                  RTC_LOG(LS_WARNING) << reason;                  OnStunBindingOrResolveRequestFailed(stun_addr, SERVER_NOT_REACHABLE_ERROR, reason);             }          }    }

    //------------------------------------------------------------------------------------------------------------------------------------------
    // stun 服务器响应流程分析 ，顺便也分析了 socket 接收数据的一整个流程
    // MessageQueue::Get 这个流程上面已经分析过了， 就是初始化启动的一个线程  network_thread_
    //------------------------------------------------------------------------------------------------------------------------------------------
    7.7.1 --- 这个 ss_ 就是上面的 SocketServer::CreateDefault() 也就是 PhysicalSocketServer::Wait 接口

    bool MessageQueue::Get(Message* pmsg, int cmsWait, bool process_io)                if (!ss_->Wait(static_cast<int>(cmsNext), process_io))

    7.7.2  --- 就以 linux 下的 epoll 网络模型进行分析吧，大体差不多

    bool PhysicalSocketServer::Wait(int cmsWait, bool process_io)        return WaitEpoll(cmsWait);

    7.7.3

    bool PhysicalSocketServer::WaitEpoll(int cmsWait)        ProcessEvents(pdispatcher, readable, writable, check_error);

   7.7.4

    static void ProcessEvents(Dispatcher* dispatcher, bool readable, bool writable, bool check_error)        dispatcher->OnEvent(ff, errcode);

    7.7.5  ---  这个绑定关系参见开篇的 new AsyncUDPSocket(socket)
    ./rtc_base/physical_socket_server.cc

    void SocketDispatcher::OnEvent(uint32_t ff, int err)        SignalReadEvent(this);

   7.7.6  ---  这个的绑定关系参见上面的 7.4.2 UDPPort::Init()
    ./rtc_base/async_udp_socket.cc

    void AsyncUDPSocket::OnReadEvent(AsyncSocket* socket)        SignalReadPacket(this, buf_, static_cast<size_t>(len), remote_addr, (timestamp > -1 ? timestamp : TimeMicros()));

    7.7.7 ---  requests_ 其实就是对象 StunRequestManager
    ./p2p/base/stun_port.cc

    void UDPPort::OnReadPacket(rtc::AsyncPacketSocket* socket, const char* data, size_t size,        const rtc::SocketAddress& remote_addr, const int64_t& packet_time_us)        requests_.CheckResponse(data, size);

    7.7.8  ---   对应上面的 requests_.Send stun 的绑定命令          
    ./p2p/base/stun_request.cc

    bool StunRequestManager::CheckResponse(StunMessage* msg)        RequestMap::iterator iter = requests_.find(msg->transaction_id());        StunRequest* request = iter->second;        request->OnResponse(msg);

    7.7.9
    ./p2p/base/stun_port.cc

    void StunBindingRequest::OnResponse(StunMessage* response)         port_->OnStunBindingRequestSucceeded(this->Elapsed(), server_addr_, addr);

    7.8  --- stun 成功，则进入
    ./p2p/base/stun_port.cc

    void UDPPort::OnStunBindingRequestSucceeded(int rtt_ms,        const rtc::SocketAddress& stun_server_addr,        const rtc::SocketAddress& stun_reflected_addr)                AddAddress(stun_reflected_addr, socket_->GetLocalAddress(), related_address,               UDP_PROTOCOL_NAME, "", "", STUN_PORT_TYPE,               ICE_TYPE_PREFERENCE_SRFLX, 0, url.str(), false);

    7.9        

    void Port::AddAddress(const rtc::SocketAddress& address,                      const rtc::SocketAddress& base_address,                      const rtc::SocketAddress& related_address,                      const std::string& protocol,                      const std::string& relay_protocol,                      const std::string& tcptype,                      const std::string& type,                      uint32_t type_preference,                      uint32_t relay_preference,                      const std::string& url, bool is_final)                                                FinishAddingAddress(c, is_final);

    7.10  ---  SignalCandidateReady 的绑定关系见 7.5 

    void Port::FinishAddingAddress(const Candidate& c, bool is_final) {        candidates_.push_back(c);        SignalCandidateReady(this, c);        PostAddAddress(is_final);    }

    7.11 
    ./p2p/client/basic_port_allocator.cc

    void BasicPortAllocatorSession::OnCandidateReady(Port* port, const Candidate& c)             SignalPortReady(this, port);  --->             ./p2p/base/p2p_transport_channel.cc:168:  session->SignalPortReady.connect(this, &P2PTransportChannel::OnPortReady);            void P2PTransportChannel::OnPortReady(PortAllocatorSession* session, PortInterface* port)                CreateConnection(port, *iter, iter->origin_port());            bool P2PTransportChannel::CreateConnections(const Candidate& remote_candidate, PortInterface* origin_port)                if (CreateConnection(origin_port, remote_candidate, origin_port))            bool P2PTransportChannel::CreateConnection(PortInterface* port, const Candidate& remote_candidate, PortInterface* origin_port)                // 针对这个 port 创建一个 Connection，后续的数据接收都是通过这个进行了，表明 stun 已经成功                Connection* connection = port->CreateConnection(remote_candidate, origin);                AddConnection(connection);            void P2PTransportChannel::AddConnection(Connection* connection)                connection->SignalReadPacket.connect(this, &P2PTransportChannel::OnReadPacket);                                           SignalCandidatesReady(this, candidates);

    7.12
    ./p2p/base/p2p_transport_channel.cc

    void P2PTransportChannel::OnCandidatesReady(        PortAllocatorSession* session,        const std::vector<Candidate>& candidates) {        RTC_DCHECK_RUN_ON(network_thread_);        for (size_t i = 0; i < candidates.size(); ++i) {            SignalCandidateGathered(this, candidates[i]);        }    }

    7.13
    ./pc/jsep_transport_controller.cc

    void JsepTransportController::OnTransportCandidateGathered_n(        cricket::IceTransportInternal* transport,        const cricket::Candidate& candidate) {        RTC_DCHECK(network_thread_->IsCurrent());        // We should never signal peer-reflexive candidates.        if (candidate.type() == cricket::PRFLX_PORT_TYPE) {            RTC_NOTREACHED();            return;        }        std::string transport_name = transport->transport_name();        invoker_.AsyncInvoke<void>(            RTC_FROM_HERE, signaling_thread_, [this, transport_name, candidate] {            SignalIceCandidatesGathered(transport_name, {candidate});        });    }

    7.14
    ./pc/peer_connection.cc

    void PeerConnection::OnTransportControllerCandidatesGathered(        const std::string& transport_name,        const cricket::Candidates& candidates) {        int sdp_mline_index;        if (!GetLocalCandidateMediaIndex(transport_name, &sdp_mline_index)) {            RTC_LOG(LS_ERROR)                << "OnTransportControllerCandidatesGathered: content name "                << transport_name << " not found";            return;        }        for (cricket::Candidates::const_iterator citer = candidates.begin();            citer != candidates.end(); ++citer) {            // Use transport_name as the candidate media id.            std::unique_ptr<JsepIceCandidate> candidate(                new JsepIceCandidate(transport_name, sdp_mline_index, *citer));            if (local_description()) {                mutable_local_description()->AddCandidate(candidate.get());            }            OnIceCandidate(std::move(candidate));        }    }        void PeerConnection::OnIceCandidate(        std::unique_ptr<IceCandidateInterface> candidate) {      if (IsClosed()) {        return;      }      ReportIceCandidateCollected(candidate->candidate());      // 这个地方回调到 Java 层的接口，并把自己的 candidate 发送给对方      Observer()->OnIceCandidate(candidate.get());    }

8.

void BasicPortAllocatorSession::OnAllocationSequenceObjectsCreated() {  RTC_DCHECK_RUN_ON(network_thread_);  allocation_sequences_created_ = true;  // Send candidate allocation complete signal if we have no sequences.  MaybeSignalCandidatesAllocationDone();}

./p2p/base/p2p_transport_channel.cc

void P2PTransportChannel::OnCandidatesAllocationDone(    PortAllocatorSession* session) {  RTC_DCHECK_RUN_ON(network_thread_);  if (config_.gather_continually()) {    RTC_LOG(LS_INFO) << "P2PTransportChannel: " << transport_name()                     << ", component " << component()                     << " gathering complete, but using continual "                        "gathering so not changing gathering state.";    return;  }  gathering_state_ = kIceGatheringComplete;  RTC_LOG(LS_INFO) << "P2PTransportChannel: " << transport_name()                   << ", component " << component() << " gathering complete";  SignalGatheringState(this);}

./pc/jsep_transport_controller.cc

void JsepTransportController::OnTransportGatheringState_n(    cricket::IceTransportInternal* transport) {  RTC_DCHECK(network_thread_->IsCurrent());  UpdateAggregateStates_n();}