WebRtc信令交换的过程实际上是基于JSEP01(javascript session Establishment PRotocol)。 在上一篇[WebRtc建立P2P链接的总体流程]中只是描述了一种简单的形式，建立链接主要需要经过如下过程： 
以此为基础总结下p2p建立的过程！
peerconnection.cc
void PeerConnection::CreateOffer(CreateSessionDescriptionObserver* observer, const MediaConstraintsInterface* constraints) { if (!VERIFY(observer != NULL)) { LOG(LS_ERROR) << "CreateOffer - observer is NULL."; return; } RTCOfferAnswerOptions options; bool value; size_t mandatory_constraints = 0; //根据应用层传入的MediaConstraints设置RTCOfferAnswerOptions if (FindConstraint(constraints, MediaConstraintsInterface::kOfferToReceiveAudio, &value, &mandatory_constraints)) { options.offer_to_receive_audio = value ? RTCOfferAnswerOptions::kOfferToReceiveMediaTrue : 0; } ...... CreateOffer(observer, options);}void PeerConnection::CreateOffer(CreateSessionDescriptionObserver* observer, const RTCOfferAnswerOptions& options) { if (!VERIFY(observer != NULL)) { LOG(LS_ERROR) << "CreateOffer - observer is NULL."; return; } //此处的session为PeerConnection初始化时创建的WebRtcSession对象 session_->CreateOffer(observer, options);}接下来看WebRtcSession中的CreateOffer是如何实现的，见webrtcsession.cc中：
void WebRtcSession::CreateOffer( CreateSessionDescriptionObserver* observer, const PeerConnectionInterface::RTCOfferAnswerOptions& options) { //webrtc_session_desc_factory_是WebRtcSession在Initialize创建 //根据需求创建是否DTLS加密的WebRtcSessionDescriptionFactory webrtc_session_desc_factory_->CreateOffer(observer, options);}void WebRtcSessionDescriptionFactory::CreateOffer( CreateSessionDescriptionObserver* observer, const PeerConnectionInterface::RTCOfferAnswerOptions& options) { cricket::MediaSessionOptions session_options; std::string error = "CreateOffer"; if (certificate_request_state_ == CERTIFICATE_FAILED) { error += kFailedDueToIdentityFailed; LOG(LS_ERROR) << error; PostCreateSessionDescriptionFailed(observer, error); return; }/*在Java层org/appspot/apprtc/PeerConnectionClient.java中 private void createPeerConnectionInternal(EGLContext renderEGLContext){ ..... peerConnection = factory.createPeerConnection( rtcConfig, pcConstraints, pcObserver); isInitiator = false;..... mediaStream.addTrack(createVideoTrack(videoCapturer)); .... mediaStream.addTrack(factory.createAudioTrack( AUDIO_TRACK_ID, factory.createAudioSource(audioConstraints))); peerConnection.addStream(mediaStream); ..... } 最终是将会话中需要的MediaStream传入mediastream_signaling_进行管理！ 所以在mediastream_signaling_中可以获取MediaSessionOptions*/ if (!mediastream_signaling_->GetOptionsForOffer(options, &session_options)) { error += " called with invalid options."; LOG(LS_ERROR) << error; PostCreateSessionDescriptionFailed(observer, error); return; }//检测提供的audio video流是否合法即不能有相同的id if (!ValidStreams(session_options.streams)) { error += " called with invalid media streams."; LOG(LS_ERROR) << error; PostCreateSessionDescriptionFailed(observer, error); return; } if (data_channel_type_ == cricket::DCT_SCTP && mediastream_signaling_->HasDataChannels()) { //若在应用层建立了datachannel传输用户数据，设置成SCTP协议传输 session_options.data_channel_type = cricket::DCT_SCTP; } CreateSessionDescriptionRequest request( CreateSessionDescriptionRequest::kOffer, observer, session_options); if (certificate_request_state_ == CERTIFICATE_WAITING) { create_session_description_requests_.push(request); } else { ASSERT(certificate_request_state_ == CERTIFICATE_SUCCEEDED || certificate_request_state_ == CERTIFICATE_NOT_NEEDED); InternalCreateOffer(request); //根据request创建SDP }}//从上可以看出mediastreamsignaling.cc实际上是peerconnection.cc和webrtcsession.cc沟通的桥梁mediastreamsignaling.cc负责多媒体相关的管理！//根据request创建SDPvoid WebRtcSessionDescriptionFactory::InternalCreateOffer( CreateSessionDescriptionRequest request) { cricket::SessionDescription* desc( session_desc_factory_.CreateOffer( request.options, static_cast<cricket::BaseSession*>(session_)->local_description())); // RFC 3264 // When issuing an offer that modifies the session, // the "o=" line of the new SDP MUST be identical to that in the // previous SDP, except that the version in the origin field MUST // increment by one from the previous SDP. // Just increase the version number by one each time when a new offer // is created regardless if it's identical to the previous one or not. // The |session_version_| is a uint64, the wrap around should not happen. ASSERT(session_version_ + 1 > session_version_); //根据JSEP规范创建JsepSessionDescription JsepSessionDescription* offer(new JsepSessionDescription( JsepSessionDescription::kOffer)); if (!offer->Initialize(desc, session_id_, rtc::ToString(session_version_++))) { delete offer; PostCreateSessionDescriptionFailed(request.observer, "Failed to initialize the offer."); return; } if (session_->local_description() && !request.options.transport_options.ice_restart) { // Include all local ice candidates in the SessionDescription unless // the an ice restart has been requested. CopyCandidatesFromSessionDescription(session_->local_description(), offer); } //将创建的JsepSessionDescription回调给上层的应用，应用层可以结合自己的情况做相应更改 PostCreateSessionDescriptionSucceeded(request.observer, offer); }/*onCreateSuccess为org/appspot/apprtc/PeerConnectionClient.java中offer创建成功后调用的回调，在此会调用根据实际情况修改了音视频编解码相关的信息private class SDPObserver implements SdpObserver { @Override public void onCreateSuccess(final SessionDescription origSdp) { if (localSdp != null) { reportError("Multiple SDP create."); return; } String sdpDescription = origSdp.description; if (preferIsac) { sdpDescription = preferCodec(sdpDescription, AUDIO_CODEC_ISAC, true); } if (videoCallEnabled && preferH264) { sdpDescription = preferCodec(sdpDescription, VIDEO_CODEC_H264, false); } final SessionDescription sdp = new SessionDescription( origSdp.type, sdpDescription); localSdp = sdp; executor.execute(new Runnable() { @Override public void run() { if (peerConnection != null && !isError) { Log.d(TAG, "Set local SDP from " + sdp.type); peerConnection.setLocalDescription(sdpObserver, sdp); } } }); }*/下面我们看看PeerConnectionClient.java中回调的实现
private class SDPObserver implements SdpObserver { @Override public void onCreateSuccess(final SessionDescription origSdp) { if (localSdp != null) { reportError("Multiple SDP create."); return; } String sdpDescription = origSdp.description; //根据平台，更改音视频的编码方式 if (preferIsac) { sdpDescription = preferCodec(sdpDescription, AUDIO_CODEC_ISAC, true); } if (videoCallEnabled && preferH264) { sdpDescription = preferCodec(sdpDescription, VIDEO_CODEC_H264, false); } //创建最终的SessionDescription final SessionDescription sdp = new SessionDescription( origSdp.type, sdpDescription); localSdp = sdp; executor.execute(new Runnable() { @Override public void run() { if (peerConnection != null && !isError) { Log.d(TAG, "Set local SDP from " + sdp.type); //更新本地的SessionDescription,若设置成功将回调SDPObserver的onSetSuccess方法 peerConnection.setLocalDescription(sdpObserver, sdp); } } }); } @Override public void onSetSuccess() { executor.execute(new Runnable() { @Override public void run() { if (peerConnection == null || isError) { return; } if (isInitiator) { //发起呼叫端逻辑 // For offering peer connection we first create offer and set // local SDP, then after receiving answer set remote SDP. if (peerConnection.getRemoteDescription() == null) { // We've just set our local SDP so time to send it. Log.d(TAG, "Local SDP set succesfully"); //将local SDP发送到服务器 events.onLocalDescription(localSdp); } else { // We've just set remote description, so drain remote // and send local ICE candidates. Log.d(TAG, "Remote SDP set succesfully"); drainCandidates(); } } else { //被动应答端逻辑 // For answering peer connection we set remote SDP and then // create answer and set local SDP. if (peerConnection.getLocalDescription() != null) { // We've just set our local SDP so time to send it, drain // remote and send local ICE candidates. Log.d(TAG, "Local SDP set succesfully"); events.onLocalDescription(localSdp); drainCandidates(); } else { // We've just set remote SDP - do nothing for now - // answer will be created soon. Log.d(TAG, "Remote SDP set succesfully"); } } } }); } .....}peerConnection.setLocalDescription(sdpObserver, sdp);的实现如下：
void PeerConnection::SetLocalDescription( SetSessionDescriptionObserver* observer, SessionDescriptionInterface* desc) { ...... //检测传入的desc是否合法更新状态后，设置到 if (!session_->SetLocalDescription(desc, &error)) { PostSetSessionDescriptionFailure(observer, error); return; } SetSessionDescriptionMsg* msg = new SetSessionDescriptionMsg(observer); //回调sdpObserver.onSetSuccess将local SDP发送给服务器 signaling_thread()->Post(this, MSG_SET_SESSIONDESCRIPTION_SUCCESS, msg); // MaybeStartGathering needs to be called after posting // MSG_SET_SESSIONDESCRIPTION_SUCCESS, so that we don't signal any candidates // before signaling that SetLocalDescription completed. //设置LocalDescription后向ICE服务器发出请求StartGathering session_->MaybeStartGathering(); }bool WebRtcSession::SetLocalDescription(SessionDescriptionInterface* desc, std::string* err_desc) { ASSERT(signaling_thread()->IsCurrent()); // Takes the ownership of |desc| regardless of the result. rtc::scoped_ptr<SessionDescriptionInterface> desc_temp(desc); // Validate SDP. if (!ValidateSessionDescription(desc, cricket::CS_LOCAL, err_desc)) { return false; } // Update the initiator flag if this session is the initiator. Action action = GetAction(desc->type()); if (state() == STATE_INIT && action == kOffer) { set_initiator(true); } ..... //设置本地description set_local_description(desc->description()->Copy()); local_desc_.reset(desc_temp.release()); // Transport and Media channels will be created only when offer is set. //CreateChannels根据本地的description创建audiochannel videochannel datachannel if (action == kOffer && !CreateChannels(local_desc_->description())) { // TODO(mallinath) - Handle CreateChannel failure, as new local description // is applied. Restore back to old description. return BadLocalSdp(desc->type(), kCreateChannelFailed, err_desc); } ...... if (remote_description()) { //如果会话已经设置了远程description // Now that we have a local description, we can push down remote candidates // that we stored, and those from the remote description. if (!saved_candidates_.empty()) { // If there are saved candidates which arrived before the local // description was set, copy those to the remote description. CopySavedCandidates(remote_desc_.get()); } // Push remote candidates in remote description to transport channels. //从远程SDP中获取condidate并通过 //TransportController::AddRemoteCandidates-->P2PTransportChannel::AddRemoteCandidate--->bool P2PTransportChannel::CreateConnections //最终会建立P2P的链接，前面文章提到在ClientB链接到服务器时，服务器会将ClientA的SDP返回给B端，B端首先会设置remote_description //所以在ClientB设置local description时开始与ClientA建立p2p链接 UseCandidatesInSessionDescription(remote_desc_.get()); } // Update state and SSRC of local MediaStreams and DataChannels based on the // local session description. mediastream_signaling_->OnLocalDescriptionChanged(local_desc_.get()); rtc::SSLRole role; if (data_channel_type_ == cricket::DCT_SCTP && GetSslRole(&role)) { mediastream_signaling_->OnDtlsRoleReadyForSctp(role); } if (error() != cricket::BaseSession::ERROR_NONE) { return BadLocalSdp(desc->type(), GetSessionErrorMsg(), err_desc); } return true;}上面的分析提到，在void PeerConnection::SetLocalDescription中会调用 session_->MaybeStartGathering()开始访问iceserver获取本地的condidate
void BaseSession::MaybeStartGathering() {//直接调用TransportController类中的MaybeStartGathering方法 transport_controller_->MaybeStartGathering();}void TransportController::MaybeStartGathering() { //在工作线程worker_thread_中调用MaybeStartGathering_w方法！ worker_thread_->Invoke<void>( rtc::Bind(&TransportController::MaybeStartGathering_w, this));}void TransportController::MaybeStartGathering_w() {//transports_为map<std::string, Transport*> TransportMap，最终会调用P2PTransportChannel::MaybeStartGathering() for (const auto& kv : transports_) { kv.second->MaybeStartGathering(); }}void P2PTransportChannel::MaybeStartGathering() { // Start gathering if we never started before, or if an ICE restart occurred. if (allocator_sessions_.empty() || IceCredentialsChanged(allocator_sessions_.back()->ice_ufrag(), allocator_sessions_.back()->ice_pwd(), ice_ufrag_, ice_pwd_)) { if (gathering_state_ != kIceGatheringGathering) { gathering_state_ = kIceGatheringGathering; SignalGatheringState(this); } // Time for a new allocator AddAllocatorSession(allocator_->CreateSession( SessionId(), transport_name(), component(), ice_ufrag_, ice_pwd_)); }}void P2PTransportChannel::AddAllocatorSession(PortAllocatorSession* session) { session->set_generation(static_cast<uint32>(allocator_sessions_.size())); allocator_sessions_.push_back(session); // We now only want to apply new candidates that we receive to the ports // created by this new session because these are replacing those of the // previous sessions. ports_.clear();//通过信号与槽的方式获取底层的通知事件，此处的port不仅仅是传统意义上的端口//实际代表的是一种通信协议，eg:TCPPort ,UDPPort,StunPort,RelayPort session->SignalPortReady.connect(this, &P2PTransportChannel::OnPortReady); //底层每发现一个condiatate都会通知到P2PTransportChannel::OnCandidatesReady session->SignalCandidatesReady.connect( this, &P2PTransportChannel::OnCandidatesReady); session->SignalCandidatesAllocationDone.connect( this, &P2PTransportChannel::OnCandidatesAllocationDone); // session实际为BasicPortAllocatorSession session->StartGettingPorts();}void BasicPortAllocatorSession::StartGettingPorts() { network_thread_ = rtc::Thread::Current(); if (!socket_factory_) { owned_socket_factory_.reset( new rtc::BasicPacketSocketFactory(network_thread_)); socket_factory_ = owned_socket_factory_.get(); } running_ = true; network_thread_->Post(this, MSG_CONFIG_START); if (flags() & PORTALLOCATOR_ENABLE_SHAKER) network_thread_->PostDelayed(ShakeDelay(), this, MSG_SHAKE);}//经过一次处理MSG_CONFIG_START ==>MSG_CONFIG_READY==>MSG_ALLOCATE 消息==>OnAllocate==>DoAllocate()//为本机每一个物理网络分配portsvoid BasicPortAllocatorSession::DoAllocate() { bool done_signal_needed = false; std::vector<rtc::Network*> networks; //获取本机的网络信息eg：ip 。。。 GetNetworks(&networks); if (networks.empty()) { LOG(LS_WARNING) << "Machine has no networks; no ports will be allocated"; done_signal_needed = true; } else { for (uint32 i = 0; i < networks.size(); ++i) { PortConfiguration* config = NULL; if (configs_.size() > 0) config = configs_.back(); uint32 sequence_flags = flags(); if ((sequence_flags & DISABLE_ALL_PHASES) == DISABLE_ALL_PHASES) { // If all the ports are disabled we should just fire the allocation // done event and return. done_signal_needed = true; break; } if (!config || config->relays.empty()) { // No relay ports specified in this config. sequence_flags |= PORTALLOCATOR_DISABLE_RELAY; } if (!(sequence_flags & PORTALLOCATOR_ENABLE_IPV6) && networks[i]->GetBestIP().family() == AF_INET6) { // Skip IPv6 networks unless the flag's been set. continue; } // Disable phases that would only create ports equivalent to // ones that we have already made. DisableEquivalentPhases(networks[i], config, &sequence_flags); if ((sequence_flags & DISABLE_ALL_PHASES) == DISABLE_ALL_PHASES) { // New AllocationSequence would have nothing to do, so don't make it. continue; } AllocationSequence* sequence = new AllocationSequence(this, networks[i], config, sequence_flags); if (!sequence->Init()) {//初始化udp_socket_:AsyncPacketSocket delete sequence; continue; } done_signal_needed = true; //ports分配完毕后会触发BasicPortAllocatorSession::OnPortAllocationComplete 将事件抛给上层 sequence->SignalPortAllocationComplete.connect( this, &BasicPortAllocatorSession::OnPortAllocationComplete); if (running_) //开始分配各种ports PHASE_UDP ->PHASE_RELAY->PHASE_TCP ->PHASE_SSLTCP sequence->Start(); sequences_.push_back(sequence); } } if (done_signal_needed) { network_thread_->Post(this, MSG_SEQUENCEOBJECTS_CREATED); }}//每种端口在分配时，会根据相应的协议获取Condidate将会通过SignalCandidatesReady信号通知到上层！//BasicPortAllocatorSession::SignalCandidatesReady==>//P2PTransportChannel::OnCandidatesReady==>P2PTransportChannel::SignalCandidateGathered==>//Transport::OnChannelCandidateGathered==>Transport::SignalCandidatesGathered==>//TransportController::OnTransportCandidatesGathered_w==>TransportController::SignalCandidatesGathered==>//WebRtcSession::OnTransportControllerCandidatesGathered==> ice_observer_->OnIceCandidate(&candidate);//最终会调用应用层实现的IceObserver.OnIceCandidate在谷歌的WebRtc的demo中由PeerConnectionClient.java PCObserver实现，如下：
private class PCObserver implements PeerConnection.Observer { @Override public void onIceCandidate(final IceCandidate candidate){ executor.execute(new Runnable() { @Override public void run() { events.onIceCandidate(candidate); } }); } @Override public void onSignalingChange( PeerConnection.SignalingState newState) { Log.d(TAG, "SignalingState: " + newState); }......}//events为PeerConnectionEvents 由CallActivity实现 public class CallActivity extends Activity implements AppRTCClient.SignalingEvents, PeerConnectionClient.PeerConnectionEvents, CallFragment.OnCallEvents{ ...... @Override public void onIceCandidate(final IceCandidate candidate) { runOnUiThread(new Runnable() { @Override public void run() { if (appRtcClient != null) { appRtcClient.sendLocalIceCandidate(candidate); } } }); } ......}public class WebSocketRTCClient implements AppRTCClient, WebSocketChannelEvents{ ..... // Send Ice candidate to the other participant. @Override public void sendLocalIceCandidate(final IceCandidate candidate) { executor.execute(new Runnable() { @Override public void run() { JSONObject json = new JSONObject(); jsonPut(json, "type", "candidate"); jsonPut(json, "label", candidate.sdpMLineIndex); jsonPut(json, "id", candidate.sdpMid); jsonPut(json, "candidate", candidate.sdp); if (initiator) { // Call initiator sends ice candidates to GAE server. if (roomState != ConnectionState.CONNECTED) { reportError("Sending ICE candidate in non connected state."); return; } //offer端通过http先将本地candidate发送到远程服务器，再由远程服务器发送到响应端！ sendPostMessage(MessageType.MESSAGE, messageUrl, json.toString()); if (connectionParameters.loopback) { events.onRemoteIceCandidate(candidate); } } else { // Call receiver sends ice candidates to websocket server. wsClient.send(json.toString()); } } }); } ..... }到目前位置ClientA端的工作告一段落，假如服务器服务器接受到响应端的SDP并转发给ClientA后，ClientA接下来会发生些什么呢？
服务器通过WebSocket将SDP发送到ClientA端后，最终会触发WebSocketChannelEvents.onWebSocketMessage方法，实现如下：
/*WebSocketRTCClient.java*/public class WebSocketRTCClient implements AppRTCClient, WebSocketChannelEvents { ...... @Override public void onWebSocketMessage(final String msg) { if (wsClient.getState() != WebSocketConnectionState.REGISTERED) { Log.e(TAG, "Got WebSocket message in non registered state."); return; } try { JSONObject json = new JSONObject(msg); String msgText = json.getString("msg"); String errorText = json.optString("error"); if (msgText.length() > 0) { json = new JSONObject(msgText); String type = json.optString("type"); //从服务器消息中获取到candidate if (type.equals("candidate")) { IceCandidate candidate = new IceCandidate( json.getString("id"), json.getInt("label"), json.getString("candidate")); events.onRemoteIceCandidate(candidate); } else if (type.equals("answer")) { if (initiator) { //Offer端收到了响应端的SDP SessionDescription sdp = new SessionDescription( SessionDescription.Type.fromCanonicalForm(type), json.getString("sdp")); events.onRemoteDescription(sdp); } else { reportError("Received answer for call initiator: " + msg); } } else if (type.equals("offer")) { if (!initiator) { //响应端收到了Offer端的SDP SessionDescription sdp = new SessionDescription( SessionDescription.Type.fromCanonicalForm(type), json.getString("sdp")); events.onRemoteDescription(sdp); } else { reportError("Received offer for call receiver: " + msg); } } else if (type.equals("bye")) { events.onChannelClose(); } else { reportError("Unexpected WebSocket message: " + msg); } } else { if (errorText != null && errorText.length() > 0) { reportError("WebSocket error message: " + errorText); } else { reportError("Unexpected WebSocket message: " + msg); } } } catch (JSONException e) { reportError("WebSocket message JSON parsing error: " + e.toString()); } } ......}我们先看events.onRemoteDescription(sdp)的实现，后面再看 events.onRemoteIceCandidate(candidate)
events.onRemoteDescription(sdp)--> peerConnectionClient.setRemoteDescription(sdp); public void setRemoteDescription(final SessionDescription sdp) { executor.execute(new Runnable() { @Override public void run() { ...... SessionDescription sdpRemote = new SessionDescription( sdp.type, sdpDescription); peerConnection.setRemoteDescription(sdpObserver, sdpRemote); ...... } }); }void PeerConnection::SetRemoteDescription( SetSessionDescriptionObserver* observer, SessionDescriptionInterface* desc) { if (!VERIFY(observer != NULL)) { LOG(LS_ERROR) << "SetRemoteDescription - observer is NULL."; return; } if (!desc) { PostSetSessionDescriptionFailure(observer, "SessionDescription is NULL."); return; } // Update stats here so that we have the most recent stats for tracks and // streams that might be removed by updating the session description. stats_->UpdateStats(kStatsOutputLevelStandard); std::string error; //设置会话中的RemoteDescription，会根据RemoteDescription创建响应的channel if (!session_->SetRemoteDescription(desc, &error)) { PostSetSessionDescriptionFailure(observer, error); return; } SetSessionDescriptionMsg* msg = new SetSessionDescriptionMsg(observer); //设置成功后调用相应的回调通知应用 signaling_thread()->Post(this, MSG_SET_SESSIONDESCRIPTION_SUCCESS, msg);}bool WebRtcSession::SetRemoteDescription(SessionDescriptionInterface* desc, std::string* err_desc) { ASSERT(signaling_thread()->IsCurrent()); // Takes the ownership of |desc| regardless of the result. rtc::scoped_ptr<SessionDescriptionInterface> desc_temp(desc); // Validate SDP. if (!ValidateSessionDescription(desc, cricket::CS_REMOTE, err_desc)) { return false; } // Transport and Media channels will be created only when offer is set. Action action = GetAction(desc->type()); //根据远程的SDP 创建会话需要的channel if (action == kOffer && !CreateChannels(desc->description())) { // TODO(mallinath) - Handle CreateChannel failure, as new local description // is applied. Restore back to old description. return BadRemoteSdp(desc->type(), kCreateChannelFailed, err_desc); } // Remove unused channels if MediaContentDescription is rejected. RemoveUnusedChannels(desc->description()); // NOTE: Candidates allocation will be initiated only when SetLocalDescription // is called. //设置远程的description set_remote_description(desc->description()->Copy()); if (!UpdateSessionState(action, cricket::CS_REMOTE, err_desc)) { return false; }......}响应端的各个过程与此类似，只是相应的顺序不一样，结合Offer端不难明白！
新闻热点
疑难解答
