external/srs
1
0
Fork 0
mirror of https://github.com/ossrs/srs.git synced 2025-03-09 15:49:59 +00:00
Code Issues Releases Wiki Activity

refine RTMP protocol completed, to 0.9.81

Browse source
This commit is contained in:
winlin 2014-04-29 14:44:07 +08:00
parent be4c182d11
commit 80ddddf823
20 changed files with 343 additions and 353 deletions
Download patch file Download diff file Expand all files Collapse all files

45
trunk/src/app/srs_app_bandwidth.cpp
View file

@ -189,21 +189,21 @@ int SrsBandwidth::do_bandwidth_check()
pkt->data->set("publish_bytes", SrsAmf0Any::number(publish_bytes)); pkt->data->set("publish_bytes", SrsAmf0Any::number(publish_bytes));
pkt->data->set("publish_time", SrsAmf0Any::number(publish_actual_duration_ms)); pkt->data->set("publish_time", SrsAmf0Any::number(publish_actual_duration_ms));
if ((ret = rtmp->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = rtmp->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send bandwidth check finish message failed. ret=%d", ret); srs_error("send bandwidth check finish message failed. ret=%d", ret);
return ret; return ret;
} }
// if flash, we notice the result, and expect a final packet. // if flash, we notice the result, and expect a final packet.
while (true) { while (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsBandwidthPacket* pkt = NULL; SrsBandwidthPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
// info level to ignore and return success. // info level to ignore and return success.
srs_info("expect final message failed. ret=%d", ret); srs_info("expect final message failed. ret=%d", ret);
return ERROR_SUCCESS; return ERROR_SUCCESS;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsBandwidthPacket, pkt, false); SrsAutoFree(SrsBandwidthPacket, pkt, false);
srs_info("get final message success."); srs_info("get final message success.");
@ -233,7 +233,7 @@ int SrsBandwidth::check_play(
pkt->data->set("duration_ms", SrsAmf0Any::number(duration_ms)); pkt->data->set("duration_ms", SrsAmf0Any::number(duration_ms));
pkt->data->set("interval_ms", SrsAmf0Any::number(interval_ms)); pkt->data->set("interval_ms", SrsAmf0Any::number(interval_ms));
if ((ret = rtmp->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = rtmp->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send bandwidth check start play message failed. ret=%d", ret); srs_error("send bandwidth check start play message failed. ret=%d", ret);
return ret; return ret;
} }
@ -242,13 +242,13 @@ int SrsBandwidth::check_play(
while (true) { while (true) {
// recv client's starting play response // recv client's starting play response
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsBandwidthPacket* pkt = NULL; SrsBandwidthPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
srs_error("expect bandwidth message failed. ret=%d", ret); srs_error("expect bandwidth message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsBandwidthPacket, pkt, false); SrsAutoFree(SrsBandwidthPacket, pkt, false);
srs_info("get bandwidth message succes."); srs_info("get bandwidth message succes.");
@ -281,10 +281,10 @@ int SrsBandwidth::check_play(
} }
data_count += 2; data_count += 2;
// TODO: FIXME: get length from the rtmp protocol stack. // get length from the rtmp protocol stack.
play_bytes += pkt->get_payload_length(); play_bytes = rtmp->get_send_bytes();
if ((ret = rtmp->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = rtmp->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send bandwidth check play messages failed. ret=%d", ret); srs_error("send bandwidth check play messages failed. ret=%d", ret);
return ret; return ret;
} }
@ -314,7 +314,7 @@ int SrsBandwidth::check_play(
pkt->data->set("duration_delta", SrsAmf0Any::number(actual_duration_ms)); pkt->data->set("duration_delta", SrsAmf0Any::number(actual_duration_ms));
pkt->data->set("bytes_delta", SrsAmf0Any::number(play_bytes)); pkt->data->set("bytes_delta", SrsAmf0Any::number(play_bytes));
if ((ret = rtmp->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = rtmp->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send bandwidth check stop play message failed. ret=%d", ret); srs_error("send bandwidth check stop play message failed. ret=%d", ret);
return ret; return ret;
} }
@ -323,13 +323,13 @@ int SrsBandwidth::check_play(
while (true) { while (true) {
// recv client's stop play response. // recv client's stop play response.
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsBandwidthPacket* pkt = NULL; SrsBandwidthPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
srs_error("expect bandwidth message failed. ret=%d", ret); srs_error("expect bandwidth message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsBandwidthPacket, pkt, false); SrsAutoFree(SrsBandwidthPacket, pkt, false);
srs_info("get bandwidth message succes."); srs_info("get bandwidth message succes.");
@ -357,7 +357,7 @@ int SrsBandwidth::check_publish(
pkt->data->set("duration_ms", SrsAmf0Any::number(duration_ms)); pkt->data->set("duration_ms", SrsAmf0Any::number(duration_ms));
pkt->data->set("interval_ms", SrsAmf0Any::number(interval_ms)); pkt->data->set("interval_ms", SrsAmf0Any::number(interval_ms));
if ((ret = rtmp->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = rtmp->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send bandwidth check start publish message failed. ret=%d", ret); srs_error("send bandwidth check start publish message failed. ret=%d", ret);
return ret; return ret;
} }
@ -366,13 +366,13 @@ int SrsBandwidth::check_publish(
while (true) { while (true) {
// read client's notification of starting publish // read client's notification of starting publish
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsBandwidthPacket* pkt = NULL; SrsBandwidthPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
srs_error("expect bandwidth message failed. ret=%d", ret); srs_error("expect bandwidth message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsBandwidthPacket, pkt, false); SrsAutoFree(SrsBandwidthPacket, pkt, false);
srs_info("get bandwidth message succes."); srs_info("get bandwidth message succes.");
@ -387,15 +387,14 @@ int SrsBandwidth::check_publish(
while ( (srs_get_system_time_ms() - current_time) < duration_ms ) { while ( (srs_get_system_time_ms() - current_time) < duration_ms ) {
st_usleep(0); st_usleep(0);
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
if ((ret = rtmp->__recv_message(&msg)) != ERROR_SUCCESS) { if ((ret = rtmp->recv_message(&msg)) != ERROR_SUCCESS) {
srs_error("recv message failed. ret=%d", ret); srs_error("recv message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
// TODO: FIXME. publish_bytes = rtmp->get_recv_bytes();
publish_bytes += msg->header.payload_length;
int kbps = 0; int kbps = 0;
while (true) { while (true) {
@ -420,7 +419,7 @@ int SrsBandwidth::check_publish(
pkt->data->set("duration_delta", SrsAmf0Any::number(actual_duration_ms)); pkt->data->set("duration_delta", SrsAmf0Any::number(actual_duration_ms));
pkt->data->set("bytes_delta", SrsAmf0Any::number(publish_bytes)); pkt->data->set("bytes_delta", SrsAmf0Any::number(publish_bytes));
if ((ret = rtmp->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = rtmp->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send bandwidth check stop publish message failed. ret=%d", ret); srs_error("send bandwidth check stop publish message failed. ret=%d", ret);
return ret; return ret;
} }
@ -435,13 +434,13 @@ int SrsBandwidth::check_publish(
// TODO: FIXME: check whether flash client. // TODO: FIXME: check whether flash client.
while (false) { while (false) {
// recv client's stop publish response. // recv client's stop publish response.
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsBandwidthPacket* pkt = NULL; SrsBandwidthPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
srs_error("expect bandwidth message failed. ret=%d", ret); srs_error("expect bandwidth message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsBandwidthPacket, pkt, false); SrsAutoFree(SrsBandwidthPacket, pkt, false);
srs_info("get bandwidth message succes."); srs_info("get bandwidth message succes.");

18
trunk/src/app/srs_app_dvr.cpp
View file

@ -455,7 +455,7 @@ int SrsDvrPlan::on_meta_data(SrsOnMetaDataPacket* metadata)
return ret; return ret;
} }
int SrsDvrPlan::on_audio(__SrsSharedPtrMessage* audio) int SrsDvrPlan::on_audio(SrsSharedPtrMessage* audio)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -481,7 +481,7 @@ int SrsDvrPlan::on_audio(__SrsSharedPtrMessage* audio)
return ret; return ret;
} }
int SrsDvrPlan::on_video(__SrsSharedPtrMessage* video) int SrsDvrPlan::on_video(SrsSharedPtrMessage* video)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -573,7 +573,7 @@ int SrsDvrPlan::flv_close()
return ret; return ret;
} }
int SrsDvrPlan::update_duration(__SrsSharedPtrMessage* msg) int SrsDvrPlan::update_duration(SrsSharedPtrMessage* msg)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -720,7 +720,7 @@ void SrsDvrSegmentPlan::on_unpublish()
dvr_enabled = false; dvr_enabled = false;
} }
int SrsDvrSegmentPlan::update_duration(__SrsSharedPtrMessage* msg) int SrsDvrSegmentPlan::update_duration(SrsSharedPtrMessage* msg)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -903,7 +903,7 @@ int SrsDvrHssPlan::on_dvr_reap_flv_header(string path)
return ret; return ret;
} }
int SrsDvrHssPlan::update_duration(__SrsSharedPtrMessage* msg) int SrsDvrHssPlan::update_duration(SrsSharedPtrMessage* msg)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -998,11 +998,11 @@ int SrsDvr::on_meta_data(SrsOnMetaDataPacket* metadata)
return ret; return ret;
} }
int SrsDvr::on_audio(__SrsSharedPtrMessage* audio) int SrsDvr::on_audio(SrsSharedPtrMessage* audio)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
SrsAutoFree(__SrsSharedPtrMessage, audio, false); SrsAutoFree(SrsSharedPtrMessage, audio, false);
if ((ret = plan->on_audio(audio)) != ERROR_SUCCESS) { if ((ret = plan->on_audio(audio)) != ERROR_SUCCESS) {
return ret; return ret;
@ -1011,11 +1011,11 @@ int SrsDvr::on_audio(__SrsSharedPtrMessage* audio)
return ret; return ret;
} }
int SrsDvr::on_video(__SrsSharedPtrMessage* video) int SrsDvr::on_video(SrsSharedPtrMessage* video)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
SrsAutoFree(__SrsSharedPtrMessage, video, false); SrsAutoFree(SrsSharedPtrMessage, video, false);
if ((ret = plan->on_video(video)) != ERROR_SUCCESS) { if ((ret = plan->on_video(video)) != ERROR_SUCCESS) {
return ret; return ret;

16
trunk/src/app/srs_app_dvr.hpp
View file

@ -36,7 +36,7 @@ class SrsRequest;
class SrsStream; class SrsStream;
class SrsRtmpJitter; class SrsRtmpJitter;
class SrsOnMetaDataPacket; class SrsOnMetaDataPacket;
class __SrsSharedPtrMessage; class SrsSharedPtrMessage;
/** /**
* file stream to read/write file. * file stream to read/write file.
@ -185,13 +185,13 @@ public:
virtual int on_publish(); virtual int on_publish();
virtual void on_unpublish() = 0; virtual void on_unpublish() = 0;
virtual int on_meta_data(SrsOnMetaDataPacket* metadata); virtual int on_meta_data(SrsOnMetaDataPacket* metadata);
virtual int on_audio(__SrsSharedPtrMessage* audio); virtual int on_audio(SrsSharedPtrMessage* audio);
virtual int on_video(__SrsSharedPtrMessage* video); virtual int on_video(SrsSharedPtrMessage* video);
protected: protected:
virtual int flv_open(std::string stream, std::string path); virtual int flv_open(std::string stream, std::string path);
virtual int flv_close(); virtual int flv_close();
virtual int open_new_segment(); virtual int open_new_segment();
virtual int update_duration(__SrsSharedPtrMessage* msg); virtual int update_duration(SrsSharedPtrMessage* msg);
virtual int write_flv_header(); virtual int write_flv_header();
virtual int on_dvr_request_sh(); virtual int on_dvr_request_sh();
virtual int on_video_keyframe(); virtual int on_video_keyframe();
@ -233,7 +233,7 @@ public:
virtual int on_publish(); virtual int on_publish();
virtual void on_unpublish(); virtual void on_unpublish();
private: private:
virtual int update_duration(__SrsSharedPtrMessage* msg); virtual int update_duration(SrsSharedPtrMessage* msg);
}; };
/** /**
@ -260,7 +260,7 @@ protected:
virtual int64_t filter_timestamp(int64_t timestamp); virtual int64_t filter_timestamp(int64_t timestamp);
private: private:
virtual int on_dvr_reap_flv_header(std::string path); virtual int on_dvr_reap_flv_header(std::string path);
virtual int update_duration(__SrsSharedPtrMessage* msg); virtual int update_duration(SrsSharedPtrMessage* msg);
}; };
/** /**
@ -300,11 +300,11 @@ public:
/** /**
* mux the audio packets to dvr. * mux the audio packets to dvr.
*/ */
virtual int on_audio(__SrsSharedPtrMessage* audio); virtual int on_audio(SrsSharedPtrMessage* audio);
/** /**
* mux the video packets to dvr. * mux the video packets to dvr.
*/ */
virtual int on_video(__SrsSharedPtrMessage* video); virtual int on_video(SrsSharedPtrMessage* video);
}; };
#endif #endif

30
trunk/src/app/srs_app_edge.cpp
View file

@ -175,15 +175,15 @@ int SrsEdgeIngester::ingest()
} }
// read from client. // read from client.
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
if ((ret = client->__recv_message(&msg)) != ERROR_SUCCESS) { if ((ret = client->recv_message(&msg)) != ERROR_SUCCESS) {
srs_error("ingest recv origin server message failed. ret=%d", ret); srs_error("ingest recv origin server message failed. ret=%d", ret);
return ret; return ret;
} }
srs_verbose("edge loop recv message. ret=%d", ret); srs_verbose("edge loop recv message. ret=%d", ret);
srs_assert(msg); srs_assert(msg);
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
if ((ret = process_publish_message(msg)) != ERROR_SUCCESS) { if ((ret = process_publish_message(msg)) != ERROR_SUCCESS) {
return ret; return ret;
@ -193,7 +193,7 @@ int SrsEdgeIngester::ingest()
return ret; return ret;
} }
int SrsEdgeIngester::process_publish_message(__SrsMessage* msg) int SrsEdgeIngester::process_publish_message(SrsMessage* msg)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -218,7 +218,7 @@ int SrsEdgeIngester::process_publish_message(__SrsMessage* msg)
// process onMetaData // process onMetaData
if (msg->header.is_amf0_data() || msg->header.is_amf3_data()) { if (msg->header.is_amf0_data() || msg->header.is_amf3_data()) {
SrsPacket* pkt = NULL; SrsPacket* pkt = NULL;
if ((ret = client->__decode_message(msg, &pkt)) != ERROR_SUCCESS) { if ((ret = client->decode_message(msg, &pkt)) != ERROR_SUCCESS) {
srs_error("decode onMetaData message failed. ret=%d", ret); srs_error("decode onMetaData message failed. ret=%d", ret);
return ret; return ret;
} }
@ -420,8 +420,8 @@ int SrsEdgeForwarder::cycle()
// read from client. // read from client.
if (true) { if (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
ret = client->__recv_message(&msg); ret = client->recv_message(&msg);
srs_verbose("edge loop recv message. ret=%d", ret); srs_verbose("edge loop recv message. ret=%d", ret);
if (ret != ERROR_SUCCESS && ret != ERROR_SOCKET_TIMEOUT) { if (ret != ERROR_SUCCESS && ret != ERROR_SOCKET_TIMEOUT) {
@ -435,7 +435,7 @@ int SrsEdgeForwarder::cycle()
// forward all messages. // forward all messages.
int count = 0; int count = 0;
__SrsSharedPtrMessage** msgs = NULL; SrsSharedPtrMessage** msgs = NULL;
if ((ret = queue->get_packets(0, msgs, count)) != ERROR_SUCCESS) { if ((ret = queue->get_packets(0, msgs, count)) != ERROR_SUCCESS) {
srs_error("get message to forward to origin failed. ret=%d", ret); srs_error("get message to forward to origin failed. ret=%d", ret);
return ret; return ret;
@ -456,16 +456,16 @@ int SrsEdgeForwarder::cycle()
srs_verbose("no packets to forward."); srs_verbose("no packets to forward.");
continue; continue;
} }
SrsAutoFree(__SrsSharedPtrMessage*, msgs, true); SrsAutoFree(SrsSharedPtrMessage*, msgs, true);
// all msgs to forward. // all msgs to forward.
for (int i = 0; i < count; i++) { for (int i = 0; i < count; i++) {
__SrsSharedPtrMessage* msg = msgs[i]; SrsSharedPtrMessage* msg = msgs[i];
srs_assert(msg); srs_assert(msg);
msgs[i] = NULL; msgs[i] = NULL;
if ((ret = client->__send_and_free_message(msg)) != ERROR_SUCCESS) { if ((ret = client->send_and_free_message(msg)) != ERROR_SUCCESS) {
srs_error("edge publish forwarder send message to server failed. ret=%d", ret); srs_error("edge publish forwarder send message to server failed. ret=%d", ret);
return ret; return ret;
} }
@ -475,7 +475,7 @@ int SrsEdgeForwarder::cycle()
return ret; return ret;
} }
int SrsEdgeForwarder::proxy(__SrsMessage* msg) int SrsEdgeForwarder::proxy(SrsMessage* msg)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -495,8 +495,8 @@ int SrsEdgeForwarder::proxy(__SrsMessage* msg)
} }
// TODO: FIXME: use utility to copy msg to shared ptr msg. // TODO: FIXME: use utility to copy msg to shared ptr msg.
__SrsSharedPtrMessage* copy = new __SrsSharedPtrMessage(); SrsSharedPtrMessage* copy = new SrsSharedPtrMessage();
SrsAutoFree(__SrsSharedPtrMessage, copy, false); SrsAutoFree(SrsSharedPtrMessage, copy, false);
if ((ret = copy->initialize(msg)) != ERROR_SUCCESS) { if ((ret = copy->initialize(msg)) != ERROR_SUCCESS) {
srs_error("initialize the msg failed. ret=%d", ret); srs_error("initialize the msg failed. ret=%d", ret);
return ret; return ret;
@ -724,7 +724,7 @@ int SrsPublishEdge::on_client_publish()
return ret; return ret;
} }
int SrsPublishEdge::on_proxy_publish(__SrsMessage* msg) int SrsPublishEdge::on_proxy_publish(SrsMessage* msg)
{ {
return forwarder->proxy(msg); return forwarder->proxy(msg);
} }

8
trunk/src/app/srs_app_edge.hpp
View file

@ -40,7 +40,7 @@ class SrsRequest;
class SrsPlayEdge; class SrsPlayEdge;
class SrsPublishEdge; class SrsPublishEdge;
class SrsRtmpClient; class SrsRtmpClient;
class __SrsMessage; class SrsMessage;
class SrsMessageQueue; class SrsMessageQueue;
class ISrsProtocolReaderWriter; class ISrsProtocolReaderWriter;
@ -99,7 +99,7 @@ private:
virtual int ingest(); virtual int ingest();
virtual void close_underlayer_socket(); virtual void close_underlayer_socket();
virtual int connect_server(); virtual int connect_server();
virtual int process_publish_message(__SrsMessage* msg); virtual int process_publish_message(SrsMessage* msg);
}; };
/** /**
@ -142,7 +142,7 @@ public:
public: public:
virtual int cycle(); virtual int cycle();
public: public:
virtual int proxy(__SrsMessage* msg); virtual int proxy(SrsMessage* msg);
private: private:
virtual void close_underlayer_socket(); virtual void close_underlayer_socket();
virtual int connect_server(); virtual int connect_server();
@ -202,7 +202,7 @@ public:
/** /**
* proxy publish stream to edge * proxy publish stream to edge
*/ */
virtual int on_proxy_publish(__SrsMessage* msg); virtual int on_proxy_publish(SrsMessage* msg);
/** /**
* proxy unpublish stream to edge. * proxy unpublish stream to edge.
*/ */

18
trunk/src/app/srs_app_forward.cpp
View file

@ -148,7 +148,7 @@ void SrsForwarder::on_unpublish()
srs_freep(io); srs_freep(io);
} }
int SrsForwarder::on_meta_data(__SrsSharedPtrMessage* metadata) int SrsForwarder::on_meta_data(SrsSharedPtrMessage* metadata)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -164,7 +164,7 @@ int SrsForwarder::on_meta_data(__SrsSharedPtrMessage* metadata)
return ret; return ret;
} }
int SrsForwarder::on_audio(__SrsSharedPtrMessage* msg) int SrsForwarder::on_audio(SrsSharedPtrMessage* msg)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -180,7 +180,7 @@ int SrsForwarder::on_audio(__SrsSharedPtrMessage* msg)
return ret; return ret;
} }
int SrsForwarder::on_video(__SrsSharedPtrMessage* msg) int SrsForwarder::on_video(SrsSharedPtrMessage* msg)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -315,8 +315,8 @@ int SrsForwarder::forward()
// read from client. // read from client.
if (true) { if (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
ret = client->__recv_message(&msg); ret = client->recv_message(&msg);
srs_verbose("play loop recv message. ret=%d", ret); srs_verbose("play loop recv message. ret=%d", ret);
if (ret != ERROR_SUCCESS && ret != ERROR_SOCKET_TIMEOUT) { if (ret != ERROR_SUCCESS && ret != ERROR_SOCKET_TIMEOUT) {
@ -329,7 +329,7 @@ int SrsForwarder::forward()
// forward all messages. // forward all messages.
int count = 0; int count = 0;
__SrsSharedPtrMessage** msgs = NULL; SrsSharedPtrMessage** msgs = NULL;
if ((ret = queue->get_packets(0, msgs, count)) != ERROR_SUCCESS) { if ((ret = queue->get_packets(0, msgs, count)) != ERROR_SUCCESS) {
srs_error("get message to forward failed. ret=%d", ret); srs_error("get message to forward failed. ret=%d", ret);
return ret; return ret;
@ -348,16 +348,16 @@ int SrsForwarder::forward()
srs_verbose("no packets to forward."); srs_verbose("no packets to forward.");
continue; continue;
} }
SrsAutoFree(__SrsSharedPtrMessage*, msgs, true); SrsAutoFree(SrsSharedPtrMessage*, msgs, true);
// all msgs to forward. // all msgs to forward.
for (int i = 0; i < count; i++) { for (int i = 0; i < count; i++) {
__SrsSharedPtrMessage* msg = msgs[i]; SrsSharedPtrMessage* msg = msgs[i];
srs_assert(msg); srs_assert(msg);
msgs[i] = NULL; msgs[i] = NULL;
if ((ret = client->__send_and_free_message(msg)) != ERROR_SUCCESS) { if ((ret = client->send_and_free_message(msg)) != ERROR_SUCCESS) {
srs_error("forwarder send message to server failed. ret=%d", ret); srs_error("forwarder send message to server failed. ret=%d", ret);
return ret; return ret;
} }

8
trunk/src/app/srs_app_forward.hpp
View file

@ -35,7 +35,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <srs_app_thread.hpp> #include <srs_app_thread.hpp>
class ISrsProtocolReaderWriter; class ISrsProtocolReaderWriter;
class __SrsSharedPtrMessage; class SrsSharedPtrMessage;
class SrsOnMetaDataPacket; class SrsOnMetaDataPacket;
class SrsMessageQueue; class SrsMessageQueue;
class SrsRtmpJitter; class SrsRtmpJitter;
@ -72,9 +72,9 @@ public:
public: public:
virtual int on_publish(SrsRequest* req, std::string forward_server); virtual int on_publish(SrsRequest* req, std::string forward_server);
virtual void on_unpublish(); virtual void on_unpublish();
virtual int on_meta_data(__SrsSharedPtrMessage* metadata); virtual int on_meta_data(SrsSharedPtrMessage* metadata);
virtual int on_audio(__SrsSharedPtrMessage* msg); virtual int on_audio(SrsSharedPtrMessage* msg);
virtual int on_video(__SrsSharedPtrMessage* msg); virtual int on_video(SrsSharedPtrMessage* msg);
// interface ISrsThreadHandler. // interface ISrsThreadHandler.
public: public:
virtual int cycle(); virtual int cycle();

8
trunk/src/app/srs_app_hls.cpp
View file

@ -1379,11 +1379,11 @@ int SrsHls::on_meta_data(SrsAmf0Object* metadata)
return ret; return ret;
} }
int SrsHls::on_audio(__SrsSharedPtrMessage* audio) int SrsHls::on_audio(SrsSharedPtrMessage* audio)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
SrsAutoFree(__SrsSharedPtrMessage, audio, false); SrsAutoFree(SrsSharedPtrMessage, audio, false);
if (!hls_enabled) { if (!hls_enabled) {
return ret; return ret;
@ -1423,11 +1423,11 @@ int SrsHls::on_audio(__SrsSharedPtrMessage* audio)
return ret; return ret;
} }
int SrsHls::on_video(__SrsSharedPtrMessage* video) int SrsHls::on_video(SrsSharedPtrMessage* video)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
SrsAutoFree(__SrsSharedPtrMessage, video, false); SrsAutoFree(SrsSharedPtrMessage, video, false);
if (!hls_enabled) { if (!hls_enabled) {
return ret; return ret;

6
trunk/src/app/srs_app_hls.hpp
View file

@ -34,7 +34,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <string> #include <string>
#include <vector> #include <vector>
class __SrsSharedPtrMessage; class SrsSharedPtrMessage;
class SrsCodecSample; class SrsCodecSample;
class SrsCodecBuffer; class SrsCodecBuffer;
class SrsMpegtsFrame; class SrsMpegtsFrame;
@ -314,11 +314,11 @@ public:
/** /**
* mux the audio packets to ts. * mux the audio packets to ts.
*/ */
virtual int on_audio(__SrsSharedPtrMessage* audio); virtual int on_audio(SrsSharedPtrMessage* audio);
/** /**
* mux the video packets to ts. * mux the video packets to ts.
*/ */
virtual int on_video(__SrsSharedPtrMessage* video); virtual int on_video(SrsSharedPtrMessage* video);
private: private:
virtual void hls_mux(); virtual void hls_mux();
}; };

38
trunk/src/app/srs_app_rtmp_conn.cpp
View file

@ -472,8 +472,8 @@ int SrsRtmpConn::playing(SrsSource* source)
// read from client. // read from client.
int ctl_msg_ret = ERROR_SUCCESS; int ctl_msg_ret = ERROR_SUCCESS;
if (true) { if (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
ctl_msg_ret = ret = rtmp->__recv_message(&msg); ctl_msg_ret = ret = rtmp->recv_message(&msg);
srs_verbose("play loop recv message. ret=%d", ret); srs_verbose("play loop recv message. ret=%d", ret);
if (ret != ERROR_SUCCESS && ret != ERROR_SOCKET_TIMEOUT) { if (ret != ERROR_SUCCESS && ret != ERROR_SOCKET_TIMEOUT) {
@ -491,7 +491,7 @@ int SrsRtmpConn::playing(SrsSource* source)
} }
// get messages from consumer. // get messages from consumer.
__SrsSharedPtrMessage** msgs = NULL; SrsSharedPtrMessage** msgs = NULL;
int count = 0; int count = 0;
if ((ret = consumer->get_packets(0, msgs, count)) != ERROR_SUCCESS) { if ((ret = consumer->get_packets(0, msgs, count)) != ERROR_SUCCESS) {
srs_error("get messages from consumer failed. ret=%d", ret); srs_error("get messages from consumer failed. ret=%d", ret);
@ -510,11 +510,11 @@ int SrsRtmpConn::playing(SrsSource* source)
srs_verbose("no packets in queue."); srs_verbose("no packets in queue.");
continue; continue;
} }
SrsAutoFree(__SrsSharedPtrMessage*, msgs, true); SrsAutoFree(SrsSharedPtrMessage*, msgs, true);
// sendout messages // sendout messages
for (int i = 0; i < count; i++) { for (int i = 0; i < count; i++) {
__SrsSharedPtrMessage* msg = msgs[i]; SrsSharedPtrMessage* msg = msgs[i];
// the send_message will free the msg, // the send_message will free the msg,
// so set the msgs[i] to NULL. // so set the msgs[i] to NULL.
@ -530,7 +530,7 @@ int SrsRtmpConn::playing(SrsSource* source)
duration += msg->header.timestamp - starttime; duration += msg->header.timestamp - starttime;
starttime = msg->header.timestamp; starttime = msg->header.timestamp;
if ((ret = rtmp->__send_and_free_message(msg)) != ERROR_SUCCESS) { if ((ret = rtmp->send_and_free_message(msg)) != ERROR_SUCCESS) {
srs_error("send message to client failed. ret=%d", ret); srs_error("send message to client failed. ret=%d", ret);
return ret; return ret;
} }
@ -573,13 +573,13 @@ int SrsRtmpConn::fmle_publish(SrsSource* source)
// switch to other st-threads. // switch to other st-threads.
st_usleep(0); st_usleep(0);
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
if ((ret = rtmp->__recv_message(&msg)) != ERROR_SUCCESS) { if ((ret = rtmp->recv_message(&msg)) != ERROR_SUCCESS) {
srs_error("fmle recv identify client message failed. ret=%d", ret); srs_error("fmle recv identify client message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
pithy_print.elapse(); pithy_print.elapse();
@ -594,7 +594,7 @@ int SrsRtmpConn::fmle_publish(SrsSource* source)
// process UnPublish event. // process UnPublish event.
if (msg->header.is_amf0_command() || msg->header.is_amf3_command()) { if (msg->header.is_amf0_command() || msg->header.is_amf3_command()) {
SrsPacket* pkt = NULL; SrsPacket* pkt = NULL;
if ((ret = rtmp->__decode_message(msg, &pkt)) != ERROR_SUCCESS) { if ((ret = rtmp->decode_message(msg, &pkt)) != ERROR_SUCCESS) {
srs_error("fmle decode unpublish message failed. ret=%d", ret); srs_error("fmle decode unpublish message failed. ret=%d", ret);
return ret; return ret;
} }
@ -648,15 +648,15 @@ int SrsRtmpConn::flash_publish(SrsSource* source)
// switch to other st-threads. // switch to other st-threads.
st_usleep(0); st_usleep(0);
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
if ((ret = rtmp->__recv_message(&msg)) != ERROR_SUCCESS) { if ((ret = rtmp->recv_message(&msg)) != ERROR_SUCCESS) {
if (!srs_is_client_gracefully_close(ret)) { if (!srs_is_client_gracefully_close(ret)) {
srs_error("flash recv identify client message failed. ret=%d", ret); srs_error("flash recv identify client message failed. ret=%d", ret);
} }
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
pithy_print.elapse(); pithy_print.elapse();
@ -671,7 +671,7 @@ int SrsRtmpConn::flash_publish(SrsSource* source)
// process UnPublish event. // process UnPublish event.
if (msg->header.is_amf0_command() || msg->header.is_amf3_command()) { if (msg->header.is_amf0_command() || msg->header.is_amf3_command()) {
SrsPacket* pkt = NULL; SrsPacket* pkt = NULL;
if ((ret = rtmp->__decode_message(msg, &pkt)) != ERROR_SUCCESS) { if ((ret = rtmp->decode_message(msg, &pkt)) != ERROR_SUCCESS) {
srs_error("flash decode unpublish message failed. ret=%d", ret); srs_error("flash decode unpublish message failed. ret=%d", ret);
return ret; return ret;
} }
@ -694,7 +694,7 @@ int SrsRtmpConn::flash_publish(SrsSource* source)
return ret; return ret;
} }
int SrsRtmpConn::process_publish_message(SrsSource* source, __SrsMessage* msg, bool vhost_is_edge) int SrsRtmpConn::process_publish_message(SrsSource* source, SrsMessage* msg, bool vhost_is_edge)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -727,7 +727,7 @@ int SrsRtmpConn::process_publish_message(SrsSource* source, __SrsMessage* msg, b
// process onMetaData // process onMetaData
if (msg->header.is_amf0_data() || msg->header.is_amf3_data()) { if (msg->header.is_amf0_data() || msg->header.is_amf3_data()) {
SrsPacket* pkt = NULL; SrsPacket* pkt = NULL;
if ((ret = rtmp->__decode_message(msg, &pkt)) != ERROR_SUCCESS) { if ((ret = rtmp->decode_message(msg, &pkt)) != ERROR_SUCCESS) {
srs_error("decode onMetaData message failed. ret=%d", ret); srs_error("decode onMetaData message failed. ret=%d", ret);
return ret; return ret;
} }
@ -750,7 +750,7 @@ int SrsRtmpConn::process_publish_message(SrsSource* source, __SrsMessage* msg, b
return ret; return ret;
} }
int SrsRtmpConn::process_play_control_msg(SrsConsumer* consumer, __SrsMessage* msg) int SrsRtmpConn::process_play_control_msg(SrsConsumer* consumer, SrsMessage* msg)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -758,7 +758,7 @@ int SrsRtmpConn::process_play_control_msg(SrsConsumer* consumer, __SrsMessage* m
srs_verbose("ignore all empty message."); srs_verbose("ignore all empty message.");
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
if (!msg->header.is_amf0_command() && !msg->header.is_amf3_command()) { if (!msg->header.is_amf0_command() && !msg->header.is_amf3_command()) {
srs_info("ignore all message except amf0/amf3 command."); srs_info("ignore all message except amf0/amf3 command.");
@ -766,7 +766,7 @@ int SrsRtmpConn::process_play_control_msg(SrsConsumer* consumer, __SrsMessage* m
} }
SrsPacket* pkt = NULL; SrsPacket* pkt = NULL;
if ((ret = rtmp->__decode_message(msg, &pkt)) != ERROR_SUCCESS) { if ((ret = rtmp->decode_message(msg, &pkt)) != ERROR_SUCCESS) {
srs_error("decode the amf0/amf3 command packet failed. ret=%d", ret); srs_error("decode the amf0/amf3 command packet failed. ret=%d", ret);
return ret; return ret;
} }

6
trunk/src/app/srs_app_rtmp_conn.hpp
View file

@ -40,7 +40,7 @@ class SrsResponse;
class SrsSource; class SrsSource;
class SrsRefer; class SrsRefer;
class SrsConsumer; class SrsConsumer;
class __SrsMessage; class SrsMessage;
class SrsSocket; class SrsSocket;
#ifdef SRS_AUTO_HTTP_CALLBACK #ifdef SRS_AUTO_HTTP_CALLBACK
class SrsHttpHooks; class SrsHttpHooks;
@ -80,8 +80,8 @@ private:
virtual int playing(SrsSource* source); virtual int playing(SrsSource* source);
virtual int fmle_publish(SrsSource* source); virtual int fmle_publish(SrsSource* source);
virtual int flash_publish(SrsSource* source); virtual int flash_publish(SrsSource* source);
virtual int process_publish_message(SrsSource* source, __SrsMessage* msg, bool vhost_is_edge); virtual int process_publish_message(SrsSource* source, SrsMessage* msg, bool vhost_is_edge);
virtual int process_play_control_msg(SrsConsumer* consumer, __SrsMessage* msg); virtual int process_play_control_msg(SrsConsumer* consumer, SrsMessage* msg);
private: private:
virtual int http_hooks_on_connect(); virtual int http_hooks_on_connect();
virtual void http_hooks_on_close(); virtual void http_hooks_on_close();

65
trunk/src/app/srs_app_source.cpp
View file

@ -52,7 +52,7 @@ SrsRtmpJitter::~SrsRtmpJitter()
{ {
} }
int SrsRtmpJitter::correct(__SrsSharedPtrMessage* msg, int tba, int tbv) int SrsRtmpJitter::correct(SrsSharedPtrMessage* msg, int tba, int tbv)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -130,7 +130,7 @@ void SrsMessageQueue::set_queue_size(double queue_size)
queue_size_ms = (int)(queue_size * 1000); queue_size_ms = (int)(queue_size * 1000);
} }
int SrsMessageQueue::enqueue(__SrsSharedPtrMessage* msg) int SrsMessageQueue::enqueue(SrsSharedPtrMessage* msg)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -151,7 +151,7 @@ int SrsMessageQueue::enqueue(__SrsSharedPtrMessage* msg)
return ret; return ret;
} }
int SrsMessageQueue::get_packets(int max_count, __SrsSharedPtrMessage**& pmsgs, int& count) int SrsMessageQueue::get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& count)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -169,13 +169,13 @@ int SrsMessageQueue::get_packets(int max_count, __SrsSharedPtrMessage**& pmsgs,
return ret; return ret;
} }
pmsgs = new __SrsSharedPtrMessage*[count]; pmsgs = new SrsSharedPtrMessage*[count];
for (int i = 0; i < count; i++) { for (int i = 0; i < count; i++) {
pmsgs[i] = msgs[i]; pmsgs[i] = msgs[i];
} }
__SrsSharedPtrMessage* last = msgs[count - 1]; SrsSharedPtrMessage* last = msgs[count - 1];
av_start_time = last->header.timestamp; av_start_time = last->header.timestamp;
if (count == (int)msgs.size()) { if (count == (int)msgs.size()) {
@ -196,7 +196,7 @@ void SrsMessageQueue::shrink()
// for when we shrinked, the first is the iframe, // for when we shrinked, the first is the iframe,
// we will directly remove the gop next time. // we will directly remove the gop next time.
for (int i = 1; i < (int)msgs.size(); i++) { for (int i = 1; i < (int)msgs.size(); i++) {
__SrsSharedPtrMessage* msg = msgs[i]; SrsSharedPtrMessage* msg = msgs[i];
if (msg->header.is_video()) { if (msg->header.is_video()) {
if (SrsCodec::video_is_keyframe(msg->payload, msg->size)) { if (SrsCodec::video_is_keyframe(msg->payload, msg->size)) {
@ -222,7 +222,7 @@ void SrsMessageQueue::shrink()
// remove the first gop from the front // remove the first gop from the front
for (int i = 0; i < iframe_index; i++) { for (int i = 0; i < iframe_index; i++) {
__SrsSharedPtrMessage* msg = msgs[i]; SrsSharedPtrMessage* msg = msgs[i];
srs_freep(msg); srs_freep(msg);
} }
msgs.erase(msgs.begin(), msgs.begin() + iframe_index); msgs.erase(msgs.begin(), msgs.begin() + iframe_index);
@ -230,10 +230,10 @@ void SrsMessageQueue::shrink()
void SrsMessageQueue::clear() void SrsMessageQueue::clear()
{ {
std::vector<__SrsSharedPtrMessage*>::iterator it; std::vector<SrsSharedPtrMessage*>::iterator it;
for (it = msgs.begin(); it != msgs.end(); ++it) { for (it = msgs.begin(); it != msgs.end(); ++it) {
__SrsSharedPtrMessage* msg = *it; SrsSharedPtrMessage* msg = *it;
srs_freep(msg); srs_freep(msg);
} }
@ -267,7 +267,7 @@ int SrsConsumer::get_time()
return jitter->get_time(); return jitter->get_time();
} }
int SrsConsumer::enqueue(__SrsSharedPtrMessage* msg, int tba, int tbv) int SrsConsumer::enqueue(SrsSharedPtrMessage* msg, int tba, int tbv)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -285,7 +285,7 @@ int SrsConsumer::enqueue(__SrsSharedPtrMessage* msg, int tba, int tbv)
return ret; return ret;
} }
int SrsConsumer::get_packets(int max_count, __SrsSharedPtrMessage**& pmsgs, int& count) int SrsConsumer::get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& count)
{ {
// paused, return nothing. // paused, return nothing.
if (paused) { if (paused) {
@ -329,7 +329,7 @@ void SrsGopCache::set(bool enabled)
srs_info("enable gop cache"); srs_info("enable gop cache");
} }
int SrsGopCache::cache(__SrsSharedPtrMessage* msg) int SrsGopCache::cache(SrsSharedPtrMessage* msg)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -368,9 +368,9 @@ int SrsGopCache::cache(__SrsSharedPtrMessage* msg)
void SrsGopCache::clear() void SrsGopCache::clear()
{ {
std::vector<__SrsSharedPtrMessage*>::iterator it; std::vector<SrsSharedPtrMessage*>::iterator it;
for (it = gop_cache.begin(); it != gop_cache.end(); ++it) { for (it = gop_cache.begin(); it != gop_cache.end(); ++it) {
__SrsSharedPtrMessage* msg = *it; SrsSharedPtrMessage* msg = *it;
srs_freep(msg); srs_freep(msg);
} }
gop_cache.clear(); gop_cache.clear();
@ -382,9 +382,9 @@ int SrsGopCache::dump(SrsConsumer* consumer, int tba, int tbv)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
std::vector<__SrsSharedPtrMessage*>::iterator it; std::vector<SrsSharedPtrMessage*>::iterator it;
for (it = gop_cache.begin(); it != gop_cache.end(); ++it) { for (it = gop_cache.begin(); it != gop_cache.end(); ++it) {
__SrsSharedPtrMessage* msg = *it; SrsSharedPtrMessage* msg = *it;
if ((ret = consumer->enqueue(msg->copy(), tba, tbv)) != ERROR_SUCCESS) { if ((ret = consumer->enqueue(msg->copy(), tba, tbv)) != ERROR_SUCCESS) {
srs_error("dispatch cached gop failed. ret=%d", ret); srs_error("dispatch cached gop failed. ret=%d", ret);
return ret; return ret;
@ -406,7 +406,7 @@ int64_t SrsGopCache::get_start_time()
return 0; return 0;
} }
__SrsSharedPtrMessage* msg = gop_cache[0]; SrsSharedPtrMessage* msg = gop_cache[0];
srs_assert(msg); srs_assert(msg);
return msg->header.timestamp; return msg->header.timestamp;
@ -789,7 +789,7 @@ bool SrsSource::can_publish()
return _can_publish; return _can_publish;
} }
int SrsSource::on_meta_data(__SrsMessage* msg, SrsOnMetaDataPacket* metadata) int SrsSource::on_meta_data(SrsMessage* msg, SrsOnMetaDataPacket* metadata)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -823,13 +823,7 @@ int SrsSource::on_meta_data(__SrsMessage* msg, SrsOnMetaDataPacket* metadata)
} }
// encode the metadata to payload // encode the metadata to payload
int size = metadata->get_payload_length(); int size = 0;
if (size <= 0) {
srs_warn("ignore the invalid metadata. size=%d", size);
return ret;
}
srs_verbose("get metadata size success.");
char* payload = NULL; char* payload = NULL;
if ((ret = metadata->encode(size, payload)) != ERROR_SUCCESS) { if ((ret = metadata->encode(size, payload)) != ERROR_SUCCESS) {
srs_error("encode metadata error. ret=%d", ret); srs_error("encode metadata error. ret=%d", ret);
@ -838,9 +832,14 @@ int SrsSource::on_meta_data(__SrsMessage* msg, SrsOnMetaDataPacket* metadata)
} }
srs_verbose("encode metadata success."); srs_verbose("encode metadata success.");
if (size <= 0) {
srs_warn("ignore the invalid metadata. size=%d", size);
return ret;
}
// create a shared ptr message. // create a shared ptr message.
srs_freep(cache_metadata); srs_freep(cache_metadata);
cache_metadata = new __SrsSharedPtrMessage(); cache_metadata = new SrsSharedPtrMessage();
// dump message to shared ptr message. // dump message to shared ptr message.
if ((ret = cache_metadata->initialize(&msg->header, payload, size)) != ERROR_SUCCESS) { if ((ret = cache_metadata->initialize(&msg->header, payload, size)) != ERROR_SUCCESS) {
@ -877,12 +876,12 @@ int SrsSource::on_meta_data(__SrsMessage* msg, SrsOnMetaDataPacket* metadata)
return ret; return ret;
} }
int SrsSource::on_audio(__SrsMessage* audio) int SrsSource::on_audio(SrsMessage* audio)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
__SrsSharedPtrMessage* msg = new __SrsSharedPtrMessage(); SrsSharedPtrMessage* msg = new SrsSharedPtrMessage();
SrsAutoFree(__SrsSharedPtrMessage, msg, false); SrsAutoFree(SrsSharedPtrMessage, msg, false);
if ((ret = msg->initialize(audio)) != ERROR_SUCCESS) { if ((ret = msg->initialize(audio)) != ERROR_SUCCESS) {
srs_error("initialize the audio failed. ret=%d", ret); srs_error("initialize the audio failed. ret=%d", ret);
return ret; return ret;
@ -966,12 +965,12 @@ int SrsSource::on_audio(__SrsMessage* audio)
return ret; return ret;
} }
int SrsSource::on_video(__SrsMessage* video) int SrsSource::on_video(SrsMessage* video)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
__SrsSharedPtrMessage* msg = new __SrsSharedPtrMessage(); SrsSharedPtrMessage* msg = new SrsSharedPtrMessage();
SrsAutoFree(__SrsSharedPtrMessage, msg, false); SrsAutoFree(SrsSharedPtrMessage, msg, false);
if ((ret = msg->initialize(video)) != ERROR_SUCCESS) { if ((ret = msg->initialize(video)) != ERROR_SUCCESS) {
srs_error("initialize the video failed. ret=%d", ret); srs_error("initialize the video failed. ret=%d", ret);
return ret; return ret;
@ -1207,7 +1206,7 @@ int SrsSource::on_edge_start_publish()
return publish_edge->on_client_publish(); return publish_edge->on_client_publish();
} }
int SrsSource::on_edge_proxy_publish(__SrsMessage* msg) int SrsSource::on_edge_proxy_publish(SrsMessage* msg)
{ {
return publish_edge->on_proxy_publish(msg); return publish_edge->on_proxy_publish(msg);
} }

34
trunk/src/app/srs_app_source.hpp
View file

@ -40,9 +40,9 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
class SrsPlayEdge; class SrsPlayEdge;
class SrsPublishEdge; class SrsPublishEdge;
class SrsSource; class SrsSource;
class __SrsMessage; class SrsMessage;
class SrsOnMetaDataPacket; class SrsOnMetaDataPacket;
class __SrsSharedPtrMessage; class SrsSharedPtrMessage;
class SrsForwarder; class SrsForwarder;
class SrsRequest; class SrsRequest;
class SrsSocket; class SrsSocket;
@ -74,7 +74,7 @@ public:
/** /**
* detect the time jitter and correct it. * detect the time jitter and correct it.
*/ */
virtual int correct(__SrsSharedPtrMessage* msg, int tba, int tbv); virtual int correct(SrsSharedPtrMessage* msg, int tba, int tbv);
/** /**
* get current client time, the last packet time. * get current client time, the last packet time.
*/ */
@ -91,7 +91,7 @@ private:
int64_t av_start_time; int64_t av_start_time;
int64_t av_end_time; int64_t av_end_time;
int queue_size_ms; int queue_size_ms;
std::vector<__SrsSharedPtrMessage*> msgs; std::vector<SrsSharedPtrMessage*> msgs;
public: public:
SrsMessageQueue(); SrsMessageQueue();
virtual ~SrsMessageQueue(); virtual ~SrsMessageQueue();
@ -106,14 +106,14 @@ public:
* enqueue the message, the timestamp always monotonically. * enqueue the message, the timestamp always monotonically.
* @param msg, the msg to enqueue, user never free it whatever the return code. * @param msg, the msg to enqueue, user never free it whatever the return code.
*/ */
virtual int enqueue(__SrsSharedPtrMessage* msg); virtual int enqueue(SrsSharedPtrMessage* msg);
/** /**
* get packets in consumer queue. * get packets in consumer queue.
* @pmsgs SrsMessages*[], output the prt array. * @pmsgs SrsMessages*[], output the prt array.
* @count the count in array. * @count the count in array.
* @max_count the max count to dequeue, 0 to dequeue all. * @max_count the max count to dequeue, 0 to dequeue all.
*/ */
virtual int get_packets(int max_count, __SrsSharedPtrMessage**& pmsgs, int& count); virtual int get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& count);
private: private:
/** /**
* remove a gop from the front. * remove a gop from the front.
@ -150,14 +150,14 @@ public:
* @param tbv timebase of video. * @param tbv timebase of video.
* used to calc the video time delta if time-jitter detected. * used to calc the video time delta if time-jitter detected.
*/ */
virtual int enqueue(__SrsSharedPtrMessage* msg, int tba, int tbv); virtual int enqueue(SrsSharedPtrMessage* msg, int tba, int tbv);
/** /**
* get packets in consumer queue. * get packets in consumer queue.
* @pmsgs SrsMessages*[], output the prt array. * @pmsgs SrsMessages*[], output the prt array.
* @count the count in array. * @count the count in array.
* @max_count the max count to dequeue, 0 to dequeue all. * @max_count the max count to dequeue, 0 to dequeue all.
*/ */
virtual int get_packets(int max_count, __SrsSharedPtrMessage**& pmsgs, int& count); virtual int get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& count);
/** /**
* when client send the pause message. * when client send the pause message.
*/ */
@ -185,7 +185,7 @@ private:
/** /**
* cached gop. * cached gop.
*/ */
std::vector<__SrsSharedPtrMessage*> gop_cache; std::vector<SrsSharedPtrMessage*> gop_cache;
public: public:
SrsGopCache(); SrsGopCache();
virtual ~SrsGopCache(); virtual ~SrsGopCache();
@ -196,7 +196,7 @@ public:
* 1. cache the gop when got h264 video packet. * 1. cache the gop when got h264 video packet.
* 2. clear gop when got keyframe. * 2. clear gop when got keyframe.
*/ */
virtual int cache(__SrsSharedPtrMessage* msg); virtual int cache(SrsSharedPtrMessage* msg);
virtual void clear(); virtual void clear();
virtual int dump(SrsConsumer* consumer, int tba, int tbv); virtual int dump(SrsConsumer* consumer, int tba, int tbv);
/** /**
@ -267,11 +267,11 @@ private:
// TODO: FIXME: to support reload atc. // TODO: FIXME: to support reload atc.
bool atc; bool atc;
private: private:
__SrsSharedPtrMessage* cache_metadata; SrsSharedPtrMessage* cache_metadata;
// the cached video sequence header. // the cached video sequence header.
__SrsSharedPtrMessage* cache_sh_video; SrsSharedPtrMessage* cache_sh_video;
// the cached audio sequence header. // the cached audio sequence header.
__SrsSharedPtrMessage* cache_sh_audio; SrsSharedPtrMessage* cache_sh_audio;
public: public:
/** /**
* @param _req the client request object, * @param _req the client request object,
@ -299,9 +299,9 @@ public:
virtual int on_dvr_request_sh(); virtual int on_dvr_request_sh();
public: public:
virtual bool can_publish(); virtual bool can_publish();
virtual int on_meta_data(__SrsMessage* msg, SrsOnMetaDataPacket* metadata); virtual int on_meta_data(SrsMessage* msg, SrsOnMetaDataPacket* metadata);
virtual int on_audio(__SrsMessage* audio); virtual int on_audio(SrsMessage* audio);
virtual int on_video(__SrsMessage* video); virtual int on_video(SrsMessage* video);
/** /**
* publish stream event notify. * publish stream event notify.
* @param _req the request from client, the source will deep copy it, * @param _req the request from client, the source will deep copy it,
@ -322,7 +322,7 @@ public:
// for edge, when publish edge stream, check the state // for edge, when publish edge stream, check the state
virtual int on_edge_start_publish(); virtual int on_edge_start_publish();
// for edge, proxy the publish // for edge, proxy the publish
virtual int on_edge_proxy_publish(__SrsMessage* msg); virtual int on_edge_proxy_publish(SrsMessage* msg);
// for edge, proxy stop publish // for edge, proxy stop publish
virtual void on_edge_proxy_unpublish(); virtual void on_edge_proxy_unpublish();
private: private:

2
trunk/src/core/srs_core.hpp
View file

@ -31,7 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
// current release version // current release version
#define VERSION_MAJOR "0" #define VERSION_MAJOR "0"
#define VERSION_MINOR "9" #define VERSION_MINOR "9"
#define VERSION_REVISION "80" #define VERSION_REVISION "81"
#define RTMP_SIG_SRS_VERSION VERSION_MAJOR"."VERSION_MINOR"."VERSION_REVISION #define RTMP_SIG_SRS_VERSION VERSION_MAJOR"."VERSION_MINOR"."VERSION_REVISION
// server info. // server info.
#define RTMP_SIG_SRS_KEY "srs" #define RTMP_SIG_SRS_KEY "srs"

16
trunk/src/libs/srs_librtmp.cpp
View file

@ -283,15 +283,15 @@ int srs_read_packet(srs_rtmp_t rtmp, int* type, u_int32_t* timestamp, char** dat
Context* context = (Context*)rtmp; Context* context = (Context*)rtmp;
for (;;) { for (;;) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
if ((ret = context->rtmp->__recv_message(&msg)) != ERROR_SUCCESS) { if ((ret = context->rtmp->recv_message(&msg)) != ERROR_SUCCESS) {
return ret; return ret;
} }
if (!msg) { if (!msg) {
continue; continue;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
if (msg->header.is_audio()) { if (msg->header.is_audio()) {
*type = SRS_RTMP_TYPE_AUDIO; *type = SRS_RTMP_TYPE_AUDIO;
@ -332,13 +332,13 @@ int srs_write_packet(srs_rtmp_t rtmp, int type, u_int32_t timestamp, char* data,
srs_assert(rtmp != NULL); srs_assert(rtmp != NULL);
Context* context = (Context*)rtmp; Context* context = (Context*)rtmp;
__SrsSharedPtrMessage* msg = NULL; SrsSharedPtrMessage* msg = NULL;
if (type == SRS_RTMP_TYPE_AUDIO) { if (type == SRS_RTMP_TYPE_AUDIO) {
SrsMessageHeader header; SrsMessageHeader header;
header.initialize_audio(size, timestamp, context->stream_id); header.initialize_audio(size, timestamp, context->stream_id);
msg = new __SrsSharedPtrMessage(); msg = new SrsSharedPtrMessage();
if ((ret = msg->initialize(&header, data, size)) != ERROR_SUCCESS) { if ((ret = msg->initialize(&header, data, size)) != ERROR_SUCCESS) {
srs_freepa(data); srs_freepa(data);
return ret; return ret;
@ -347,7 +347,7 @@ int srs_write_packet(srs_rtmp_t rtmp, int type, u_int32_t timestamp, char* data,
SrsMessageHeader header; SrsMessageHeader header;
header.initialize_video(size, timestamp, context->stream_id); header.initialize_video(size, timestamp, context->stream_id);
msg = new __SrsSharedPtrMessage(); msg = new SrsSharedPtrMessage();
if ((ret = msg->initialize(&header, data, size)) != ERROR_SUCCESS) { if ((ret = msg->initialize(&header, data, size)) != ERROR_SUCCESS) {
srs_freepa(data); srs_freepa(data);
return ret; return ret;
@ -356,7 +356,7 @@ int srs_write_packet(srs_rtmp_t rtmp, int type, u_int32_t timestamp, char* data,
SrsMessageHeader header; SrsMessageHeader header;
header.initialize_amf0_script(size, context->stream_id); header.initialize_amf0_script(size, context->stream_id);
msg = new __SrsSharedPtrMessage(); msg = new SrsSharedPtrMessage();
if ((ret = msg->initialize(&header, data, size)) != ERROR_SUCCESS) { if ((ret = msg->initialize(&header, data, size)) != ERROR_SUCCESS) {
srs_freepa(data); srs_freepa(data);
return ret; return ret;
@ -365,7 +365,7 @@ int srs_write_packet(srs_rtmp_t rtmp, int type, u_int32_t timestamp, char* data,
if (msg) { if (msg) {
// send out encoded msg. // send out encoded msg.
if ((ret = context->rtmp->__send_and_free_message(msg)) != ERROR_SUCCESS) { if ((ret = context->rtmp->send_and_free_message(msg)) != ERROR_SUCCESS) {
return ret; return ret;
} }
} else { } else {

33
trunk/src/main/srs_main_bandcheck.cpp
View file

@ -203,6 +203,7 @@ static std::string g_vhost = "bandcheck.srs.com";
#define BUILD_VERSION "srs band check 0.1" #define BUILD_VERSION "srs band check 0.1"
// TODO: FIXME: by winlin, the bandwidth test tool has logic bug.
int main(int argc ,char* argv[]) int main(int argc ,char* argv[])
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -341,13 +342,13 @@ int SrsBandCheckClient::expect_start_play()
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
// expect connect _result // expect connect _result
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsBandwidthPacket* pkt = NULL; SrsBandwidthPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
srs_error("expect bandcheck start play message failed. ret=%d", ret); srs_error("expect bandcheck start play message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsBandwidthPacket, pkt, false); SrsAutoFree(SrsBandwidthPacket, pkt, false);
srs_info("get bandcheck start play message"); srs_info("get bandcheck start play message");
@ -365,7 +366,7 @@ int SrsBandCheckClient::send_starting_play()
SrsBandwidthPacket* pkt = new SrsBandwidthPacket; SrsBandwidthPacket* pkt = new SrsBandwidthPacket;
pkt->command_name = SRS_BW_CHECK_STARTING_PLAY; pkt->command_name = SRS_BW_CHECK_STARTING_PLAY;
if ((ret = __send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send starting play msg failed. ret=%d", ret); srs_error("send starting play msg failed. ret=%d", ret);
return ret; return ret;
} }
@ -378,13 +379,13 @@ int SrsBandCheckClient::expect_stop_play()
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
while (true) { while (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsBandwidthPacket* pkt = NULL; SrsBandwidthPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
srs_error("expect stop play message failed. ret=%d", ret); srs_error("expect stop play message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsBandwidthPacket, pkt, false); SrsAutoFree(SrsBandwidthPacket, pkt, false);
srs_info("get bandcheck stop play message"); srs_info("get bandcheck stop play message");
@ -402,7 +403,7 @@ int SrsBandCheckClient::send_stopped_play()
SrsBandwidthPacket* pkt = new SrsBandwidthPacket; SrsBandwidthPacket* pkt = new SrsBandwidthPacket;
pkt->command_name = SRS_BW_CHECK_STOPPED_PLAY; pkt->command_name = SRS_BW_CHECK_STOPPED_PLAY;
if ((ret = __send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send stopped play msg failed. ret=%d", ret); srs_error("send stopped play msg failed. ret=%d", ret);
return ret; return ret;
} }
@ -415,13 +416,13 @@ int SrsBandCheckClient::expect_start_pub()
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
while (true) { while (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsBandwidthPacket* pkt = NULL; SrsBandwidthPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
srs_error("expect start pub message failed. ret=%d", ret); srs_error("expect start pub message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsBandwidthPacket, pkt, false); SrsAutoFree(SrsBandwidthPacket, pkt, false);
srs_info("get bandcheck start pub message"); srs_info("get bandcheck start pub message");
@ -439,7 +440,7 @@ int SrsBandCheckClient::send_starting_pub()
SrsBandwidthPacket* pkt = new SrsBandwidthPacket; SrsBandwidthPacket* pkt = new SrsBandwidthPacket;
pkt->command_name = SRS_BW_CHECK_STARTING_PUBLISH; pkt->command_name = SRS_BW_CHECK_STARTING_PUBLISH;
if ((ret = __send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send starting play msg failed. ret=%d", ret); srs_error("send starting play msg failed. ret=%d", ret);
return ret; return ret;
} }
@ -465,7 +466,7 @@ int SrsBandCheckClient::send_pub_data()
} }
data_count += 100; data_count += 100;
if ((ret = __send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send publish message failed.ret=%d", ret); srs_error("send publish message failed.ret=%d", ret);
return ret; return ret;
} }
@ -485,12 +486,12 @@ int SrsBandCheckClient::expect_stop_pub()
this->set_recv_timeout(1000 * 1000); this->set_recv_timeout(1000 * 1000);
this->set_send_timeout(1000 * 1000); this->set_send_timeout(1000 * 1000);
__SrsMessage* msg; SrsMessage* msg;
SrsBandwidthPacket* pkt; SrsBandwidthPacket* pkt;
if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(this->protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(this->protocol, &msg, &pkt)) != ERROR_SUCCESS) {
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsBandwidthPacket, pkt, false); SrsAutoFree(SrsBandwidthPacket, pkt, false);
if (pkt->command_name == SRS_BW_CHECK_STOP_PUBLISH) { if (pkt->command_name == SRS_BW_CHECK_STOP_PUBLISH) {
return ret; return ret;
@ -504,13 +505,13 @@ int SrsBandCheckClient::expect_finished()
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
while (true) { while (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsBandwidthPacket* pkt = NULL; SrsBandwidthPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsBandwidthPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
srs_error("expect finished message failed. ret=%d", ret); srs_error("expect finished message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsBandwidthPacket, pkt, false); SrsAutoFree(SrsBandwidthPacket, pkt, false);
srs_info("get bandcheck finished message"); srs_info("get bandcheck finished message");
@ -593,7 +594,7 @@ int SrsBandCheckClient::send_stopped_pub()
SrsBandwidthPacket* pkt = new SrsBandwidthPacket; SrsBandwidthPacket* pkt = new SrsBandwidthPacket;
pkt->command_name = SRS_BW_CHECK_STOPPED_PUBLISH; pkt->command_name = SRS_BW_CHECK_STOPPED_PUBLISH;
if ((ret = __send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send stopped pub msg failed. ret=%d", ret); srs_error("send stopped pub msg failed. ret=%d", ret);
return ret; return ret;
} }
@ -608,7 +609,7 @@ int SrsBandCheckClient::send_final()
SrsBandwidthPacket* pkt = new SrsBandwidthPacket; SrsBandwidthPacket* pkt = new SrsBandwidthPacket;
pkt->command_name = SRS_BW_CHECK_FLASH_FINAL; pkt->command_name = SRS_BW_CHECK_FLASH_FINAL;
if ((ret = __send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send final msg failed. ret=%d", ret); srs_error("send final msg failed. ret=%d", ret);
return ret; return ret;
} }

150
trunk/src/rtmp/srs_protocol_rtmp.cpp
View file

@ -372,24 +372,24 @@ int SrsRtmpClient::get_send_kbps()
return protocol->get_send_kbps(); return protocol->get_send_kbps();
} }
int SrsRtmpClient::__recv_message(__SrsMessage** pmsg) int SrsRtmpClient::recv_message(SrsMessage** pmsg)
{ {
return protocol->__recv_message(pmsg); return protocol->recv_message(pmsg);
} }
int SrsRtmpClient::__decode_message(__SrsMessage* msg, SrsPacket** ppacket) int SrsRtmpClient::decode_message(SrsMessage* msg, SrsPacket** ppacket)
{ {
return protocol->__decode_message(msg, ppacket); return protocol->decode_message(msg, ppacket);
} }
int SrsRtmpClient::__send_and_free_message(__SrsMessage* msg) int SrsRtmpClient::send_and_free_message(SrsMessage* msg)
{ {
return protocol->__send_and_free_message(msg); return protocol->send_and_free_message(msg);
} }
int SrsRtmpClient::__send_and_free_packet(SrsPacket* packet, int stream_id) int SrsRtmpClient::send_and_free_packet(SrsPacket* packet, int stream_id)
{ {
return protocol->__send_and_free_packet(packet, stream_id); return protocol->send_and_free_packet(packet, stream_id);
} }
int SrsRtmpClient::handshake() int SrsRtmpClient::handshake()
@ -466,7 +466,7 @@ int SrsRtmpClient::connect_app(string app, string tc_url)
pkt->command_object->set("pageUrl", SrsAmf0Any::str()); pkt->command_object->set("pageUrl", SrsAmf0Any::str());
pkt->command_object->set("objectEncoding", SrsAmf0Any::number(0)); pkt->command_object->set("objectEncoding", SrsAmf0Any::number(0));
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
return ret; return ret;
} }
} }
@ -475,19 +475,19 @@ int SrsRtmpClient::connect_app(string app, string tc_url)
if (true) { if (true) {
SrsSetWindowAckSizePacket* pkt = new SrsSetWindowAckSizePacket(); SrsSetWindowAckSizePacket* pkt = new SrsSetWindowAckSizePacket();
pkt->ackowledgement_window_size = 2500000; pkt->ackowledgement_window_size = 2500000;
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
return ret; return ret;
} }
} }
// expect connect _result // expect connect _result
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsConnectAppResPacket* pkt = NULL; SrsConnectAppResPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsConnectAppResPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsConnectAppResPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
srs_error("expect connect app response message failed. ret=%d", ret); srs_error("expect connect app response message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsConnectAppResPacket, pkt, false); SrsAutoFree(SrsConnectAppResPacket, pkt, false);
srs_info("get connect app response message"); srs_info("get connect app response message");
@ -501,20 +501,20 @@ int SrsRtmpClient::create_stream(int& stream_id)
// CreateStream // CreateStream
if (true) { if (true) {
SrsCreateStreamPacket* pkt = new SrsCreateStreamPacket(); SrsCreateStreamPacket* pkt = new SrsCreateStreamPacket();
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
return ret; return ret;
} }
} }
// CreateStream _result. // CreateStream _result.
if (true) { if (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsCreateStreamResPacket* pkt = NULL; SrsCreateStreamResPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsCreateStreamResPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsCreateStreamResPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
srs_error("expect create stream response message failed. ret=%d", ret); srs_error("expect create stream response message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsCreateStreamResPacket, pkt, false); SrsAutoFree(SrsCreateStreamResPacket, pkt, false);
srs_info("get create stream response message"); srs_info("get create stream response message");
@ -532,7 +532,7 @@ int SrsRtmpClient::play(string stream, int stream_id)
if (true) { if (true) {
SrsPlayPacket* pkt = new SrsPlayPacket(); SrsPlayPacket* pkt = new SrsPlayPacket();
pkt->stream_name = stream; pkt->stream_name = stream;
if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
srs_error("send play stream failed. " srs_error("send play stream failed. "
"stream=%s, stream_id=%d, ret=%d", "stream=%s, stream_id=%d, ret=%d",
stream.c_str(), stream_id, ret); stream.c_str(), stream_id, ret);
@ -549,7 +549,7 @@ int SrsRtmpClient::play(string stream, int stream_id)
pkt->event_data = stream_id; pkt->event_data = stream_id;
pkt->extra_data = buffer_length_ms; pkt->extra_data = buffer_length_ms;
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send set buffer length failed. " srs_error("send set buffer length failed. "
"stream=%s, stream_id=%d, bufferLength=%d, ret=%d", "stream=%s, stream_id=%d, bufferLength=%d, ret=%d",
stream.c_str(), stream_id, buffer_length_ms, ret); stream.c_str(), stream_id, buffer_length_ms, ret);
@ -561,7 +561,7 @@ int SrsRtmpClient::play(string stream, int stream_id)
if (true) { if (true) {
SrsSetChunkSizePacket* pkt = new SrsSetChunkSizePacket(); SrsSetChunkSizePacket* pkt = new SrsSetChunkSizePacket();
pkt->chunk_size = SRS_CONF_DEFAULT_CHUNK_SIZE; pkt->chunk_size = SRS_CONF_DEFAULT_CHUNK_SIZE;
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send set chunk size failed. " srs_error("send set chunk size failed. "
"stream=%s, chunk_size=%d, ret=%d", "stream=%s, chunk_size=%d, ret=%d",
stream.c_str(), SRS_CONF_DEFAULT_CHUNK_SIZE, ret); stream.c_str(), SRS_CONF_DEFAULT_CHUNK_SIZE, ret);
@ -580,7 +580,7 @@ int SrsRtmpClient::publish(string stream, int stream_id)
if (true) { if (true) {
SrsSetChunkSizePacket* pkt = new SrsSetChunkSizePacket(); SrsSetChunkSizePacket* pkt = new SrsSetChunkSizePacket();
pkt->chunk_size = SRS_CONF_DEFAULT_CHUNK_SIZE; pkt->chunk_size = SRS_CONF_DEFAULT_CHUNK_SIZE;
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send set chunk size failed. " srs_error("send set chunk size failed. "
"stream=%s, chunk_size=%d, ret=%d", "stream=%s, chunk_size=%d, ret=%d",
stream.c_str(), SRS_CONF_DEFAULT_CHUNK_SIZE, ret); stream.c_str(), SRS_CONF_DEFAULT_CHUNK_SIZE, ret);
@ -592,7 +592,7 @@ int SrsRtmpClient::publish(string stream, int stream_id)
if (true) { if (true) {
SrsPublishPacket* pkt = new SrsPublishPacket(); SrsPublishPacket* pkt = new SrsPublishPacket();
pkt->stream_name = stream; pkt->stream_name = stream;
if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
srs_error("send publish message failed. " srs_error("send publish message failed. "
"stream=%s, stream_id=%d, ret=%d", "stream=%s, stream_id=%d, ret=%d",
stream.c_str(), stream_id, ret); stream.c_str(), stream_id, ret);
@ -612,7 +612,7 @@ int SrsRtmpClient::fmle_publish(string stream, int& stream_id)
// SrsFMLEStartPacket // SrsFMLEStartPacket
if (true) { if (true) {
SrsFMLEStartPacket* pkt = SrsFMLEStartPacket::create_release_stream(stream); SrsFMLEStartPacket* pkt = SrsFMLEStartPacket::create_release_stream(stream);
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send FMLE publish " srs_error("send FMLE publish "
"release stream failed. stream=%s, ret=%d", stream.c_str(), ret); "release stream failed. stream=%s, ret=%d", stream.c_str(), ret);
return ret; return ret;
@ -622,7 +622,7 @@ int SrsRtmpClient::fmle_publish(string stream, int& stream_id)
// FCPublish // FCPublish
if (true) { if (true) {
SrsFMLEStartPacket* pkt = SrsFMLEStartPacket::create_FC_publish(stream); SrsFMLEStartPacket* pkt = SrsFMLEStartPacket::create_FC_publish(stream);
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send FMLE publish " srs_error("send FMLE publish "
"FCPublish failed. stream=%s, ret=%d", stream.c_str(), ret); "FCPublish failed. stream=%s, ret=%d", stream.c_str(), ret);
return ret; return ret;
@ -633,7 +633,7 @@ int SrsRtmpClient::fmle_publish(string stream, int& stream_id)
if (true) { if (true) {
SrsCreateStreamPacket* pkt = new SrsCreateStreamPacket(); SrsCreateStreamPacket* pkt = new SrsCreateStreamPacket();
pkt->transaction_id = 4; pkt->transaction_id = 4;
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send FMLE publish " srs_error("send FMLE publish "
"createStream failed. stream=%s, ret=%d", stream.c_str(), ret); "createStream failed. stream=%s, ret=%d", stream.c_str(), ret);
return ret; return ret;
@ -642,13 +642,13 @@ int SrsRtmpClient::fmle_publish(string stream, int& stream_id)
// expect result of CreateStream // expect result of CreateStream
if (true) { if (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsCreateStreamResPacket* pkt = NULL; SrsCreateStreamResPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsCreateStreamResPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsCreateStreamResPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
srs_error("expect create stream response message failed. ret=%d", ret); srs_error("expect create stream response message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsCreateStreamResPacket, pkt, false); SrsAutoFree(SrsCreateStreamResPacket, pkt, false);
srs_info("get create stream response message"); srs_info("get create stream response message");
@ -659,7 +659,7 @@ int SrsRtmpClient::fmle_publish(string stream, int& stream_id)
if (true) { if (true) {
SrsPublishPacket* pkt = new SrsPublishPacket(); SrsPublishPacket* pkt = new SrsPublishPacket();
pkt->stream_name = stream; pkt->stream_name = stream;
if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
srs_error("send FMLE publish publish failed. " srs_error("send FMLE publish publish failed. "
"stream=%s, stream_id=%d, ret=%d", stream.c_str(), stream_id, ret); "stream=%s, stream_id=%d, ret=%d", stream.c_str(), stream_id, ret);
return ret; return ret;
@ -727,24 +727,24 @@ int SrsRtmpServer::get_send_kbps()
return protocol->get_send_kbps(); return protocol->get_send_kbps();
} }
int SrsRtmpServer::__recv_message(__SrsMessage** pmsg) int SrsRtmpServer::recv_message(SrsMessage** pmsg)
{ {
return protocol->__recv_message(pmsg); return protocol->recv_message(pmsg);
} }
int SrsRtmpServer::__decode_message(__SrsMessage* msg, SrsPacket** ppacket) int SrsRtmpServer::decode_message(SrsMessage* msg, SrsPacket** ppacket)
{ {
return protocol->__decode_message(msg, ppacket); return protocol->decode_message(msg, ppacket);
} }
int SrsRtmpServer::__send_and_free_message(__SrsMessage* msg) int SrsRtmpServer::send_and_free_message(SrsMessage* msg)
{ {
return protocol->__send_and_free_message(msg); return protocol->send_and_free_message(msg);
} }
int SrsRtmpServer::__send_and_free_packet(SrsPacket* packet, int stream_id) int SrsRtmpServer::send_and_free_packet(SrsPacket* packet, int stream_id)
{ {
return protocol->__send_and_free_packet(packet, stream_id); return protocol->send_and_free_packet(packet, stream_id);
} }
int SrsRtmpServer::handshake() int SrsRtmpServer::handshake()
@ -773,13 +773,13 @@ int SrsRtmpServer::connect_app(SrsRequest* req)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsConnectAppPacket* pkt = NULL; SrsConnectAppPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsConnectAppPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsConnectAppPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
srs_error("expect connect app message failed. ret=%d", ret); srs_error("expect connect app message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsConnectAppPacket, pkt, false); SrsAutoFree(SrsConnectAppPacket, pkt, false);
srs_info("get connect app message"); srs_info("get connect app message");
@ -815,7 +815,7 @@ int SrsRtmpServer::set_window_ack_size(int ack_size)
SrsSetWindowAckSizePacket* pkt = new SrsSetWindowAckSizePacket(); SrsSetWindowAckSizePacket* pkt = new SrsSetWindowAckSizePacket();
pkt->ackowledgement_window_size = ack_size; pkt->ackowledgement_window_size = ack_size;
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send ack size message failed. ret=%d", ret); srs_error("send ack size message failed. ret=%d", ret);
return ret; return ret;
} }
@ -831,7 +831,7 @@ int SrsRtmpServer::set_peer_bandwidth(int bandwidth, int type)
SrsSetPeerBandwidthPacket* pkt = new SrsSetPeerBandwidthPacket(); SrsSetPeerBandwidthPacket* pkt = new SrsSetPeerBandwidthPacket();
pkt->bandwidth = bandwidth; pkt->bandwidth = bandwidth;
pkt->type = type; pkt->type = type;
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send set bandwidth message failed. ret=%d", ret); srs_error("send set bandwidth message failed. ret=%d", ret);
return ret; return ret;
} }
@ -874,7 +874,7 @@ int SrsRtmpServer::response_connect_app(SrsRequest *req, const char* server_ip)
data->set("srs_server_ip", SrsAmf0Any::str(server_ip)); data->set("srs_server_ip", SrsAmf0Any::str(server_ip));
} }
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send connect app response message failed. ret=%d", ret); srs_error("send connect app response message failed. ret=%d", ret);
return ret; return ret;
} }
@ -894,7 +894,7 @@ void SrsRtmpServer::response_connect_reject(SrsRequest *req, const char* desc)
pkt->props->set(StatusDescription, SrsAmf0Any::str(desc)); pkt->props->set(StatusDescription, SrsAmf0Any::str(desc));
//pkt->props->set("objectEncoding", SrsAmf0Any::number(req->objectEncoding)); //pkt->props->set("objectEncoding", SrsAmf0Any::number(req->objectEncoding));
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send connect app response rejected message failed. ret=%d", ret); srs_error("send connect app response rejected message failed. ret=%d", ret);
return; return;
} }
@ -908,7 +908,7 @@ int SrsRtmpServer::on_bw_done()
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
SrsOnBWDonePacket* pkt = new SrsOnBWDonePacket(); SrsOnBWDonePacket* pkt = new SrsOnBWDonePacket();
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send onBWDone message failed. ret=%d", ret); srs_error("send onBWDone message failed. ret=%d", ret);
return ret; return ret;
} }
@ -923,13 +923,13 @@ int SrsRtmpServer::identify_client(int stream_id, SrsRtmpConnType& type, string&
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
while (true) { while (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
if ((ret = protocol->__recv_message(&msg)) != ERROR_SUCCESS) { if ((ret = protocol->recv_message(&msg)) != ERROR_SUCCESS) {
srs_error("recv identify client message failed. ret=%d", ret); srs_error("recv identify client message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
if (!msg->header.is_amf0_command() && !msg->header.is_amf3_command()) { if (!msg->header.is_amf0_command() && !msg->header.is_amf3_command()) {
srs_trace("identify ignore messages except " srs_trace("identify ignore messages except "
@ -938,7 +938,7 @@ int SrsRtmpServer::identify_client(int stream_id, SrsRtmpConnType& type, string&
} }
SrsPacket* pkt = NULL; SrsPacket* pkt = NULL;
if ((ret = protocol->__decode_message(msg, &pkt)) != ERROR_SUCCESS) { if ((ret = protocol->decode_message(msg, &pkt)) != ERROR_SUCCESS) {
srs_error("identify decode message failed. ret=%d", ret); srs_error("identify decode message failed. ret=%d", ret);
return ret; return ret;
} }
@ -970,7 +970,7 @@ int SrsRtmpServer::set_chunk_size(int chunk_size)
SrsSetChunkSizePacket* pkt = new SrsSetChunkSizePacket(); SrsSetChunkSizePacket* pkt = new SrsSetChunkSizePacket();
pkt->chunk_size = chunk_size; pkt->chunk_size = chunk_size;
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send set chunk size message failed. ret=%d", ret); srs_error("send set chunk size message failed. ret=%d", ret);
return ret; return ret;
} }
@ -988,7 +988,7 @@ int SrsRtmpServer::start_play(int stream_id)
SrsUserControlPacket* pkt = new SrsUserControlPacket(); SrsUserControlPacket* pkt = new SrsUserControlPacket();
pkt->event_type = SrcPCUCStreamBegin; pkt->event_type = SrcPCUCStreamBegin;
pkt->event_data = stream_id; pkt->event_data = stream_id;
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send PCUC(StreamBegin) message failed. ret=%d", ret); srs_error("send PCUC(StreamBegin) message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1005,7 +1005,7 @@ int SrsRtmpServer::start_play(int stream_id)
pkt->data->set(StatusDetails, SrsAmf0Any::str("stream")); pkt->data->set(StatusDetails, SrsAmf0Any::str("stream"));
pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID)); pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID));
if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
srs_error("send onStatus(NetStream.Play.Reset) message failed. ret=%d", ret); srs_error("send onStatus(NetStream.Play.Reset) message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1022,7 +1022,7 @@ int SrsRtmpServer::start_play(int stream_id)
pkt->data->set(StatusDetails, SrsAmf0Any::str("stream")); pkt->data->set(StatusDetails, SrsAmf0Any::str("stream"));
pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID)); pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID));
if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
srs_error("send onStatus(NetStream.Play.Reset) message failed. ret=%d", ret); srs_error("send onStatus(NetStream.Play.Reset) message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1032,7 +1032,7 @@ int SrsRtmpServer::start_play(int stream_id)
// |RtmpSampleAccess(false, false) // |RtmpSampleAccess(false, false)
if (true) { if (true) {
SrsSampleAccessPacket* pkt = new SrsSampleAccessPacket(); SrsSampleAccessPacket* pkt = new SrsSampleAccessPacket();
if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
srs_error("send |RtmpSampleAccess(false, false) message failed. ret=%d", ret); srs_error("send |RtmpSampleAccess(false, false) message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1043,7 +1043,7 @@ int SrsRtmpServer::start_play(int stream_id)
if (true) { if (true) {
SrsOnStatusDataPacket* pkt = new SrsOnStatusDataPacket(); SrsOnStatusDataPacket* pkt = new SrsOnStatusDataPacket();
pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodeDataStart)); pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodeDataStart));
if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
srs_error("send onStatus(NetStream.Data.Start) message failed. ret=%d", ret); srs_error("send onStatus(NetStream.Data.Start) message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1068,7 +1068,7 @@ int SrsRtmpServer::on_play_client_pause(int stream_id, bool is_pause)
pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodeStreamPause)); pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodeStreamPause));
pkt->data->set(StatusDescription, SrsAmf0Any::str("Paused stream.")); pkt->data->set(StatusDescription, SrsAmf0Any::str("Paused stream."));
if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
srs_error("send onStatus(NetStream.Pause.Notify) message failed. ret=%d", ret); srs_error("send onStatus(NetStream.Pause.Notify) message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1081,7 +1081,7 @@ int SrsRtmpServer::on_play_client_pause(int stream_id, bool is_pause)
pkt->event_type = SrcPCUCStreamEOF; pkt->event_type = SrcPCUCStreamEOF;
pkt->event_data = stream_id; pkt->event_data = stream_id;
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send PCUC(StreamEOF) message failed. ret=%d", ret); srs_error("send PCUC(StreamEOF) message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1096,7 +1096,7 @@ int SrsRtmpServer::on_play_client_pause(int stream_id, bool is_pause)
pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodeStreamUnpause)); pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodeStreamUnpause));
pkt->data->set(StatusDescription, SrsAmf0Any::str("Unpaused stream.")); pkt->data->set(StatusDescription, SrsAmf0Any::str("Unpaused stream."));
if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
srs_error("send onStatus(NetStream.Unpause.Notify) message failed. ret=%d", ret); srs_error("send onStatus(NetStream.Unpause.Notify) message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1109,7 +1109,7 @@ int SrsRtmpServer::on_play_client_pause(int stream_id, bool is_pause)
pkt->event_type = SrcPCUCStreamBegin; pkt->event_type = SrcPCUCStreamBegin;
pkt->event_data = stream_id; pkt->event_data = stream_id;
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send PCUC(StreanBegin) message failed. ret=%d", ret); srs_error("send PCUC(StreanBegin) message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1127,7 +1127,7 @@ int SrsRtmpServer::start_fmle_publish(int stream_id)
// FCPublish // FCPublish
double fc_publish_tid = 0; double fc_publish_tid = 0;
if (true) { if (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsFMLEStartPacket* pkt = NULL; SrsFMLEStartPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsFMLEStartPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsFMLEStartPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
srs_error("recv FCPublish message failed. ret=%d", ret); srs_error("recv FCPublish message failed. ret=%d", ret);
@ -1135,7 +1135,7 @@ int SrsRtmpServer::start_fmle_publish(int stream_id)
} }
srs_info("recv FCPublish request message success."); srs_info("recv FCPublish request message success.");
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsFMLEStartPacket, pkt, false); SrsAutoFree(SrsFMLEStartPacket, pkt, false);
fc_publish_tid = pkt->transaction_id; fc_publish_tid = pkt->transaction_id;
@ -1143,7 +1143,7 @@ int SrsRtmpServer::start_fmle_publish(int stream_id)
// FCPublish response // FCPublish response
if (true) { if (true) {
SrsFMLEStartResPacket* pkt = new SrsFMLEStartResPacket(fc_publish_tid); SrsFMLEStartResPacket* pkt = new SrsFMLEStartResPacket(fc_publish_tid);
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send FCPublish response message failed. ret=%d", ret); srs_error("send FCPublish response message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1153,7 +1153,7 @@ int SrsRtmpServer::start_fmle_publish(int stream_id)
// createStream // createStream
double create_stream_tid = 0; double create_stream_tid = 0;
if (true) { if (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsCreateStreamPacket* pkt = NULL; SrsCreateStreamPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsCreateStreamPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsCreateStreamPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
srs_error("recv createStream message failed. ret=%d", ret); srs_error("recv createStream message failed. ret=%d", ret);
@ -1161,7 +1161,7 @@ int SrsRtmpServer::start_fmle_publish(int stream_id)
} }
srs_info("recv createStream request message success."); srs_info("recv createStream request message success.");
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsCreateStreamPacket, pkt, false); SrsAutoFree(SrsCreateStreamPacket, pkt, false);
create_stream_tid = pkt->transaction_id; create_stream_tid = pkt->transaction_id;
@ -1169,7 +1169,7 @@ int SrsRtmpServer::start_fmle_publish(int stream_id)
// createStream response // createStream response
if (true) { if (true) {
SrsCreateStreamResPacket* pkt = new SrsCreateStreamResPacket(create_stream_tid, stream_id); SrsCreateStreamResPacket* pkt = new SrsCreateStreamResPacket(create_stream_tid, stream_id);
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send createStream response message failed. ret=%d", ret); srs_error("send createStream response message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1178,7 +1178,7 @@ int SrsRtmpServer::start_fmle_publish(int stream_id)
// publish // publish
if (true) { if (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
SrsPublishPacket* pkt = NULL; SrsPublishPacket* pkt = NULL;
if ((ret = __srs_rtmp_expect_message<SrsPublishPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) { if ((ret = __srs_rtmp_expect_message<SrsPublishPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
srs_error("recv publish message failed. ret=%d", ret); srs_error("recv publish message failed. ret=%d", ret);
@ -1186,7 +1186,7 @@ int SrsRtmpServer::start_fmle_publish(int stream_id)
} }
srs_info("recv publish request message success."); srs_info("recv publish request message success.");
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
SrsAutoFree(SrsPublishPacket, pkt, false); SrsAutoFree(SrsPublishPacket, pkt, false);
} }
// publish response onFCPublish(NetStream.Publish.Start) // publish response onFCPublish(NetStream.Publish.Start)
@ -1197,7 +1197,7 @@ int SrsRtmpServer::start_fmle_publish(int stream_id)
pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodePublishStart)); pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodePublishStart));
pkt->data->set(StatusDescription, SrsAmf0Any::str("Started publishing stream.")); pkt->data->set(StatusDescription, SrsAmf0Any::str("Started publishing stream."));
if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
srs_error("send onFCPublish(NetStream.Publish.Start) message failed. ret=%d", ret); srs_error("send onFCPublish(NetStream.Publish.Start) message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1212,7 +1212,7 @@ int SrsRtmpServer::start_fmle_publish(int stream_id)
pkt->data->set(StatusDescription, SrsAmf0Any::str("Started publishing stream.")); pkt->data->set(StatusDescription, SrsAmf0Any::str("Started publishing stream."));
pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID)); pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID));
if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
srs_error("send onStatus(NetStream.Publish.Start) message failed. ret=%d", ret); srs_error("send onStatus(NetStream.Publish.Start) message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1236,7 +1236,7 @@ int SrsRtmpServer::fmle_unpublish(int stream_id, double unpublish_tid)
pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodeUnpublishSuccess)); pkt->data->set(StatusCode, SrsAmf0Any::str(StatusCodeUnpublishSuccess));
pkt->data->set(StatusDescription, SrsAmf0Any::str("Stop publishing stream.")); pkt->data->set(StatusDescription, SrsAmf0Any::str("Stop publishing stream."));
if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
srs_error("send onFCUnpublish(NetStream.unpublish.Success) message failed. ret=%d", ret); srs_error("send onFCUnpublish(NetStream.unpublish.Success) message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1245,7 +1245,7 @@ int SrsRtmpServer::fmle_unpublish(int stream_id, double unpublish_tid)
// FCUnpublish response // FCUnpublish response
if (true) { if (true) {
SrsFMLEStartResPacket* pkt = new SrsFMLEStartResPacket(unpublish_tid); SrsFMLEStartResPacket* pkt = new SrsFMLEStartResPacket(unpublish_tid);
if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
srs_error("send FCUnpublish response message failed. ret=%d", ret); srs_error("send FCUnpublish response message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1260,7 +1260,7 @@ int SrsRtmpServer::fmle_unpublish(int stream_id, double unpublish_tid)
pkt->data->set(StatusDescription, SrsAmf0Any::str("Stream is now unpublished")); pkt->data->set(StatusDescription, SrsAmf0Any::str("Stream is now unpublished"));
pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID)); pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID));
if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
srs_error("send onStatus(NetStream.Unpublish.Success) message failed. ret=%d", ret); srs_error("send onStatus(NetStream.Unpublish.Success) message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1285,7 +1285,7 @@ int SrsRtmpServer::start_flash_publish(int stream_id)
pkt->data->set(StatusDescription, SrsAmf0Any::str("Started publishing stream.")); pkt->data->set(StatusDescription, SrsAmf0Any::str("Started publishing stream."));
pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID)); pkt->data->set(StatusClientId, SrsAmf0Any::str(RTMP_SIG_CLIENT_ID));
if ((ret = protocol->__send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, stream_id)) != ERROR_SUCCESS) {
srs_error("send onStatus(NetStream.Publish.Start) message failed. ret=%d", ret); srs_error("send onStatus(NetStream.Publish.Start) message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1303,7 +1303,7 @@ int SrsRtmpServer::identify_create_stream_client(SrsCreateStreamPacket* req, int
if (true) { if (true) {
SrsCreateStreamResPacket* pkt = new SrsCreateStreamResPacket(req->transaction_id, stream_id); SrsCreateStreamResPacket* pkt = new SrsCreateStreamResPacket(req->transaction_id, stream_id);
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send createStream response message failed. ret=%d", ret); srs_error("send createStream response message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1311,13 +1311,13 @@ int SrsRtmpServer::identify_create_stream_client(SrsCreateStreamPacket* req, int
} }
while (true) { while (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
if ((ret = protocol->__recv_message(&msg)) != ERROR_SUCCESS) { if ((ret = protocol->recv_message(&msg)) != ERROR_SUCCESS) {
srs_error("recv identify client message failed. ret=%d", ret); srs_error("recv identify client message failed. ret=%d", ret);
return ret; return ret;
} }
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
if (!msg->header.is_amf0_command() && !msg->header.is_amf3_command()) { if (!msg->header.is_amf0_command() && !msg->header.is_amf3_command()) {
srs_trace("identify ignore messages except " srs_trace("identify ignore messages except "
@ -1326,7 +1326,7 @@ int SrsRtmpServer::identify_create_stream_client(SrsCreateStreamPacket* req, int
} }
SrsPacket* pkt = NULL; SrsPacket* pkt = NULL;
if ((ret = protocol->__decode_message(msg, &pkt)) != ERROR_SUCCESS) { if ((ret = protocol->decode_message(msg, &pkt)) != ERROR_SUCCESS) {
srs_error("identify decode message failed. ret=%d", ret); srs_error("identify decode message failed. ret=%d", ret);
return ret; return ret;
} }
@ -1358,7 +1358,7 @@ int SrsRtmpServer::identify_fmle_publish_client(SrsFMLEStartPacket* req, SrsRtmp
// releaseStream response // releaseStream response
if (true) { if (true) {
SrsFMLEStartResPacket* pkt = new SrsFMLEStartResPacket(req->transaction_id); SrsFMLEStartResPacket* pkt = new SrsFMLEStartResPacket(req->transaction_id);
if ((ret = protocol->__send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = protocol->send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send releaseStream response message failed. ret=%d", ret); srs_error("send releaseStream response message failed. ret=%d", ret);
return ret; return ret;
} }

19
trunk/src/rtmp/srs_protocol_rtmp.hpp
View file

@ -39,10 +39,9 @@ class SrsCommonMessage;
class SrsCreateStreamPacket; class SrsCreateStreamPacket;
class SrsFMLEStartPacket; class SrsFMLEStartPacket;
class SrsPublishPacket; class SrsPublishPacket;
class SrsSharedPtrMessage;
class SrsOnMetaDataPacket; class SrsOnMetaDataPacket;
class SrsPlayPacket; class SrsPlayPacket;
class __SrsMessage; class SrsMessage;
class SrsPacket; class SrsPacket;
/** /**
@ -164,10 +163,10 @@ public:
virtual int64_t get_send_bytes(); virtual int64_t get_send_bytes();
virtual int get_recv_kbps(); virtual int get_recv_kbps();
virtual int get_send_kbps(); virtual int get_send_kbps();
virtual int __recv_message(__SrsMessage** pmsg); virtual int recv_message(SrsMessage** pmsg);
virtual int __decode_message(__SrsMessage* msg, SrsPacket** ppacket); virtual int decode_message(SrsMessage* msg, SrsPacket** ppacket);
virtual int __send_and_free_message(__SrsMessage* msg); virtual int send_and_free_message(SrsMessage* msg);
virtual int __send_and_free_packet(SrsPacket* packet, int stream_id); virtual int send_and_free_packet(SrsPacket* packet, int stream_id);
public: public:
// try complex, then simple handshake. // try complex, then simple handshake.
virtual int handshake(); virtual int handshake();
@ -211,10 +210,10 @@ public:
virtual int64_t get_send_bytes(); virtual int64_t get_send_bytes();
virtual int get_recv_kbps(); virtual int get_recv_kbps();
virtual int get_send_kbps(); virtual int get_send_kbps();
virtual int __recv_message(__SrsMessage** pmsg); virtual int recv_message(SrsMessage** pmsg);
virtual int __decode_message(__SrsMessage* msg, SrsPacket** ppacket); virtual int decode_message(SrsMessage* msg, SrsPacket** ppacket);
virtual int __send_and_free_message(__SrsMessage* msg); virtual int send_and_free_message(SrsMessage* msg);
virtual int __send_and_free_packet(SrsPacket* packet, int stream_id); virtual int send_and_free_packet(SrsPacket* packet, int stream_id);
public: public:
virtual int handshake(); virtual int handshake();
virtual int connect_app(SrsRequest* req); virtual int connect_app(SrsRequest* req);

107
trunk/src/rtmp/srs_protocol_rtmp_stack.cpp
View file

@ -306,14 +306,14 @@ SrsProtocol::SrsProtocol(ISrsProtocolReaderWriter* io)
SrsProtocol::~SrsProtocol() SrsProtocol::~SrsProtocol()
{ {
if (true) { if (true) {
std::map<int, __SrsChunkStream*>::iterator it; std::map<int, SrsChunkStream*>::iterator it;
for (it = __chunk_streams.begin(); it != __chunk_streams.end(); ++it) { for (it = chunk_streams.begin(); it != chunk_streams.end(); ++it) {
__SrsChunkStream* stream = it->second; SrsChunkStream* stream = it->second;
srs_freep(stream); srs_freep(stream);
} }
__chunk_streams.clear(); chunk_streams.clear();
} }
srs_freep(decode_stream); srs_freep(decode_stream);
@ -369,16 +369,16 @@ int SrsProtocol::get_send_kbps()
return skt->get_send_kbps(); return skt->get_send_kbps();
} }
int SrsProtocol::__recv_message(__SrsMessage** pmsg) int SrsProtocol::recv_message(SrsMessage** pmsg)
{ {
*pmsg = NULL; *pmsg = NULL;
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
while (true) { while (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
if ((ret = __recv_interlaced_message(&msg)) != ERROR_SUCCESS) { if ((ret = recv_interlaced_message(&msg)) != ERROR_SUCCESS) {
if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) { if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) {
srs_error("recv interlaced message failed. ret=%d", ret); srs_error("recv interlaced message failed. ret=%d", ret);
} }
@ -398,7 +398,7 @@ int SrsProtocol::__recv_message(__SrsMessage** pmsg)
continue; continue;
} }
if ((ret = __on_recv_message(msg)) != ERROR_SUCCESS) { if ((ret = on_recv_message(msg)) != ERROR_SUCCESS) {
srs_error("hook the received msg failed. ret=%d", ret); srs_error("hook the received msg failed. ret=%d", ret);
srs_freep(msg); srs_freep(msg);
return ret; return ret;
@ -412,7 +412,7 @@ int SrsProtocol::__recv_message(__SrsMessage** pmsg)
return ret; return ret;
} }
int SrsProtocol::__decode_message(__SrsMessage* msg, SrsPacket** ppacket) int SrsProtocol::decode_message(SrsMessage* msg, SrsPacket** ppacket)
{ {
*ppacket = NULL; *ppacket = NULL;
@ -432,7 +432,7 @@ int SrsProtocol::__decode_message(__SrsMessage* msg, SrsPacket** ppacket)
// decode the packet. // decode the packet.
SrsPacket* packet = NULL; SrsPacket* packet = NULL;
if ((ret = __do_decode_message(msg->header, decode_stream, &packet)) != ERROR_SUCCESS) { if ((ret = do_decode_message(msg->header, decode_stream, &packet)) != ERROR_SUCCESS) {
srs_freep(packet); srs_freep(packet);
return ret; return ret;
} }
@ -443,7 +443,7 @@ int SrsProtocol::__decode_message(__SrsMessage* msg, SrsPacket** ppacket)
return ret; return ret;
} }
int SrsProtocol::__do_send_and_free_message(__SrsMessage* msg, SrsPacket* packet) int SrsProtocol::do_send_and_free_message(SrsMessage* msg, SrsPacket* packet)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -561,7 +561,7 @@ int SrsProtocol::__do_send_and_free_message(__SrsMessage* msg, SrsPacket* packet
} }
} while (p < (char*)msg->payload + msg->size); } while (p < (char*)msg->payload + msg->size);
if ((ret = __on_send_message(msg, packet)) != ERROR_SUCCESS) { if ((ret = on_send_message(msg, packet)) != ERROR_SUCCESS) {
srs_error("hook the send message failed. ret=%d", ret); srs_error("hook the send message failed. ret=%d", ret);
return ret; return ret;
} }
@ -569,7 +569,7 @@ int SrsProtocol::__do_send_and_free_message(__SrsMessage* msg, SrsPacket* packet
return ret; return ret;
} }
int SrsProtocol::__do_decode_message(SrsMessageHeader& header, SrsStream* stream, SrsPacket** ppacket) int SrsProtocol::do_decode_message(SrsMessageHeader& header, SrsStream* stream, SrsPacket** ppacket)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -728,15 +728,15 @@ int SrsProtocol::__do_decode_message(SrsMessageHeader& header, SrsStream* stream
return ret; return ret;
} }
int SrsProtocol::__send_and_free_message(__SrsMessage* msg) int SrsProtocol::send_and_free_message(SrsMessage* msg)
{ {
srs_assert(msg); srs_assert(msg);
SrsAutoFree(__SrsMessage, msg, false); SrsAutoFree(SrsMessage, msg, false);
return __do_send_and_free_message(msg, NULL); return do_send_and_free_message(msg, NULL);
} }
int SrsProtocol::__send_and_free_packet(SrsPacket* packet, int stream_id) int SrsProtocol::send_and_free_packet(SrsPacket* packet, int stream_id)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -757,7 +757,7 @@ int SrsProtocol::__send_and_free_packet(SrsPacket* packet, int stream_id)
} }
// to message // to message
__SrsMessage* msg = new __SrsMessage(); SrsMessage* msg = new SrsMessage();
msg->payload = (int8_t*)payload; msg->payload = (int8_t*)payload;
msg->size = (int32_t)size; msg->size = (int32_t)size;
@ -767,14 +767,14 @@ int SrsProtocol::__send_and_free_packet(SrsPacket* packet, int stream_id)
msg->header.stream_id = stream_id; msg->header.stream_id = stream_id;
msg->header.perfer_cid = packet->get_perfer_cid(); msg->header.perfer_cid = packet->get_perfer_cid();
if ((ret = __do_send_and_free_message(msg, packet)) != ERROR_SUCCESS) { if ((ret = do_send_and_free_message(msg, packet)) != ERROR_SUCCESS) {
return ret; return ret;
} }
return ret; return ret;
} }
int SrsProtocol::__recv_interlaced_message(__SrsMessage** pmsg) int SrsProtocol::recv_interlaced_message(SrsMessage** pmsg)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -782,7 +782,7 @@ int SrsProtocol::__recv_interlaced_message(__SrsMessage** pmsg)
char fmt = 0; char fmt = 0;
int cid = 0; int cid = 0;
int bh_size = 0; int bh_size = 0;
if ((ret = __read_basic_header(fmt, cid, bh_size)) != ERROR_SUCCESS) { if ((ret = read_basic_header(fmt, cid, bh_size)) != ERROR_SUCCESS) {
if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) { if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) {
srs_error("read basic header failed. ret=%d", ret); srs_error("read basic header failed. ret=%d", ret);
} }
@ -805,16 +805,16 @@ int SrsProtocol::__recv_interlaced_message(__SrsMessage** pmsg)
} }
// get the cached chunk stream. // get the cached chunk stream.
__SrsChunkStream* chunk = NULL; SrsChunkStream* chunk = NULL;
if (__chunk_streams.find(cid) == __chunk_streams.end()) { if (chunk_streams.find(cid) == chunk_streams.end()) {
chunk = __chunk_streams[cid] = new __SrsChunkStream(cid); chunk = chunk_streams[cid] = new SrsChunkStream(cid);
// set the perfer cid of chunk, // set the perfer cid of chunk,
// which will copy to the message received. // which will copy to the message received.
chunk->header.perfer_cid = cid; chunk->header.perfer_cid = cid;
srs_verbose("cache new chunk stream: fmt=%d, cid=%d", fmt, cid); srs_verbose("cache new chunk stream: fmt=%d, cid=%d", fmt, cid);
} else { } else {
chunk = __chunk_streams[cid]; chunk = chunk_streams[cid];
srs_verbose("cached chunk stream: fmt=%d, cid=%d, size=%d, message(type=%d, size=%d, time=%"PRId64", sid=%d)", srs_verbose("cached chunk stream: fmt=%d, cid=%d, size=%d, message(type=%d, size=%d, time=%"PRId64", sid=%d)",
chunk->fmt, chunk->cid, (chunk->msg? chunk->msg->size : 0), chunk->header.message_type, chunk->header.payload_length, chunk->fmt, chunk->cid, (chunk->msg? chunk->msg->size : 0), chunk->header.message_type, chunk->header.payload_length,
chunk->header.timestamp, chunk->header.stream_id); chunk->header.timestamp, chunk->header.stream_id);
@ -822,7 +822,7 @@ int SrsProtocol::__recv_interlaced_message(__SrsMessage** pmsg)
// chunk stream message header // chunk stream message header
int mh_size = 0; int mh_size = 0;
if ((ret = __read_message_header(chunk, fmt, bh_size, mh_size)) != ERROR_SUCCESS) { if ((ret = read_message_header(chunk, fmt, bh_size, mh_size)) != ERROR_SUCCESS) {
if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) { if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) {
srs_error("read message header failed. ret=%d", ret); srs_error("read message header failed. ret=%d", ret);
} }
@ -834,9 +834,9 @@ int SrsProtocol::__recv_interlaced_message(__SrsMessage** pmsg)
chunk->header.payload_length, chunk->header.timestamp, chunk->header.stream_id); chunk->header.payload_length, chunk->header.timestamp, chunk->header.stream_id);
// read msg payload from chunk stream. // read msg payload from chunk stream.
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
int payload_size = 0; int payload_size = 0;
if ((ret = __read_message_payload(chunk, bh_size, mh_size, payload_size, &msg)) != ERROR_SUCCESS) { if ((ret = read_message_payload(chunk, bh_size, mh_size, payload_size, &msg)) != ERROR_SUCCESS) {
if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) { if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) {
srs_error("read message payload failed. ret=%d", ret); srs_error("read message payload failed. ret=%d", ret);
} }
@ -865,7 +865,7 @@ int SrsProtocol::__recv_interlaced_message(__SrsMessage** pmsg)
return ret; return ret;
} }
int SrsProtocol::__read_basic_header(char& fmt, int& cid, int& bh_size) int SrsProtocol::read_basic_header(char& fmt, int& cid, int& bh_size)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -923,7 +923,7 @@ int SrsProtocol::__read_basic_header(char& fmt, int& cid, int& bh_size)
return ret; return ret;
} }
int SrsProtocol::__read_message_header(__SrsChunkStream* chunk, char fmt, int bh_size, int& mh_size) int SrsProtocol::read_message_header(SrsChunkStream* chunk, char fmt, int bh_size, int& mh_size)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -971,7 +971,7 @@ int SrsProtocol::__read_message_header(__SrsChunkStream* chunk, char fmt, int bh
// create msg when new chunk stream start // create msg when new chunk stream start
if (!chunk->msg) { if (!chunk->msg) {
chunk->msg = new __SrsMessage(); chunk->msg = new SrsMessage();
srs_verbose("create message for new chunk, fmt=%d, cid=%d", fmt, chunk->cid); srs_verbose("create message for new chunk, fmt=%d, cid=%d", fmt, chunk->cid);
} }
@ -1157,7 +1157,7 @@ int SrsProtocol::__read_message_header(__SrsChunkStream* chunk, char fmt, int bh
return ret; return ret;
} }
int SrsProtocol::__read_message_payload(__SrsChunkStream* chunk, int bh_size, int mh_size, int& payload_size, __SrsMessage** pmsg) int SrsProtocol::read_message_payload(SrsChunkStream* chunk, int bh_size, int mh_size, int& payload_size, SrsMessage** pmsg)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -1222,7 +1222,7 @@ int SrsProtocol::__read_message_payload(__SrsChunkStream* chunk, int bh_size, in
return ret; return ret;
} }
int SrsProtocol::__on_recv_message(__SrsMessage* msg) int SrsProtocol::on_recv_message(SrsMessage* msg)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -1242,7 +1242,7 @@ int SrsProtocol::__on_recv_message(__SrsMessage* msg)
case RTMP_MSG_SetChunkSize: case RTMP_MSG_SetChunkSize:
case RTMP_MSG_UserControlMessage: case RTMP_MSG_UserControlMessage:
case RTMP_MSG_WindowAcknowledgementSize: case RTMP_MSG_WindowAcknowledgementSize:
if ((ret = __decode_message(msg, &packet)) != ERROR_SUCCESS) { if ((ret = decode_message(msg, &packet)) != ERROR_SUCCESS) {
srs_error("decode packet from message payload failed. ret=%d", ret); srs_error("decode packet from message payload failed. ret=%d", ret);
return ret; return ret;
} }
@ -1298,7 +1298,7 @@ int SrsProtocol::__on_recv_message(__SrsMessage* msg)
return ret; return ret;
} }
int SrsProtocol::__on_send_message(__SrsMessage* msg, SrsPacket* packet) int SrsProtocol::on_send_message(SrsMessage* msg, SrsPacket* packet)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -1353,7 +1353,7 @@ int SrsProtocol::response_acknowledgement_message()
SrsAcknowledgementPacket* pkt = new SrsAcknowledgementPacket(); SrsAcknowledgementPacket* pkt = new SrsAcknowledgementPacket();
in_ack_size.acked_size = pkt->sequence_number = skt->get_recv_bytes(); in_ack_size.acked_size = pkt->sequence_number = skt->get_recv_bytes();
if ((ret = __send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send acknowledgement failed. ret=%d", ret); srs_error("send acknowledgement failed. ret=%d", ret);
return ret; return ret;
} }
@ -1373,7 +1373,7 @@ int SrsProtocol::response_ping_message(int32_t timestamp)
pkt->event_type = SrcPCUCPingResponse; pkt->event_type = SrcPCUCPingResponse;
pkt->event_data = timestamp; pkt->event_data = timestamp;
if ((ret = __send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) { if ((ret = send_and_free_packet(pkt, 0)) != ERROR_SUCCESS) {
srs_error("send ping response failed. ret=%d", ret); srs_error("send ping response failed. ret=%d", ret);
return ret; return ret;
} }
@ -1474,7 +1474,7 @@ void SrsMessageHeader::initialize_video(int size, u_int32_t time, int stream)
stream_id = (int32_t)stream; stream_id = (int32_t)stream;
} }
__SrsChunkStream::__SrsChunkStream(int _cid) SrsChunkStream::SrsChunkStream(int _cid)
{ {
fmt = 0; fmt = 0;
cid = _cid; cid = _cid;
@ -1483,39 +1483,39 @@ __SrsChunkStream::__SrsChunkStream(int _cid)
msg_count = 0; msg_count = 0;
} }
__SrsChunkStream::~__SrsChunkStream() SrsChunkStream::~SrsChunkStream()
{ {
srs_freep(msg); srs_freep(msg);
} }
__SrsMessage::__SrsMessage() SrsMessage::SrsMessage()
{ {
payload = NULL; payload = NULL;
size = 0; size = 0;
} }
__SrsMessage::~__SrsMessage() SrsMessage::~SrsMessage()
{ {
} }
__SrsSharedPtrMessage::__SrsSharedPtr::__SrsSharedPtr() SrsSharedPtrMessage::__SrsSharedPtr::__SrsSharedPtr()
{ {
payload = NULL; payload = NULL;
size = 0; size = 0;
shared_count = 0; shared_count = 0;
} }
__SrsSharedPtrMessage::__SrsSharedPtr::~__SrsSharedPtr() SrsSharedPtrMessage::__SrsSharedPtr::~__SrsSharedPtr()
{ {
srs_freepa(payload); srs_freepa(payload);
} }
__SrsSharedPtrMessage::__SrsSharedPtrMessage() SrsSharedPtrMessage::SrsSharedPtrMessage()
{ {
ptr = NULL; ptr = NULL;
} }
__SrsSharedPtrMessage::~__SrsSharedPtrMessage() SrsSharedPtrMessage::~SrsSharedPtrMessage()
{ {
if (ptr) { if (ptr) {
if (ptr->shared_count == 0) { if (ptr->shared_count == 0) {
@ -1526,7 +1526,7 @@ __SrsSharedPtrMessage::~__SrsSharedPtrMessage()
} }
} }
int __SrsSharedPtrMessage::initialize(__SrsMessage* source) int SrsSharedPtrMessage::initialize(SrsMessage* source)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -1541,7 +1541,7 @@ int __SrsSharedPtrMessage::initialize(__SrsMessage* source)
return ret; return ret;
} }
int __SrsSharedPtrMessage::initialize(SrsMessageHeader* source, char* payload, int size) int SrsSharedPtrMessage::initialize(SrsMessageHeader* source, char* payload, int size)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
@ -1563,13 +1563,13 @@ int __SrsSharedPtrMessage::initialize(SrsMessageHeader* source, char* payload, i
ptr->payload = payload; ptr->payload = payload;
ptr->size = size; ptr->size = size;
__SrsMessage::payload = (int8_t*)ptr->payload; SrsMessage::payload = (int8_t*)ptr->payload;
__SrsMessage::size = ptr->size; SrsMessage::size = ptr->size;
return ret; return ret;
} }
__SrsSharedPtrMessage* __SrsSharedPtrMessage::copy() SrsSharedPtrMessage* SrsSharedPtrMessage::copy()
{ {
if (!ptr) { if (!ptr) {
srs_error("invoke initialize to initialize the ptr."); srs_error("invoke initialize to initialize the ptr.");
@ -1577,7 +1577,7 @@ __SrsSharedPtrMessage* __SrsSharedPtrMessage::copy()
return NULL; return NULL;
} }
__SrsSharedPtrMessage* copy = new __SrsSharedPtrMessage(); SrsSharedPtrMessage* copy = new SrsSharedPtrMessage();
copy->header = header; copy->header = header;
@ -1621,11 +1621,6 @@ int SrsPacket::get_message_type()
return 0; return 0;
} }
int SrsPacket::get_payload_length()
{
return get_size();
}
int SrsPacket::encode(int& psize, char*& ppayload) int SrsPacket::encode(int& psize, char*& ppayload)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;

69
trunk/src/rtmp/srs_protocol_rtmp_stack.hpp
View file

@ -43,8 +43,8 @@ class SrsStream;
class SrsAmf0Object; class SrsAmf0Object;
class SrsAmf0Any; class SrsAmf0Any;
class SrsMessageHeader; class SrsMessageHeader;
class __SrsMessage; class SrsMessage;
class __SrsChunkStream; class SrsChunkStream;
// the following is the timeout for rtmp protocol, // the following is the timeout for rtmp protocol,
// to avoid death connection. // to avoid death connection.
@ -109,8 +109,7 @@ private:
std::map<double, std::string> requests; std::map<double, std::string> requests;
// peer in // peer in
private: private:
// TODO: FIXME: rename to chunk_streams std::map<int, SrsChunkStream*> chunk_streams;
std::map<int, __SrsChunkStream*> __chunk_streams;
SrsStream* decode_stream; SrsStream* decode_stream;
SrsBuffer* buffer; SrsBuffer* buffer;
int32_t in_chunk_size; int32_t in_chunk_size;
@ -150,21 +149,21 @@ public:
* NULL for unknown packet but return success. * NULL for unknown packet but return success.
* never NULL if decode success. * never NULL if decode success.
*/ */
virtual int __recv_message(__SrsMessage** pmsg); virtual int recv_message(SrsMessage** pmsg);
/** /**
* decode bytes oriented RTMP message to RTMP packet, * decode bytes oriented RTMP message to RTMP packet,
* @param ppacket, output decoded packet, * @param ppacket, output decoded packet,
* always NULL if error, never NULL if success. * always NULL if error, never NULL if success.
* @return error when unknown packet, error when decode failed. * @return error when unknown packet, error when decode failed.
*/ */
virtual int __decode_message(__SrsMessage* msg, SrsPacket** ppacket); virtual int decode_message(SrsMessage* msg, SrsPacket** ppacket);
/** /**
* send the RTMP message and always free it. * send the RTMP message and always free it.
* user must never free or use the msg after this method, * user must never free or use the msg after this method,
* for it will always free the msg. * for it will always free the msg.
* @param msg, the msg to send out, never be NULL. * @param msg, the msg to send out, never be NULL.
*/ */
virtual int __send_and_free_message(__SrsMessage* msg); virtual int send_and_free_message(SrsMessage* msg);
/** /**
* send the RTMP packet and always free it. * send the RTMP packet and always free it.
* user must never free or use the packet after this method, * user must never free or use the packet after this method,
@ -172,50 +171,50 @@ public:
* @param packet, the packet to send out, never be NULL. * @param packet, the packet to send out, never be NULL.
* @param stream_id, the stream id of packet to send over, 0 for control message. * @param stream_id, the stream id of packet to send over, 0 for control message.
*/ */
virtual int __send_and_free_packet(SrsPacket* packet, int stream_id); virtual int send_and_free_packet(SrsPacket* packet, int stream_id);
private: private:
/** /**
* imp for __send_and_free_message * imp for send_and_free_message
* @param packet the packet of message, NULL for raw message. * @param packet the packet of message, NULL for raw message.
*/ */
virtual int __do_send_and_free_message(__SrsMessage* msg, SrsPacket* packet); virtual int do_send_and_free_message(SrsMessage* msg, SrsPacket* packet);
/** /**
* imp for __decode_message * imp for decode_message
*/ */
virtual int __do_decode_message(SrsMessageHeader& header, SrsStream* stream, SrsPacket** ppacket); virtual int do_decode_message(SrsMessageHeader& header, SrsStream* stream, SrsPacket** ppacket);
/** /**
* recv bytes oriented RTMP message from protocol stack. * recv bytes oriented RTMP message from protocol stack.
* return error if error occur and nerver set the pmsg, * return error if error occur and nerver set the pmsg,
* return success and pmsg set to NULL if no entire message got, * return success and pmsg set to NULL if no entire message got,
* return success and pmsg set to entire message if got one. * return success and pmsg set to entire message if got one.
*/ */
virtual int __recv_interlaced_message(__SrsMessage** pmsg); virtual int recv_interlaced_message(SrsMessage** pmsg);
/** /**
* read the chunk basic header(fmt, cid) from chunk stream. * read the chunk basic header(fmt, cid) from chunk stream.
* user can discovery a SrsChunkStream by cid. * user can discovery a SrsChunkStream by cid.
* @bh_size return the chunk basic header size, to remove the used bytes when finished. * @bh_size return the chunk basic header size, to remove the used bytes when finished.
*/ */
virtual int __read_basic_header(char& fmt, int& cid, int& bh_size); virtual int read_basic_header(char& fmt, int& cid, int& bh_size);
/** /**
* read the chunk message header(timestamp, payload_length, message_type, stream_id) * read the chunk message header(timestamp, payload_length, message_type, stream_id)
* from chunk stream and save to SrsChunkStream. * from chunk stream and save to SrsChunkStream.
* @mh_size return the chunk message header size, to remove the used bytes when finished. * @mh_size return the chunk message header size, to remove the used bytes when finished.
*/ */
virtual int __read_message_header(__SrsChunkStream* chunk, char fmt, int bh_size, int& mh_size); virtual int read_message_header(SrsChunkStream* chunk, char fmt, int bh_size, int& mh_size);
/** /**
* read the chunk payload, remove the used bytes in buffer, * read the chunk payload, remove the used bytes in buffer,
* if got entire message, set the pmsg. * if got entire message, set the pmsg.
* @payload_size read size in this roundtrip, generally a chunk size or left message size. * @payload_size read size in this roundtrip, generally a chunk size or left message size.
*/ */
virtual int __read_message_payload(__SrsChunkStream* chunk, int bh_size, int mh_size, int& payload_size, __SrsMessage** pmsg); virtual int read_message_payload(SrsChunkStream* chunk, int bh_size, int mh_size, int& payload_size, SrsMessage** pmsg);
/** /**
* when recv message, update the context. * when recv message, update the context.
*/ */
virtual int __on_recv_message(__SrsMessage* msg); virtual int on_recv_message(SrsMessage* msg);
/** /**
* when message sentout, update the context. * when message sentout, update the context.
*/ */
virtual int __on_send_message(__SrsMessage* msg, SrsPacket* packet); virtual int on_send_message(SrsMessage* msg, SrsPacket* packet);
private: private:
virtual int response_acknowledgement_message(); virtual int response_acknowledgement_message();
virtual int response_ping_message(int32_t timestamp); virtual int response_ping_message(int32_t timestamp);
@ -261,7 +260,6 @@ public:
* get the perfered cid(chunk stream id) which sendout over. * get the perfered cid(chunk stream id) which sendout over.
* set at decoding, and canbe used for directly send message, * set at decoding, and canbe used for directly send message,
* for example, dispatch to all connections. * for example, dispatch to all connections.
* @see: SrsSharedPtrMessage.SrsSharedPtr.perfer_cid
*/ */
int perfer_cid; int perfer_cid;
@ -289,7 +287,7 @@ public:
* incoming chunk stream maybe interlaced, * incoming chunk stream maybe interlaced,
* use the chunk stream to cache the input RTMP chunk streams. * use the chunk stream to cache the input RTMP chunk streams.
*/ */
class __SrsChunkStream class SrsChunkStream
{ {
public: public:
/** /**
@ -313,14 +311,14 @@ public:
/** /**
* partially read message. * partially read message.
*/ */
__SrsMessage* msg; SrsMessage* msg;
/** /**
* decoded msg count, to identify whether the chunk stream is fresh. * decoded msg count, to identify whether the chunk stream is fresh.
*/ */
int64_t msg_count; int64_t msg_count;
public: public:
__SrsChunkStream(int _cid); SrsChunkStream(int _cid);
virtual ~__SrsChunkStream(); virtual ~SrsChunkStream();
}; };
/** /**
@ -328,7 +326,7 @@ public:
* protcol always recv RTMP message, and can send RTMP message or RTMP packet. * protcol always recv RTMP message, and can send RTMP message or RTMP packet.
* the shared-ptr message is a special RTMP message, use ref-count for performance issue. * the shared-ptr message is a special RTMP message, use ref-count for performance issue.
*/ */
class __SrsMessage class SrsMessage
{ {
// 4.1. Message Header // 4.1. Message Header
public: public:
@ -344,8 +342,8 @@ public:
int32_t size; int32_t size;
int8_t* payload; int8_t* payload;
public: public:
__SrsMessage(); SrsMessage();
virtual ~__SrsMessage(); virtual ~SrsMessage();
}; };
/** /**
@ -353,7 +351,7 @@ public:
* for audio/video/data message that need less memory copy. * for audio/video/data message that need less memory copy.
* and only for output. * and only for output.
*/ */
class __SrsSharedPtrMessage : public __SrsMessage class SrsSharedPtrMessage : public SrsMessage
{ {
private: private:
struct __SrsSharedPtr struct __SrsSharedPtr
@ -367,15 +365,15 @@ private:
}; };
__SrsSharedPtr* ptr; __SrsSharedPtr* ptr;
public: public:
__SrsSharedPtrMessage(); SrsSharedPtrMessage();
virtual ~__SrsSharedPtrMessage(); virtual ~SrsSharedPtrMessage();
public: public:
/** /**
* set the shared payload. * set the shared payload.
* we will detach the payload of source, * we will detach the payload of source,
* so ensure donot use it before. * so ensure donot use it before.
*/ */
virtual int initialize(__SrsMessage* source); virtual int initialize(SrsMessage* source);
/** /**
* set the shared payload. * set the shared payload.
* use source header, and specified param payload. * use source header, and specified param payload.
@ -385,7 +383,7 @@ public:
/** /**
* copy current shared ptr message, use ref-count. * copy current shared ptr message, use ref-count.
*/ */
virtual __SrsSharedPtrMessage* copy(); virtual SrsSharedPtrMessage* copy();
}; };
/** /**
@ -423,7 +421,6 @@ public:
*/ */
public: public:
virtual int get_perfer_cid(); virtual int get_perfer_cid();
virtual int get_payload_length();
public: public:
/** /**
* subpacket must override to provide the right message type. * subpacket must override to provide the right message type.
@ -1133,7 +1130,7 @@ protected:
* if need to set timeout, use set timeout of SrsProtocol. * if need to set timeout, use set timeout of SrsProtocol.
*/ */
template<class T> template<class T>
int __srs_rtmp_expect_message(SrsProtocol* protocol, __SrsMessage** pmsg, T** ppacket) int __srs_rtmp_expect_message(SrsProtocol* protocol, SrsMessage** pmsg, T** ppacket)
{ {
*pmsg = NULL; *pmsg = NULL;
*ppacket = NULL; *ppacket = NULL;
@ -1141,15 +1138,15 @@ int __srs_rtmp_expect_message(SrsProtocol* protocol, __SrsMessage** pmsg, T** pp
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
while (true) { while (true) {
__SrsMessage* msg = NULL; SrsMessage* msg = NULL;
if ((ret = protocol->__recv_message(&msg)) != ERROR_SUCCESS) { if ((ret = protocol->recv_message(&msg)) != ERROR_SUCCESS) {
srs_error("recv message failed. ret=%d", ret); srs_error("recv message failed. ret=%d", ret);
return ret; return ret;
} }
srs_verbose("recv message success."); srs_verbose("recv message success.");
SrsPacket* packet = NULL; SrsPacket* packet = NULL;
if ((ret = protocol->__decode_message(msg, &packet)) != ERROR_SUCCESS) { if ((ret = protocol->decode_message(msg, &packet)) != ERROR_SUCCESS) {
srs_error("decode message failed. ret=%d", ret); srs_error("decode message failed. ret=%d", ret);
srs_freep(msg); srs_freep(msg);
return ret; return ret;