shell登录鉴权改成广播方式

删除rtsp/rtmp shell命令,添加media命令
加载配置文件后发送广播

src/Common/config.cpp

@@ -24,6 +24,7 @@
  * SOFTWARE.
  */
 
+#include <Util/NoticeCenter.h>
 #include "Common/config.h"
 #include "Util/util.h"
 #include "Util/onceToken.h"
@@ -33,12 +34,20 @@ using namespace ZL::Network;
 
 namespace Config {
 
-void loadIniConfig(){
+bool loadIniConfig(const char *ini_path){
-	auto &ini = ZL::Util::mINI::Instance();
+    string ini;
+    if(ini_path){
+        ini = ini_path;
+    }else{
+        ini = exePath() + ".ini";
+    }
 	try{
-		ini.parseFile(exePath() + ".ini");
+        mINI::Instance().parseFile(ini);
+        NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastUpdateConfig);
+        return true;
 	}catch (std::exception &ex) {
-		ini.dumpFile(exePath() + ".ini");
+        mINI::Instance().dumpFile(ini);
+        return false;
 	}
 }
 ////////////广播名称///////////
@@ -51,7 +60,11 @@ const char kBroadcastOnRtspAuth[] = "kBroadcastOnRtspAuth";
 const char kBroadcastMediaPlayed[] = "kBroadcastMediaPlayed";
 const char kBroadcastRtmpPublish[] = "kBroadcastRtmpPublish";
 const char kBroadcastFlowReport[] = "kBroadcastFlowReport";
-const char kFlowThreshold[] = "Broadcast.flowThreshold";
+const char kBroadcastUpdateConfig[] = "kBroadcastUpdateConfig";
+const char kBroadcastShellLogin[] = "kBroadcastShellLogin";
+
+const char kFlowThreshold[] = "broadcast.flowThreshold";
+
 onceToken token([](){
     mINI::Instance()[kFlowThreshold] = 1024;
 },nullptr);

src/Common/config.h

@@ -29,13 +29,19 @@
 #define COMMON_CONFIG_H
 
 #include "Util/mini.h"
+#include "Util/onceToken.h"
 #include <functional>
 using namespace std;
 using namespace ZL::Util;
 
 namespace Config {
 
-void loadIniConfig();
+//加载配置文件，如果配置文件不存在，那么会导出默认配置并生成配置文件
+//加载配置文件成功后会触发kBroadcastUpdateConfig广播
+//如果指定的文件名(ini_path)为空，那么会加载默认配置文件
+//默认配置文件名为 /path/to/your/exe.ini
+//加载配置文件成功后返回true，否则返回false
+bool loadIniConfig(const char *ini_path = nullptr);
 ////////////TCP最大连接数///////////
 #define MAX_TCP_SESSION 100000
 ////////////其他宏定义///////////
@@ -96,12 +102,22 @@ extern const char kBroadcastRtmpPublish[];
 extern const char kBroadcastMediaPlayed[];
 #define BroadcastMediaPlayedArgs const MediaInfo &args,const Broadcast::AuthInvoker &invoker
 
+//shell登录鉴权
+extern const char kBroadcastShellLogin[];
+#define BroadcastShellLoginArgs const string &user_name,const string &passwd,const Broadcast::AuthInvoker &invoker
+
 //停止rtsp/rtmp/http-flv会话后流量汇报事件广播
 extern const char kBroadcastFlowReport[];
 #define BroadcastFlowReportArgs const MediaInfo &args,const uint64_t &totalBytes
 
 //流量汇报事件流量阈值,单位KB，默认1MB
 extern const char kFlowThreshold[];
+
+//更新配置文件事件广播,执行loadIniConfig函数加载配置文件成功后会触发该广播
+extern const char kBroadcastUpdateConfig[];
+#define BroadcastUpdateConfigArgs void
+#define ReloadConfigTag  ((void *)(0xFF))
+
 } //namespace Broadcast
 
 ////////////HTTP配置///////////

src/Device/PlayerProxy.cpp

@@ -131,6 +131,7 @@ void PlayerProxy::initMedia() {
 		return;
 	}
 	m_pChn.reset(new DevChannel(m_strVhost.data(),m_strApp.data(),m_strSrc.data(),getDuration(),m_bEnableHls,m_bEnableMp4));
+    m_pChn->setListener(shared_from_this());
 	if (containVideo()) {
 		VideoInfo info;
 		info.iFrameRate = getVideoFps();
@@ -146,6 +147,18 @@ void PlayerProxy::initMedia() {
 		m_pChn->initAudio(info);
 	}
 }
+bool PlayerProxy::shutDown() {
+    //通知其停止推流
+    weak_ptr<PlayerProxy> weakSlef = dynamic_pointer_cast<PlayerProxy>(shared_from_this());
+    ASYNC_TRACE([weakSlef](){
+        auto stronSelf = weakSlef.lock();
+        if(stronSelf){
+            stronSelf->m_pChn.reset();
+            stronSelf->teardown();
+        }
+    });
+    return true;
+}
 
 
 } /* namespace Player */

src/Device/PlayerProxy.h

@@ -38,7 +38,7 @@ using namespace ZL::Player;
 namespace ZL {
 namespace DEV {
 
-class PlayerProxy :public MediaPlayer, public std::enable_shared_from_this<PlayerProxy>{
+class PlayerProxy :public MediaPlayer, public std::enable_shared_from_this<PlayerProxy> , public MediaSourceEvent {
 public:
 	typedef std::shared_ptr<PlayerProxy> Ptr;
 
@@ -54,7 +54,7 @@ public:
 	virtual ~PlayerProxy();
 
 	void play(const char* strUrl) override;
-
+    bool shutDown() override;
 private:
     bool m_bEnableHls;
     bool m_bEnableMp4;

src/Shell/ShellCMD.cpp

@@ -3,56 +3,94 @@
 //
 
 #include "Util/CMD.h"
-#include "Rtsp/RtspMediaSource.h"
+#include "Common/MediaSource.h"
-#include "Rtmp/RtmpMediaSource.h"
 
 using namespace ZL::Util;
-using namespace ZL::Rtsp;
+using namespace ZL::Media;
-using namespace ZL::Rtmp;
 
 namespace ZL {
 namespace Shell {
 
-class CMD_rtsp: public CMD {
+
+class CMD_media: public CMD {
 public:
-    CMD_rtsp(){
+    CMD_media(){
-        _parser.reset(new OptionParser(nullptr));
+        _parser.reset(new OptionParser([](const std::shared_ptr<ostream> &stream,mINI &ini){
-        (*_parser) << Option('l', "list", Option::ArgNone, nullptr,false, "list all media source of rtsp",
+            MediaSource::for_each_media([&](const string &schema,
-                             [](const std::shared_ptr<ostream> &stream, const string &arg) {
+                                            const string &vhost,
-//                                 auto mediaSet = RtspMediaSource::getMediaSet();
+                                            const string &app,
-//                                 for (auto &src : mediaSet) {
+                                            const string &streamid,
-//                                     (*stream) << "\t" << src << "\r\n";
+                                            const MediaSource::Ptr &media){
-//                                 }
+                if(!ini["schema"].empty() && ini["schema"] != schema){
-                                 return false;
+                    //筛选协议不匹配
-                             });
+                    return;
+                }
+                if(!ini["vhost"].empty() && ini["vhost"] != vhost){
+                    //筛选虚拟主机不匹配
+                    return;
+                }
+                if(!ini["app"].empty() && ini["app"] != app){
+                    //筛选应用名不匹配
+                    return;
+                }
+                if(!ini["stream"].empty() && ini["stream"] != streamid){
+                    //流id不匹配
+                    return;
+                }
+                if(ini.find("list") != ini.end()){
+                    //列出源
+                    (*stream) << "\t"
+                              << schema << "/"
+                              << vhost << "/"
+                              << app << "/"
+                              << streamid
+                              << "\r\n";
+                    return;
+                }
+
+                if(ini.find("kick") != ini.end()){
+                    //踢出源
+                    do{
+                        if(!media) {
+                            break;
+                        }
+                        if(!media->shutDown()) {
+                            break;
+                        }
+                        (*stream) << "\t踢出成功:"
+                                  << schema << "/"
+                                  << vhost << "/"
+                                  << app << "/"
+                                  << streamid
+                                  << "\r\n";
+                        return;
+                    }while(0);
+                    (*stream) << "\t踢出失败:"
+                              << schema << "/"
+                              << vhost << "/"
+                              << app << "/"
+                              << streamid
+                              << "\r\n";
+                    return;
+                }
+
+            });
+        }));
+        (*_parser) << Option('k', "kick", Option::ArgNone,nullptr,false, "踢出媒体源", nullptr);
+        (*_parser) << Option('l', "list", Option::ArgNone,nullptr,false, "列出媒体源", nullptr);
+        (*_parser) << Option('S', "schema", Option::ArgRequired,nullptr,false, "协议筛选", nullptr);
+        (*_parser) << Option('v', "vhost", Option::ArgRequired,nullptr,false, "虚拟主机筛选", nullptr);
+        (*_parser) << Option('a', "app", Option::ArgRequired,nullptr,false, "应用名筛选", nullptr);
+        (*_parser) << Option('s', "stream", Option::ArgRequired,nullptr,false, "流id筛选", nullptr);
     }
-    virtual ~CMD_rtsp() {}
+    virtual ~CMD_media() {}
     const char *description() const override {
-        return "查看rtsp服务器相关信息.";
+        return "媒体源相关操作.";
-    }
-};
-class CMD_rtmp: public CMD {
-public:
-    CMD_rtmp(){
-        _parser.reset(new OptionParser(nullptr));
-        (*_parser) << Option('l', "list", Option::ArgNone,nullptr,false, "list all media source of rtmp",
-                             [](const std::shared_ptr<ostream> &stream, const string &arg) {
-//                                 auto mediaSet = RtmpMediaSource::getMediaSet();
-//                                 for (auto &src : mediaSet) {
-//                                     (*stream) << "\t" << src << "\r\n";
-//                                 }
-                                 return false;
-                             });
-    }
-    virtual ~CMD_rtmp() {}
-    const char *description() const override {
-        return "查看rtmp服务器相关信息.";
     }
 };
 
 static onceToken s_token([]() {
-    REGIST_CMD(rtmp);
+    REGIST_CMD(media);
-    REGIST_CMD(rtsp);
 }, nullptr);
 
 

src/Shell/ShellSession.cpp

@@ -28,14 +28,14 @@
 #include "Common/config.h"
 #include "Util/CMD.h"
 #include "Util/onceToken.h"
+#include "Util/NoticeCenter.h"
 
+using namespace Config;
 using namespace ZL::Util;
 
 namespace ZL {
 namespace Shell {
 
-unordered_map<string, string> ShellSession::g_mapUser;
-
 ShellSession::ShellSession(const std::shared_ptr<ThreadPool> &_th,
                            const Socket::Ptr &_sock) :
         TcpLimitedSession(_th, _sock) {
@@ -82,8 +82,9 @@ void ShellSession::onManager() {
 }
 
 inline bool ShellSession::onCommandLine(const string& line) {
-	if (m_requestCB) {
+    auto loginInterceptor = m_loginInterceptor;
-		bool ret = m_requestCB(line);
+    if (loginInterceptor) {
+		bool ret = loginInterceptor(line);
 		return ret;
 	}
     try {
@@ -103,30 +104,55 @@ inline bool ShellSession::onCommandLine(const string& line) {
 inline void ShellSession::pleaseInputUser() {
 	send("\033[0m");
 	send(StrPrinter << SERVER_NAME << " login: " << endl);
-	m_requestCB = [this](const string &line) {
+	m_loginInterceptor = [this](const string &user_name) {
-		m_strUserName=line;
+		m_strUserName=user_name;
         pleaseInputPasswd();
 		return true;
 	};
 }
 inline void ShellSession::pleaseInputPasswd() {
 	send("Password: \033[8m");
-	m_requestCB = [this](const string &passwd) {
+	m_loginInterceptor = [this](const string &passwd) {
-		if(!onAuth(m_strUserName, passwd)) {
+        auto onAuth = [this](const string &errMessage){
-			send(StrPrinter
+            if(!errMessage.empty()){
-					<<"\033[0mPermission denied,"
+                //鉴权失败
-					<<" please try again.\r\n"
+                send(StrPrinter
-					<<m_strUserName<<"@"<<SERVER_NAME
+                     <<"\033[0mAuth failed("
-					<<"'s password: \033[8m"
+                     << errMessage
-					<<endl);
+                     <<"), please try again.\r\n"
-			return true;
+                     <<m_strUserName<<"@"<<SERVER_NAME
-		}
+                     <<"'s password: \033[8m"
-		send("\033[0m");
+                     <<endl);
-		send("-----------------------------------------\r\n");
+                return;
-		send(StrPrinter<<"欢迎来到"<<SERVER_NAME<<", 你可输入\"help\"查看帮助.\r\n"<<endl);
+            }
-		send("-----------------------------------------\r\n");
+            send("\033[0m");
-        printShellPrefix();
+            send("-----------------------------------------\r\n");
-		m_requestCB=nullptr;
+            send(StrPrinter<<"欢迎来到"<<SERVER_NAME<<", 你可输入\"help\"查看帮助.\r\n"<<endl);
+            send("-----------------------------------------\r\n");
+            printShellPrefix();
+            m_loginInterceptor=nullptr;
+        };
+
+        weak_ptr<ShellSession> weakSelf = dynamic_pointer_cast<ShellSession>(shared_from_this());
+        Broadcast::AuthInvoker invoker = [weakSelf,onAuth](const string &errMessage){
+            auto strongSelf =  weakSelf.lock();
+            if(!strongSelf){
+                return;
+            }
+            strongSelf->async([errMessage,weakSelf,onAuth](){
+                auto strongSelf =  weakSelf.lock();
+                if(!strongSelf){
+                    return;
+                }
+                onAuth(errMessage);
+            });
+        };
+
+        auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastShellLogin,m_strUserName,passwd,invoker);
+        if(!flag){
+            //如果无人监听shell登录事件，那么默认shell无法登录
+            onAuth("please listen kBroadcastShellLogin event");
+        }
 		return true;
 	};
 }
@@ -135,14 +161,5 @@ inline void ShellSession::printShellPrefix() {
 	send(StrPrinter << m_strUserName << "@" << SERVER_NAME << "# " << endl);
 }
 
-inline bool ShellSession::onAuth(const string &user, const string &pwd) {
-	auto it = g_mapUser.find(user);
-	if (it == g_mapUser.end()) {
-		//WarnL << user << " " << pwd;
-		return false;
-	}
-	return it->second == pwd;
-}
-
 }/* namespace Shell */
 } /* namespace ZL */

src/Shell/ShellSession.h

@@ -48,22 +48,16 @@ public:
     void onError(const SockException &err) override {};
 	void onManager() override;
 
-	static void addUser(const string &userName,const string &userPwd){
-		g_mapUser[userName] = userPwd;
-	}
 private:
 	inline bool onCommandLine(const string &);
-	inline bool onAuth(const string &user, const string &pwd);
 	inline void pleaseInputUser();
 	inline void pleaseInputPasswd();
 	inline void printShellPrefix();
 
-	function<bool(const string &)> m_requestCB;
+	function<bool(const string &)> m_loginInterceptor;
 	string m_strRecvBuf;
 	Ticker m_beatTicker;
 	string m_strUserName;
-
-	static unordered_map<string, string> g_mapUser;
 };
 
 } /* namespace Shell */

tests/test_server.cpp

@@ -112,16 +112,24 @@ static onceToken s_token([](){
 	NoticeCenter::Instance().addListener(nullptr,Config::Broadcast::kBroadcastRtmpPublish,[](BroadcastRtmpPublishArgs){
         InfoL << args.m_vhost << " " << args.m_app << " " << args.m_streamid << " " << args.m_param_strs ;
         EventPoller::Instance().async([invoker](){
-            //invoker("");//鉴权成功
+            invoker("");//鉴权成功
-            invoker("this is auth failed message");//鉴权失败
+            //invoker("this is auth failed message");//鉴权失败
         });
     });
 
     NoticeCenter::Instance().addListener(nullptr,Config::Broadcast::kBroadcastMediaPlayed,[](BroadcastMediaPlayedArgs){
         InfoL << args.m_schema << " " << args.m_vhost << " " << args.m_app << " " << args.m_streamid << " " << args.m_param_strs ;
         EventPoller::Instance().async([invoker](){
-            //invoker("");//鉴权成功
+            invoker("");//鉴权成功
-            invoker("this is auth failed message");//鉴权失败
+            //invoker("this is auth failed message");//鉴权失败
+        });
+    });
+
+    NoticeCenter::Instance().addListener(nullptr,Config::Broadcast::kBroadcastShellLogin,[](BroadcastShellLoginArgs){
+        InfoL << "shell login:" << user_name << " " << passwd;
+        EventPoller::Instance().async([invoker](){
+            invoker("");//鉴权成功
+            //invoker("this is auth failed message");//鉴权失败
         });
     });
 
@@ -130,7 +138,9 @@ static onceToken s_token([](){
 int main(int argc,char *argv[]){
 	//设置退出信号处理函数
 	signal(SIGINT, [](int){EventPoller::Instance().shutdown();});
-	//设置日志
+    signal(SIGHUP, [](int){Config::loadIniConfig();});
+
+    //设置日志
 	Logger::Instance().add(std::make_shared<ConsoleChannel>("stdout", LTrace));
     Logger::Instance().setWriter(std::make_shared<AsyncLogWriter>());
 	//加载配置文件，如果配置文件不存在就创建一个
@@ -176,10 +186,8 @@ int main(int argc,char *argv[]){
 #endif //ENABLE_OPENSSL
 
 	//简单的telnet服务器，可用于服务器调试，但是不能使用23端口，否则telnet上了莫名其妙的现象
-	//测试方法:telnet 127.0.0.1 8023
+	//测试方法:telnet 127.0.0.1 9000
-	//输入用户名和密码登录(user:test,pwd:123456)，输入help命令查看帮助
 	TcpServer<ShellSession>::Ptr shellSrv(new TcpServer<ShellSession>());
-	ShellSession::addUser("test", "123456");
 	shellSrv->start(mINI::Instance()[Config::Shell::kPort]);
 	//开启rtsp/rtmp/http服务器
 	TcpServer<RtspSession>::Ptr rtspSrv(new TcpServer<RtspSession>());