首页 > 代码库 > SRS学习笔记10-SrsConnection及其子类分析
SRS学习笔记10-SrsConnection及其子类分析
SrsConnection类代表一个client的连接,其中封装了st thread,用于在一个单独的st thread里处理一个client的服务请求.
SrsConnection在 int SrsServer::accept_client(SrsListenerType type, st_netfd_t client_stfd)里创建
SrsConnection* conn = NULL; if (type == SrsListenerRtmpStream) { conn = new SrsRtmpConn(this, client_stfd); } else if (type == SrsListenerHttpApi) { #ifdef SRS_AUTO_HTTP_API conn = new SrsHttpApi(this, client_stfd, http_api_mux); #else srs_warn("close http client for server not support http-api"); srs_close_stfd(client_stfd); return ret; #endif } else if (type == SrsListenerHttpStream) { #ifdef SRS_AUTO_HTTP_SERVER conn = new SrsResponseOnlyHttpConn(this, client_stfd, http_server); #else srs_warn("close http client for server not support http-server"); srs_close_stfd(client_stfd); return ret; #endif } else { // TODO: FIXME: handler others } srs_assert(conn);
创建完成后调用 conn->start()启动,此后这个client的请求都在这个st thread 里完成
SrsConection类
/** * the manager for connection.
* SrsServer类实现了这一接口,用来对SrsConnection类进行管理. */ class IConnectionManager { public: IConnectionManager(); virtual ~IConnectionManager(); public: /** * remove the specified connection. */ virtual void remove(SrsConnection* c) = 0; }; /** * the basic connection of SRS, * all connections accept from listener must extends from this base class, * server will add the connection to manager, and delete it when remove. */ class SrsConnection : public virtual ISrsOneCycleThreadHandler, public virtual IKbpsDelta { private: /** * each connection start a green thread, * when thread stop, the connection will be delete by server.
* 业务处理线程 */ SrsOneCycleThread* pthread; /** * the id of connection. */ int id; protected: /** * the manager object to manage the connection. */ IConnectionManager* manager; /** * the underlayer st fd handler. */ st_netfd_t stfd; /** * the ip of client. */ std::string ip; /** * whether the connection is disposed, * when disposed, connection should stop cycle and cleanup itself. */ bool disposed; /** * whether connection is expired, application definition. * when expired, the connection must never be served and quit ASAP. */ bool expired; public: SrsConnection(IConnectionManager* cm, st_netfd_t c); virtual ~SrsConnection(); public: /** * to dipose the connection. */ virtual void dispose(); /** * start the client green thread. * when server get a client from listener, * 1. server will create an concrete connection(for instance, RTMP connection), * 2. then add connection to its connection manager, * 3. start the client thread by invoke this start() * when client cycle thread stop, invoke the on_thread_stop(), which will use server * to remove the client by server->remove(this). */ virtual int start(); // interface ISrsOneCycleThreadHandler public: /** * the thread cycle function, * when serve connection completed, terminate the loop which will terminate the thread, * thread will invoke the on_thread_stop() when it terminated. */ virtual int cycle(); /** * when the thread cycle finished, thread will invoke the on_thread_stop(), * which will remove self from server, server will remove the connection from manager * then delete the connection. */ virtual void on_thread_stop(); public: /** * get the srs id which identify the client. */ virtual int srs_id(); /** * set connection to expired. */ virtual void expire(); protected: /** * for concrete connection to do the cycle.
* 具体connection 的处理函数,参见SrsRtmpConn. */ virtual int do_cycle() = 0; };
int SrsConnection::start() {
// 启动SrsOneCycleThread类 return pthread->start(); } int SrsConnection::cycle() { int ret = ERROR_SUCCESS; //这是个全局的id生成器,也用来生成st thread的id _srs_context->generate_id(); id = _srs_context->get_id(); // 取得客户端的ip地址 ip = srs_get_peer_ip(st_netfd_fileno(stfd)); // 调用子类如SrsRtmpConn的处理函数 ret = do_cycle(); // if socket io error, set to closed. if (srs_is_client_gracefully_close(ret)) { ret = ERROR_SOCKET_CLOSED; } // success. if (ret == ERROR_SUCCESS) { srs_trace("client finished."); } // client close peer. if (ret == ERROR_SOCKET_CLOSED) { srs_warn("client disconnect peer. ret=%d", ret); } return ERROR_SUCCESS; } void SrsConnection::on_thread_stop() { // TODO: FIXME: never remove itself, use isolate thread to do cleanup.
// 用在SrsServer类的conns向量数组中删除此对象
manager->remove(this); } int SrsConnection::srs_id() { return id; } void SrsConnection::expire() { expired = true; }
SrsRtmpConn类
SrsRtmpConn::SrsRtmpConn(SrsServer* svr, st_netfd_t c) : SrsConnection(svr, c) { server = svr; req = new SrsRequest(); res = new SrsResponse(); skt = new SrsStSocket(c);//实现ISrsProtocolReaderWriter接口,即st socket的读写函数,以及读写超时,发送和接受到的字节数 rtmp = new SrsRtmpServer(skt);// 提供rtmp-command-protocol服务,是一个上层的面向协议,media stream的服务,
// 例如连接vhost/app,播放流,获取视频/音频数据,与之对应的是 SrsRtmpClient.
refer = new SrsRefer(); bandwidth = new SrsBandwidth(); security = new SrsSecurity(); duration = 0; kbps = new SrsKbps(); kbps->set_io(skt, skt); wakable = NULL; mw_sleep = SRS_PERF_MW_SLEEP; mw_enabled = false; realtime = SRS_PERF_MIN_LATENCY_ENABLED; send_min_interval = 0; tcp_nodelay = false; client_type = SrsRtmpConnUnknown; // 在配置改变时,支持reload _srs_config->subscribe(this); }
构造函数初始化成员变量
// TODO: return detail message when error for client. int SrsRtmpConn::do_cycle() { int ret = ERROR_SUCCESS; srs_trace("RTMP client ip=%s", ip.c_str());
// 设置SrsRtmpServer里面的底层数据通讯类SrsProtocol的读写超时 rtmp->set_recv_timeout(SRS_CONSTS_RTMP_RECV_TIMEOUT_US); rtmp->set_send_timeout(SRS_CONSTS_RTMP_SEND_TIMEOUT_US); // 握手 if ((ret = rtmp->handshake()) != ERROR_SUCCESS) { srs_error("rtmp handshake failed. ret=%d", ret); return ret; } srs_verbose("rtmp handshake success"); // 获取连接的app if ((ret = rtmp->connect_app(req)) != ERROR_SUCCESS) { srs_error("rtmp connect vhost/app failed. ret=%d", ret); return ret; } srs_verbose("rtmp connect app success"); // set client ip to request. req->ip = ip; // 假设推流地址 rtmp://192.168.151.151/live/marstv // discovery vhost, resolve the vhost from config
req->vhost 是192.168.151.151
SrsConfDirective* parsed_vhost = _srs_config->get_vhost(req->vhost);
// 没有设置 192.168.151.151 这样的vhost parsed_vhost是默认的vhost if (parsed_vhost) { req->vhost = parsed_vhost->arg0();// req->vhost 是 __defaultVhost__ } srs_info("discovery app success. schema=%s, vhost=%s, port=%s, app=%s", req->schema.c_str(), req->vhost.c_str(), req->port.c_str(), req->app.c_str()); if (req->schema.empty() || req->vhost.empty() || req->port.empty() || req->app.empty()) { ret = ERROR_RTMP_REQ_TCURL; srs_error("discovery tcUrl failed. " "tcUrl=%s, schema=%s, vhost=%s, port=%s, app=%s, ret=%d", req->tcUrl.c_str(), req->schema.c_str(), req->vhost.c_str(), req->port.c_str(), req->app.c_str(), ret); return ret; } // check vhost
// 在 check_vhost里面会调用 http_hooks_on_connect if ((ret = check_vhost()) != ERROR_SUCCESS) { srs_error("check vhost failed. ret=%d", ret); return ret; } srs_verbose("check vhost success."); srs_trace("connect app, " "tcUrl=%s, pageUrl=%s, swfUrl=%s, schema=%s, vhost=%s, port=%s, app=%s, args=%s", req->tcUrl.c_str(), req->pageUrl.c_str(), req->swfUrl.c_str(), req->schema.c_str(), req->vhost.c_str(), req->port.c_str(), req->app.c_str(), (req->args? "(obj)":"null")); // show client identity if(req->args) { std::string srs_version; std::string srs_server_ip; int srs_pid = 0; int srs_id = 0; SrsAmf0Any* prop = NULL; if ((prop = req->args->ensure_property_string("srs_version")) != NULL) { srs_version = prop->to_str(); } if ((prop = req->args->ensure_property_string("srs_server_ip")) != NULL) { srs_server_ip = prop->to_str(); } if ((prop = req->args->ensure_property_number("srs_pid")) != NULL) { srs_pid = (int)prop->to_number(); } if ((prop = req->args->ensure_property_number("srs_id")) != NULL) { srs_id = (int)prop->to_number(); } srs_info("edge-srs ip=%s, version=%s, pid=%d, id=%d", srs_server_ip.c_str(), srs_version.c_str(), srs_pid, srs_id); if (srs_pid > 0) { srs_trace("edge-srs ip=%s, version=%s, pid=%d, id=%d", srs_server_ip.c_str(), srs_version.c_str(), srs_pid, srs_id); } } ret = service_cycle(); http_hooks_on_close(); return ret; }
SRS学习笔记10-SrsConnection及其子类分析
声明:以上内容来自用户投稿及互联网公开渠道收集整理发布,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任,若内容有误或涉及侵权可进行投诉: 投诉/举报 工作人员会在5个工作日内联系你,一经查实,本站将立刻删除涉嫌侵权内容。