RtmpPusher.cpp 8.79 KB
Newer Older
xiongziliang committed
1
/*
xiongziliang committed
2
 * MIT License
xzl committed
3
 *
xiongziliang committed
4
 * Copyright (c) 2016-2019 xiongziliang <771730766@qq.com>
xiongziliang committed
5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24
 *
 * This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
xzl committed
25 26 27
 */
#include "RtmpPusher.h"
#include "Rtmp/utils.h"
xiongzilaing committed
28 29 30
#include "Util/util.h"
#include "Util/onceToken.h"
#include "Thread/ThreadPool.h"
xiongziliang committed
31
using namespace toolkit;
xiongziliang committed
32
using namespace mediakit::Client;
xzl committed
33

xiongziliang committed
34
namespace mediakit {
xzl committed
35

36 37
static int kSockFlags = SOCKET_DEFAULE_FLAGS | FLAG_MORE;

38
RtmpPusher::RtmpPusher(const EventPoller::Ptr &poller,const RtmpMediaSource::Ptr &src) : TcpClient(poller){
39
	_pMediaSrc=src;
xzl committed
40 41 42 43 44 45 46 47
}

RtmpPusher::~RtmpPusher() {
	teardown();
	DebugL << endl;
}
void RtmpPusher::teardown() {
	if (alive()) {
48 49 50
		_strApp.clear();
		_strStream.clear();
		_strTcUrl.clear();
xzl committed
51
		{
52 53
			lock_guard<recursive_mutex> lck(_mtxOnResultCB);
			_mapOnResultCB.clear();
xzl committed
54
		}
xzl committed
55
        {
56 57
            lock_guard<recursive_mutex> lck(_mtxOnStatusCB);
            _dqOnStatusCB.clear();
xzl committed
58
        }
59
		_pPublishTimer.reset();
xiongziliang committed
60
        reset();
xiongziliang committed
61
        shutdown(SockException(Err_shutdown,"teardown"));
xzl committed
62 63 64
	}
}

xiongziliang committed
65
void RtmpPusher::onPublishResult(const SockException &ex) {
66 67 68 69 70 71 72 73 74 75 76
	if(_pPublishTimer){
		//播放结果回调
		_pPublishTimer.reset();
		if(_onPublished){
			_onPublished(ex);
		}
	} else {
		//播放成功后异常断开回调
		if(_onShutdown){
			_onShutdown(ex);
		}
xiongziliang committed
77 78 79 80 81 82 83
	}

	if(ex){
		teardown();
	}
}

xiongziliang committed
84
void RtmpPusher::publish(const string &strUrl)  {
xzl committed
85
	teardown();
xiongziliang committed
86 87 88
	string strHost = FindField(strUrl.data(), "://", "/");
	_strApp = 	FindField(strUrl.data(), (strHost + "/").data(), "/");
    _strStream = FindField(strUrl.data(), (strHost + "/" + _strApp + "/").data(), NULL);
89
    _strTcUrl = string("rtmp://") + strHost + "/" + _strApp;
xzl committed
90

91
    if (!_strApp.size() || !_strStream.size()) {
xiongziliang committed
92
        onPublishResult(SockException(Err_other,"rtmp url非法"));
xzl committed
93 94
        return;
    }
95
	DebugL << strHost << " " << _strApp << " " << _strStream;
xzl committed
96

xiongziliang committed
97
	auto iPort = atoi(FindField(strHost.data(), ":", NULL).data());
xzl committed
98 99 100 101 102
	if (iPort <= 0) {
        //rtmp 默认端口1935
		iPort = 1935;
	} else {
        //服务器域名
xiongziliang committed
103
		strHost = FindField(strHost.data(), NULL, ":");
xzl committed
104
	}
xiongziliang committed
105 106

    weak_ptr<RtmpPusher> weakSelf = dynamic_pointer_cast<RtmpPusher>(shared_from_this());
xiongziliang committed
107
    float playTimeOutSec = (*this)[kTimeoutMS].as<int>() / 1000.0;
xiongziliang committed
108 109 110 111 112 113 114 115 116 117 118 119 120
    _pPublishTimer.reset( new Timer(playTimeOutSec,  [weakSelf]() {
        auto strongSelf=weakSelf.lock();
        if(!strongSelf) {
            return false;
        }
        strongSelf->onPublishResult(SockException(Err_timeout,"publish rtmp timeout"));
        return false;
    },getPoller()));

    if(!(*this)[kNetAdapter].empty()){
        setNetAdapter((*this)[kNetAdapter]);
    }

xzl committed
121 122 123 124
	startConnect(strHost, iPort);
}

void RtmpPusher::onErr(const SockException &ex){
xiongziliang committed
125
	onPublishResult(ex);
xzl committed
126 127
}
void RtmpPusher::onConnect(const SockException &err){
xiongziliang committed
128
	if(err) {
xiongziliang committed
129
		onPublishResult(err);
xzl committed
130 131
		return;
	}
132 133 134
	//推流器不需要多大的接收缓存,节省内存占用
	_sock->setReadBuffer(std::make_shared<BufferRaw>(1 * 1024));

xzl committed
135 136 137 138 139 140
	weak_ptr<RtmpPusher> weakSelf = dynamic_pointer_cast<RtmpPusher>(shared_from_this());
	startClientSession([weakSelf](){
        auto strongSelf=weakSelf.lock();
        if(!strongSelf) {
            return;
        }
xiongziliang committed
141 142

        strongSelf->sendChunkSize(60000);
xzl committed
143 144 145
        strongSelf->send_connect();
	});
}
146
void RtmpPusher::onRecv(const Buffer::Ptr &pBuf){
xzl committed
147 148 149 150
	try {
		onParseRtmp(pBuf->data(), pBuf->size());
	} catch (exception &e) {
		SockException ex(Err_other, e.what());
xiongziliang committed
151
		onPublishResult(ex);
xzl committed
152 153 154 155 156 157
	}
}


inline void RtmpPusher::send_connect() {
	AMFValue obj(AMF_OBJECT);
158
	obj.set("app", _strApp);
xzl committed
159
	obj.set("type", "nonprivate");
160 161
	obj.set("tcUrl", _strTcUrl);
	obj.set("swfUrl", _strTcUrl);
xzl committed
162 163 164 165 166 167 168 169
	sendInvoke("connect", obj);
	addOnResultCB([this](AMFDecoder &dec){
		//TraceL << "connect result";
		dec.load<AMFValue>();
		auto val = dec.load<AMFValue>();
		auto level = val["level"].as_string();
		auto code = val["code"].as_string();
		if(level != "status"){
170
			throw std::runtime_error(StrPrinter <<"connect 失败:" << level << " " << code << endl);
xzl committed
171 172 173 174 175 176 177 178 179 180 181
		}
		send_createStream();
	});
}

inline void RtmpPusher::send_createStream() {
	AMFValue obj(AMF_NULL);
	sendInvoke("createStream", obj);
	addOnResultCB([this](AMFDecoder &dec){
		//TraceL << "createStream result";
		dec.load<AMFValue>();
182
		_ui32StreamId = dec.load<int>();
xzl committed
183 184 185 186 187
		send_publish();
	});
}
inline void RtmpPusher::send_publish() {
	AMFEncoder enc;
188
	enc << "publish" << ++_iReqID << nullptr << _strStream << _strApp ;
xzl committed
189 190 191 192 193 194
	sendRequest(MSG_CMD, enc.data());

	addOnStatusCB([this](AMFValue &val) {
		auto level = val["level"].as_string();
		auto code = val["code"].as_string();
		if(level != "status") {
195
			throw std::runtime_error(StrPrinter <<"publish 失败:" << level << " " << code << endl);
xzl committed
196 197 198 199 200 201 202
		}
		//start send media
		send_metaData();
	});
}

inline void RtmpPusher::send_metaData(){
203
    auto src = _pMediaSrc.lock();
xzl committed
204
    if (!src) {
205
        throw std::runtime_error("the media source was released");
xzl committed
206
    }
207

xzl committed
208 209 210 211
    AMFEncoder enc;
    enc << "@setDataFrame" << "onMetaData" <<  src->getMetaData();
    sendRequest(MSG_DATA, enc.data());
    
xiongziliang committed
212
    src->getConfigFrame([&](const RtmpPacket::Ptr &pkt){
213
        sendRtmp(pkt->typeId, _ui32StreamId, pkt, pkt->timeStamp, pkt->chunkId );
xzl committed
214 215
    });
    
216
    _pRtmpReader = src->getRing()->attach(getPoller());
xzl committed
217
    weak_ptr<RtmpPusher> weakSelf = dynamic_pointer_cast<RtmpPusher>(shared_from_this());
218
    _pRtmpReader->setReadCB([weakSelf](const RtmpPacket::Ptr &pkt){
xzl committed
219 220 221 222
    	auto strongSelf = weakSelf.lock();
    	if(!strongSelf) {
    		return;
    	}
223
    	strongSelf->sendRtmp(pkt->typeId, strongSelf->_ui32StreamId, pkt, pkt->timeStamp, pkt->chunkId);
xzl committed
224
    });
225
    _pRtmpReader->setDetachCB([weakSelf](){
xzl committed
226 227
        auto strongSelf = weakSelf.lock();
        if(strongSelf){
xiongziliang committed
228
            strongSelf->onPublishResult(SockException(Err_other,"媒体源被释放"));
xzl committed
229 230
        }
    });
xiongziliang committed
231
    onPublishResult(SockException(Err_success,"success"));
232 233 234
	//提高发送性能
	(*this) << SocketFlags(kSockFlags);
	SockUtil::setNoDelay(_sock->rawFD(),false);
xzl committed
235 236 237
}
void RtmpPusher::onCmd_result(AMFDecoder &dec){
	auto iReqId = dec.load<int>();
238 239 240
	lock_guard<recursive_mutex> lck(_mtxOnResultCB);
	auto it = _mapOnResultCB.find(iReqId);
	if(it != _mapOnResultCB.end()){
xzl committed
241
		it->second(dec);
242
		_mapOnResultCB.erase(it);
xzl committed
243 244 245 246 247 248 249 250 251 252 253 254 255
	}else{
		WarnL << "unhandled _result";
	}
}
void RtmpPusher::onCmd_onStatus(AMFDecoder &dec) {
	AMFValue val;
	while(true){
		val = dec.load<AMFValue>();
		if(val.type() == AMF_OBJECT){
			break;
		}
	}
	if(val.type() != AMF_OBJECT){
256
		throw std::runtime_error("onStatus:the result object was not found");
xzl committed
257 258
	}

259 260 261 262
    lock_guard<recursive_mutex> lck(_mtxOnStatusCB);
	if(_dqOnStatusCB.size()){
		_dqOnStatusCB.front()(val);
		_dqOnStatusCB.pop_front();
xzl committed
263 264 265 266 267 268 269 270 271 272 273 274 275 276 277
	}else{
		auto level = val["level"];
		auto code = val["code"].as_string();
		if(level.type() == AMF_STRING){
			if(level.as_string() != "status"){
				throw std::runtime_error(StrPrinter <<"onStatus 失败:" << level.as_string() << " " << code << endl);
			}
		}
    }
}

void RtmpPusher::onRtmpChunk(RtmpPacket &chunkData) {
	switch (chunkData.typeId) {
		case MSG_CMD:
		case MSG_CMD3: {
278 279 280 281 282 283 284 285
			typedef void (RtmpPusher::*rtmpCMDHandle)(AMFDecoder &dec);
			static unordered_map<string, rtmpCMDHandle> g_mapCmd;
			static onceToken token([]() {
				g_mapCmd.emplace("_error",&RtmpPusher::onCmd_result);
				g_mapCmd.emplace("_result",&RtmpPusher::onCmd_result);
				g_mapCmd.emplace("onStatus",&RtmpPusher::onCmd_onStatus);
			}, []() {});

xzl committed
286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303
			AMFDecoder dec(chunkData.strBuf, 0);
			std::string type = dec.load<std::string>();
			auto it = g_mapCmd.find(type);
			if(it != g_mapCmd.end()){
				auto fun = it->second;
				(this->*fun)(dec);
			}else{
				WarnL << "can not support cmd:" << type;
			}
		}
			break;
		default:
			//WarnL << "unhandled message:" << (int) chunkData.typeId << hexdump(chunkData.strBuf.data(), chunkData.strBuf.size());
			break;
		}
}


xiongziliang committed
304
} /* namespace mediakit */
xzl committed
305