mediasoup源码分析--channel创建及信令交互

• • 概述
  • 跨职能图
  • 业务流程图
  • 代码剖析
  • tips

概述

在golang实现mediasoup的tcp服务及channel通道一文中，已经介绍过信令服务中tcp和channel的创建，本文主要讲解c++中mediasoup的channel创建，以及信令服务和mediasoup服务如何交互

跨职能图

业务流程图

数据发送有两种方式:
应用层发送的request最后被封装在Requst对象中，其中包含着"id"，因为Request对象中包含着Channel::UnixStreamSocket对象，所以可以直接调用Request::Accept()将处理后的结果告诉应用层进程。
Worker进程也可以主动给应用层进程发送消息，通过Notifier::Emit()即可以给应用进程发送消息，Notifier类中有Channel::UnixStreamSocket，所以直接调用Channel::UnixStreamSocket::Send()就可以发送消息。Notifier类内部的数据成员和函数成员都是静态的，所以在任意位置可以直接通过Channel::Notifier::Emit()函数发送消息。

代码剖析

1.channel创建

int main(int argc, char* argv[]) {     // Ensure we are called by our Node library.     if (argc == 1)     {         std::cerr << "ERROR: you don't seem to be my real father" << std::endl;          std::_Exit(EXIT_FAILURE);     }      std::string id = std::string(argv[1]);     std::string ip = std::string(argv[2]);     int port = atoi(argv[3]);     int iperfPort = atoi(argv[4]);      // Initialize libuv stuff (we need it for the Channel).     DepLibUV::ClassInit();     //..........省略部分代码..............     // Set the Channel socket (this will be handled and deleted by the Worker).     printf("new Channel to %s:%d\n",ip.c_str(),port);     auto* channel = new Channel::UnixStreamSocket(ip,port);     //..........省略部分代码..............     try         {             // Run the Worker.             Worker worker(id,channel);              // Worker ended.             destroy();              exitSuccess();         }     catch (const MediaSoupError& error)         {             MS_ERROR_STD("failure exit: %s", error.what());              destroy();             exitWithError();         } } 

UnixStreamSocket构造函数

UnixStreamSocket::UnixStreamSocket(const std::string& ip,int port) : ::UnixStreamSocket::UnixStreamSocket(ip,port, MaxSize) {     MS_TRACE_STD();      // Create the JSON reader.     {         Json::CharReaderBuilder builder;         Json::Value settings = Json::nullValue;         Json::Value invalidSettings;          builder.strictMode(&settings);          MS_ASSERT(builder.validate(&invalidSettings), "invalid Json::CharReaderBuilder");          this->jsonReader = builder.newCharReader();     }      // Create the JSON writer.     {         Json::StreamWriterBuilder builder;         Json::Value invalidSettings;          builder["commentStyle"]            = "None";         builder["indentation"]             = "";         builder["enableYAMLCompatibility"] = false;         builder["dropNullPlaceholders"]    = false;          MS_ASSERT(builder.validate(&invalidSettings), "invalid Json::StreamWriterBuilder");          this->jsonWriter = builder.newStreamWriter();     } } 

跳转到handles\UnixStreamSocket.cpp下

UnixStreamSocket::UnixStreamSocket( const std::string& ip,int port,size_t bufferSize) : bufferSize(bufferSize) {     printf("::UnixStreamSocket::UnixStreamSocket\n");     MS_TRACE_STD();      int err;      this->uvHandle       = new uv_tcp_t;     this->uvHandle->data = (void*)this;     err = uv_tcp_init(DepLibUV::GetLoop(), this->uvHandle);     if (err != 0)     {         delete this->uvHandle;         this->uvHandle = nullptr;         printf("uv_tcp_init() failed: %s\n", uv_strerror(err));         MS_THROW_ERROR_STD("uv_tcp_init() failed: %s", uv_strerror(err));     }     struct sockaddr_in dest;     uv_ip4_addr(ip.c_str(), port, &dest);     this->connect = new uv_connect_t;     printf("will connect to %s:%d\n",ip.c_str(),port);     err = uv_tcp_connect(this->connect, this->uvHandle, (const struct sockaddr*)&dest, onConnect);     if (err != 0)     {         delete this->uvHandle;         this->uvHandle = nullptr;         printf("uv_tcp_connect() failed: %s\n", uv_strerror(err));         MS_THROW_ERROR_STD("uv_tcp_connect() failed: %s", uv_strerror(err));     }     // Start reading.     err = uv_read_start(         reinterpret_cast(this->uvHandle),         static_cast(onAlloc),         static_cast(onRead));     if (err != 0)     {         uv_close(reinterpret_cast(this->uvHandle), static_cast(onClose));          MS_THROW_ERROR_STD("uv_read_start() failed: %s", uv_strerror(err));     }     // NOTE: Don't allocate the buffer here. Instead wait for the first uv_alloc_cb(). } 

代码中的uv_read_start接口中onRead回调

    err = uv_read_start(         reinterpret_cast(this->uvHandle),         static_cast(onAlloc),         static_cast(onRead)); 

跳转到onRead中

inline static void onRead(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) { 	auto* socket = static_cast(handle->data);  	if (socket == nullptr) 		return;  	socket->OnUvRead(nread, buf); } 

OnUvRead中调用UserOnUnixStreamRead

	void UnixStreamSocket::UserOnUnixStreamRead() 	{ 		MS_TRACE_STD();  		// Be ready to parse more than a single message in a single TCP chunk. 		while (true) 		{ 			if (IsClosed()) 				return;  			size_t readLen  = this->bufferDataLen - this->msgStart; 			char* jsonStart = nullptr; 			size_t jsonLen; 			int nsRet = netstring_read( 			  reinterpret_cast(this->buffer + this->msgStart), readLen, &jsonStart, &jsonLen);             //.............省略部分代码.............. 			// If here it means that jsonStart points to the beginning of a JSON string 			// with jsonLen bytes length, so recalculate readLen. 			readLen = 			  reinterpret_cast(jsonStart) - (this->buffer + this->msgStart) + jsonLen + 1;  			Json::Value json; 			std::string jsonParseError;  			if (this->jsonReader->parse( 			      (const char*)jsonStart, (const char*)jsonStart + jsonLen, &json, &jsonParseError)) 			{ 				Channel::Request* request = nullptr;  				try 				{ 					request = new Channel::Request(this, json); 				} 				catch (const MediaSoupError& error) 				{ 					MS_ERROR_STD("discarding wrong Channel request"); 				}  				if (request != nullptr) 				{ 					// Notify the listener. 					this->listener->OnChannelRequest(this, request);  					// Delete the Request. 					delete request; 				}                 //.............省略部分代码..............                 ... 		} 	}  

channel创建完成，至此，跳转到worker.cpp中的OnChannelRequest接口。mediasoup监听channel信令并根据request->methodId分类处理
根据request->methodId，分别执行不同的业务
request->methodId有如下分类

	std::unordered_map Request::string2MethodId = 	{ 		{ "worker.dump",                       Request::MethodId::WORKER_DUMP                          }, 		{ "worker.updateSettings",             Request::MethodId::WORKER_UPDATE_SETTINGS               }, 		{ "worker.createRouter",               Request::MethodId::WORKER_CREATE_ROUTER                 }, 		{ "router.close",                      Request::MethodId::ROUTER_CLOSE                         }, 		{ "router.dump",                       Request::MethodId::ROUTER_DUMP                          }, 		{ "router.createWebRtcTransport",      Request::MethodId::ROUTER_CREATE_WEBRTC_TRANSPORT       }, 		{ "router.createPlainRtpTransport",    Request::MethodId::ROUTER_CREATE_PLAIN_RTP_TRANSPORT    }, 		{ "router.createProducer",             Request::MethodId::ROUTER_CREATE_PRODUCER               }, 		{ "router.createConsumer",             Request::MethodId::ROUTER_CREATE_CONSUMER               }, 		{ "router.setAudioLevelsEvent",        Request::MethodId::ROUTER_SET_AUDIO_LEVELS_EVENT        }, 		{ "transport.close",                   Request::MethodId::TRANSPORT_CLOSE                      }, 		{ "transport.dump",                    Request::MethodId::TRANSPORT_DUMP                       }, 		{ "transport.getStats",                Request::MethodId::TRANSPORT_GET_STATS                  }, 		{ "transport.setRemoteDtlsParameters", Request::MethodId::TRANSPORT_SET_REMOTE_DTLS_PARAMETERS }, 		{ "transport.setRemoteParameters",     Request::MethodId::TRANSPORT_SET_REMOTE_PARAMETERS      }, 		{ "transport.setMaxBitrate",           Request::MethodId::TRANSPORT_SET_MAX_BITRATE            }, 		{ "transport.changeUfragPwd",          Request::MethodId::TRANSPORT_CHANGE_UFRAG_PWD           }, 		{ "transport.startMirroring",          Request::MethodId::TRANSPORT_START_MIRRORING            }, 		{ "transport.stopMirroring",           Request::MethodId::TRANSPORT_STOP_MIRRORING             }, 		{ "producer.close",                    Request::MethodId::PRODUCER_CLOSE                       }, 		{ "producer.dump",                     Request::MethodId::PRODUCER_DUMP                        }, 		{ "producer.getStats",                 Request::MethodId::PRODUCER_GET_STATS                   }, 		{ "producer.pause",                    Request::MethodId::PRODUCER_PAUSE                       }, 		{ "producer.resume" ,                  Request::MethodId::PRODUCER_RESUME                      }, 		{ "producer.setPreferredProfile",      Request::MethodId::PRODUCER_SET_PREFERRED_PROFILE       }, 		{ "consumer.close",                    Request::MethodId::CONSUMER_CLOSE                       }, 		{ "consumer.dump",                     Request::MethodId::CONSUMER_DUMP                        }, 		{ "consumer.getStats",                 Request::MethodId::CONSUMER_GET_STATS                   }, 		{ "consumer.enable",                   Request::MethodId::CONSUMER_ENABLE                      }, 		{ "consumer.pause",                    Request::MethodId::CONSUMER_PAUSE                       }, 		{ "consumer.resume",                   Request::MethodId::CONSUMER_RESUME                      }, 		{ "consumer.setPreferredProfile",      Request::MethodId::CONSUMER_SET_PREFERRED_PROFILE       }, 		{ "consumer.setEncodingPreferences",   Request::MethodId::CONSUMER_SET_ENCODING_PREFERENCES    }, 		{ "consumer.requestKeyFrame",          Request::MethodId::CONSUMER_REQUEST_KEY_FRAME           } 	}; 

下一章节介绍mediasoup如何将信令返回值及其他通知信息推送到信令服务，敬请期待！

tips

更多关于mediasoup的文章可以进入我的专栏查看
http://t.csdnimg.cn/3UQeL