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

merge wenjie. fix jw/flower player pause bug, which send closeStream actually.

Browse source
This commit is contained in:
winlin 2014-01-11 19:55:55 +08:00
parent 4148f5231d
commit 574536d9ed
9 changed files with 170 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

52
trunk/src/core/srs_core_client.cpp
View file

@ -168,6 +168,41 @@ int SrsClient::service_cycle()
}
srs_verbose("on_bw_done success");
while (true) {
ret = stream_service_cycle();
// stream service must terminated with error, never success.
srs_assert(ret != ERROR_SUCCESS);
// when not system control error, fatal error, return.
if (!srs_is_system_control_error(ret)) {
srs_error("stream service cycle failed. ret=%d", ret);
return ret;
}
// for "some" system control error,
// logical accept and retry stream service.
if (ret == ERROR_CONTROL_RTMP_CLOSE) {
// set timeout to a larger value, for user paused.
rtmp->set_recv_timeout(SRS_PAUSED_SEND_TIMEOUT_US);
rtmp->set_send_timeout(SRS_PAUSED_SEND_TIMEOUT_US);
srs_trace("control message(close) accept, retry stream service.");
continue;
}
// for other system control message, fatal error.
srs_error("control message(%d) reject as error. ret=%d", ret, ret);
return ret;
}
return ret;
}
int SrsClient::stream_service_cycle()
{
int ret = ERROR_SUCCESS;
SrsClientType type;
if ((ret = rtmp->identify_client(res->stream_id, type, req->stream)) != ERROR_SUCCESS) {
srs_error("identify client failed. ret=%d", ret);
@ -175,7 +210,12 @@ int SrsClient::service_cycle()
}
req->strip();
srs_trace("identify client success. type=%d, stream_name=%s", type, req->stream.c_str());
// client is identified, set the timeout to service timeout.
rtmp->set_recv_timeout(SRS_RECV_TIMEOUT_US);
rtmp->set_send_timeout(SRS_SEND_TIMEOUT_US);
// set timeout to larger.
int chunk_size = config->get_chunk_size(req->vhost);
if ((ret = rtmp->set_chunk_size(chunk_size)) != ERROR_SUCCESS) {
srs_error("set chunk_size=%d failed. ret=%d", chunk_size, ret);
@ -341,7 +381,9 @@ int SrsClient::playing(SrsSource* source)
return ret;
}
if ((ret = process_play_control_msg(consumer, msg)) != ERROR_SUCCESS) {
srs_error("process play control message failed. ret=%d", ret);
if (!srs_is_system_control_error(ret)) {
srs_error("process play control message failed. ret=%d", ret);
}
return ret;
}
}
@ -555,6 +597,13 @@ int SrsClient::process_play_control_msg(SrsConsumer* consumer, SrsCommonMessage*
}
srs_info("decode the amf0/amf3 command packet success.");
SrsCloseStreamPacket* close = dynamic_cast<SrsCloseStreamPacket*>(msg->get_packet());
if (close) {
ret = ERROR_CONTROL_RTMP_CLOSE;
srs_trace("system control message: rtmp close stream. ret=%d", ret);
return ret;
}
SrsPausePacket* pause = dynamic_cast<SrsPausePacket*>(msg->get_packet());
if (!pause) {
srs_info("ignore all amf0/amf3 command except pause.");
@ -700,4 +749,3 @@ void SrsClient::on_stop()
}
#endif
}

4
trunk/src/core/srs_core_client.hpp
View file

@ -71,6 +71,8 @@ public:
private:
// when valid and connected to vhost/app, service the client.
virtual int service_cycle();
// stream(play/publish) service cycle, identify client first.
virtual int stream_service_cycle();
virtual int check_vhost();
virtual int playing(SrsSource* source);
virtual int publish(SrsSource* source, bool is_fmle);
@ -86,4 +88,4 @@ private:
virtual void on_stop();
};
#endif
#endif

6
trunk/src/core/srs_core_error.cpp
View file

@ -22,3 +22,9 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include <srs_core_error.hpp>
bool srs_is_system_control_error(int error_code)
{
return error_code == ERROR_CONTROL_RTMP_CLOSE;
}

12
trunk/src/core/srs_core_error.hpp
View file

@ -147,4 +147,14 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#define ERROR_HTTP_DATA_INVLIAD 801
#define ERROR_HTTP_PARSE_HEADER 802
#endif
// system control message,
// not an error, but special control logic.
// sys ctl: rtmp close stream, support replay.
#define ERROR_CONTROL_RTMP_CLOSE 900
/**
* whether the error code is an system control error.
*/
extern bool srs_is_system_control_error(int error_code);
#endif

41
trunk/src/core/srs_core_protocol.cpp
View file

@ -196,6 +196,7 @@ messages.
*/
#define RTMP_AMF0_COMMAND_CONNECT "connect"
#define RTMP_AMF0_COMMAND_CREATE_STREAM "createStream"
#define RTMP_AMF0_COMMAND_CLOSE_STREAM "closeStream"
#define RTMP_AMF0_COMMAND_PLAY "play"
#define RTMP_AMF0_COMMAND_PAUSE "pause"
#define RTMP_AMF0_COMMAND_ON_BW_DONE "onBWDone"
@ -1363,6 +1364,10 @@ int SrsCommonMessage::decode_packet(SrsProtocol* protocol)
srs_info("decode the AMF0/AMF3 band width check message.");
packet = new SrsBandwidthPacket();
return packet->decode(stream);
} else if (command == RTMP_AMF0_COMMAND_CLOSE_STREAM) {
srs_info("decode the AMF0/AMF3 closeStream message.");
packet = new SrsCloseStreamPacket();
return packet->decode(stream);
}
// default packet to drop message.
@ -2064,6 +2069,41 @@ int SrsCreateStreamResPacket::encode_packet(SrsStream* stream)
return ret;
}
SrsCloseStreamPacket::SrsCloseStreamPacket()
{
command_name = RTMP_AMF0_COMMAND_CLOSE_STREAM;
transaction_id = 0;
command_object = new SrsAmf0Null();
}
SrsCloseStreamPacket::~SrsCloseStreamPacket()
{
srs_freep(command_object);
}
int SrsCloseStreamPacket::decode(SrsStream* stream)
{
int ret = ERROR_SUCCESS;
if ((ret = srs_amf0_read_string(stream, command_name)) != ERROR_SUCCESS) {
srs_error("amf0 decode closeStream command_name failed. ret=%d", ret);
return ret;
}
if ((ret = srs_amf0_read_number(stream, transaction_id)) != ERROR_SUCCESS) {
srs_error("amf0 decode closeStream transaction_id failed. ret=%d", ret);
return ret;
}
if ((ret = srs_amf0_read_null(stream)) != ERROR_SUCCESS) {
srs_error("amf0 decode closeStream command_object failed. ret=%d", ret);
return ret;
}
srs_info("amf0 decode closeStream packet success");
return ret;
}
SrsFMLEStartPacket::SrsFMLEStartPacket()
{
command_name = RTMP_AMF0_COMMAND_CREATE_STREAM;
@ -3325,4 +3365,3 @@ int SrsUserControlPacket::encode_packet(SrsStream* stream)
return ret;
}

32
trunk/src/core/srs_core_protocol.hpp
View file

@ -56,6 +56,14 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
// if timeout, close the connection.
#define SRS_RECV_TIMEOUT_US 30*1000*1000L
// the timeout to wait client data, when client paused
// if timeout, close the connection.
#define SRS_PAUSED_SEND_TIMEOUT_US 30*60*1000*1000L
// the timeout to send data to client, when client paused
// if timeout, close the connection.
#define SRS_PAUSED_RECV_TIMEOUT_US 30*60*1000*1000L
// when stream is busy, for example, streaming is already
// publishing, when a new client to request to publish,
// sleep a while and close the connection.
@ -625,6 +633,28 @@ protected:
virtual int get_size();
virtual int encode_packet(SrsStream* stream);
};
/**
* client close stream packet.
*/
class SrsCloseStreamPacket : public SrsPacket
{
private:
typedef SrsPacket super;
protected:
virtual const char* get_class_name()
{
return CLASS_NAME_STRING(SrsCloseStreamPacket);
}
public:
std::string command_name;
double transaction_id;
SrsAmf0Null* command_object;
public:
SrsCloseStreamPacket();
virtual ~SrsCloseStreamPacket();
public:
virtual int decode(SrsStream* stream);
};
/**
* FMLE start publish: ReleaseStream/PublishStream
@ -1223,4 +1253,4 @@ int srs_rtmp_expect_message(SrsProtocol* protocol, SrsCommonMessage** pmsg, T**
return ret;
}
#endif
#endif

37
trunk/src/core/srs_core_rtmp.cpp
View file

@ -171,7 +171,7 @@ void SrsRequest::strip()
trim(stream, "/ \n\r\t");
}
std::string& SrsRequest::trim(string& str, string chs)
string& SrsRequest::trim(string& str, string chs)
{
for (int i = 0; i < (int)chs.length(); i++) {
char ch = chs.at(i);
@ -695,7 +695,7 @@ int SrsRtmp::on_bw_done()
return ret;
}
int SrsRtmp::identify_client(int stream_id, SrsClientType& type, std::string& stream_name)
int SrsRtmp::identify_client(int stream_id, SrsClientType& type, string& stream_name)
{
type = SrsClientUnknown;
int ret = ERROR_SUCCESS;
@ -723,13 +723,15 @@ int SrsRtmp::identify_client(int stream_id, SrsClientType& type, std::string& st
SrsPacket* pkt = msg->get_packet();
if (dynamic_cast<SrsCreateStreamPacket*>(pkt)) {
srs_info("identify client by create stream, play or flash publish.");
return identify_create_stream_client(
dynamic_cast<SrsCreateStreamPacket*>(pkt), stream_id, type, stream_name);
return identify_create_stream_client(dynamic_cast<SrsCreateStreamPacket*>(pkt), stream_id, type, stream_name);
}
if (dynamic_cast<SrsFMLEStartPacket*>(pkt)) {
srs_info("identify client by releaseStream, fmle publish.");
return identify_fmle_publish_client(
dynamic_cast<SrsFMLEStartPacket*>(pkt), type, stream_name);
return identify_fmle_publish_client(dynamic_cast<SrsFMLEStartPacket*>(pkt), type, stream_name);
}
if (dynamic_cast<SrsPlayPacket*>(pkt)) {
srs_info("level0 identify client by play.");
return identify_play_client(dynamic_cast<SrsPlayPacket*>(pkt), type, stream_name);
}
srs_trace("ignore AMF0/AMF3 command message.");
@ -1165,16 +1167,12 @@ int SrsRtmp::identify_create_stream_client(SrsCreateStreamPacket* req, int strea
SrsPacket* pkt = msg->get_packet();
if (dynamic_cast<SrsPlayPacket*>(pkt)) {
SrsPlayPacket* play = dynamic_cast<SrsPlayPacket*>(pkt);
type = SrsClientPlay;
stream_name = play->stream_name;
srs_trace("identity client type=play, stream_name=%s", stream_name.c_str());
return ret;
srs_info("level1 identify client by play.");
return identify_play_client(dynamic_cast<SrsPlayPacket*>(pkt), type, stream_name);
}
if (dynamic_cast<SrsPublishPacket*>(pkt)) {
srs_info("identify client by publish, falsh publish.");
return identify_flash_publish_client(
dynamic_cast<SrsPublishPacket*>(pkt), type, stream_name);
return identify_flash_publish_client(dynamic_cast<SrsPublishPacket*>(pkt), type, stream_name);
}
srs_trace("ignore AMF0/AMF3 command message.");
@ -1216,3 +1214,16 @@ int SrsRtmp::identify_flash_publish_client(SrsPublishPacket* req, SrsClientType&
return ret;
}
int SrsRtmp::identify_play_client(SrsPlayPacket* req, SrsClientType& type, string& stream_name)
{
int ret = ERROR_SUCCESS;
type = SrsClientPlay;
stream_name = req->stream_name;
srs_trace("identity client type=play, stream_name=%s", stream_name.c_str());
return ret;
}

5
trunk/src/core/srs_core_rtmp.hpp
View file

@ -40,6 +40,7 @@ class SrsFMLEStartPacket;
class SrsPublishPacket;
class SrsSharedPtrMessage;
class SrsOnMetaDataPacket;
class SrsPlayPacket;
/**
* the original request from client.
@ -226,6 +227,8 @@ private:
virtual int identify_create_stream_client(SrsCreateStreamPacket* req, int stream_id, SrsClientType& type, std::string& stream_name);
virtual int identify_fmle_publish_client(SrsFMLEStartPacket* req, SrsClientType& type, std::string& stream_name);
virtual int identify_flash_publish_client(SrsPublishPacket* req, SrsClientType& type, std::string& stream_name);
private:
virtual int identify_play_client(SrsPlayPacket* req, SrsClientType& type, std::string& stream_name);
};
#endif
#endif