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

概述

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

跨职能图

mediasoup源码分析(三)channel创建及信令交互-LMLPHP

业务流程图

mediasoup源码分析(三)channel创建及信令交互-LMLPHP

数据发送有两种方式:
应用层发送的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<uv_stream_t*>(this->uvHandle),
        static_cast<uv_alloc_cb>(onAlloc),
        static_cast<uv_read_cb>(onRead));
    if (err != 0)
    {
        uv_close(reinterpret_cast<uv_handle_t*>(this->uvHandle), static_cast<uv_close_cb>(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<uv_stream_t*>(this->uvHandle),
        static_cast<uv_alloc_cb>(onAlloc),
        static_cast<uv_read_cb>(onRead));

跳转到onRead中

inline static void onRead(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf)
{
	auto* socket = static_cast<UnixStreamSocket*>(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<char*>(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<const uint8_t*>(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<std::string, Request::MethodId> 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

06-24 12:15