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

send set peer bandwidth packet.

This commit is contained in:
winlin 2013-10-20 15:59:26 +08:00
parent 2abfcee10f
commit 57dd58d557
9 changed files with 155 additions and 53 deletions

View file

@ -225,7 +225,7 @@ int srs_amf0_read_string(SrsStream* stream, std::string& value)
return ret; return ret;
} }
char marker = stream->read_char(); char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_String) { if (marker != RTMP_AMF0_String) {
ret = ERROR_RTMP_AMF0_DECODE; ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check string marker failed. " srs_error("amf0 check string marker failed. "
@ -248,7 +248,7 @@ int srs_amf0_read_boolean(SrsStream* stream, bool& value)
return ret; return ret;
} }
char marker = stream->read_char(); char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_Boolean) { if (marker != RTMP_AMF0_Boolean) {
ret = ERROR_RTMP_AMF0_DECODE; ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check bool marker failed. " srs_error("amf0 check bool marker failed. "
@ -264,7 +264,7 @@ int srs_amf0_read_boolean(SrsStream* stream, bool& value)
return ret; return ret;
} }
if (stream->read_char() == 0) { if (stream->read_1bytes() == 0) {
value = false; value = false;
} else { } else {
value = true; value = true;
@ -286,7 +286,7 @@ int srs_amf0_read_number(SrsStream* stream, double& value)
return ret; return ret;
} }
char marker = stream->read_char(); char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_Number) { if (marker != RTMP_AMF0_Number) {
ret = ERROR_RTMP_AMF0_DECODE; ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check number marker failed. " srs_error("amf0 check number marker failed. "
@ -321,7 +321,7 @@ int srs_amf0_read_any(SrsStream* stream, SrsAmf0Any*& value)
return ret; return ret;
} }
char marker = stream->read_char(); char marker = stream->read_1bytes();
srs_verbose("amf0 any marker success"); srs_verbose("amf0 any marker success");
// backward the 1byte marker. // backward the 1byte marker.
@ -393,7 +393,7 @@ int srs_amf0_read_object_eof(SrsStream* stream, SrsAmf0ObjectEOF*& value)
return ret; return ret;
} }
char marker = stream->read_char(); char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_ObjectEnd) { if (marker != RTMP_AMF0_ObjectEnd) {
ret = ERROR_RTMP_AMF0_DECODE; ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check object eof marker failed. " srs_error("amf0 check object eof marker failed. "
@ -420,7 +420,7 @@ int srs_amf0_read_object(SrsStream* stream, SrsAmf0Object*& value)
return ret; return ret;
} }
char marker = stream->read_char(); char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_Object) { if (marker != RTMP_AMF0_Object) {
ret = ERROR_RTMP_AMF0_DECODE; ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check object marker failed. " srs_error("amf0 check object marker failed. "

View file

@ -86,6 +86,12 @@ int SrsClient::do_cycle()
return ret; return ret;
} }
srs_verbose("set window acknowledgement size success"); srs_verbose("set window acknowledgement size success");
if ((ret = rtmp->set_peer_bandwidth(2.5 * 1000 * 1000, 2)) != ERROR_SUCCESS) {
srs_error("set peer bandwidth failed. ret=%d", ret);
return ret;
}
srs_verbose("set peer bandwidth success");
return ret; return ret;
} }

View file

@ -923,9 +923,16 @@ SrsPacket::~SrsPacket()
{ {
} }
int SrsPacket::decode(SrsStream* /*stream*/) int SrsPacket::decode(SrsStream* stream)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
srs_assert(stream != NULL);
ret = ERROR_SYSTEM_PACKET_INVALID;
srs_error("current packet is not support to decode. "
"paket=%s, ret=%d", get_class_name(), ret);
return ret; return ret;
} }
@ -988,7 +995,7 @@ int SrsPacket::encode_packet(SrsStream* stream)
srs_assert(stream != NULL); srs_assert(stream != NULL);
ret = ERROR_SYSTEM_PACKET_INVALID; ret = ERROR_SYSTEM_PACKET_INVALID;
srs_error("current packet is not support to sendout. " srs_error("current packet is not support to encode. "
"paket=%s, ret=%d", get_class_name(), ret); "paket=%s, ret=%d", get_class_name(), ret);
return ret; return ret;
@ -1008,10 +1015,6 @@ SrsConnectAppPacket::~SrsConnectAppPacket()
int SrsConnectAppPacket::decode(SrsStream* stream) int SrsConnectAppPacket::decode(SrsStream* stream)
{ {
int ret = ERROR_SUCCESS; int ret = ERROR_SUCCESS;
if ((ret = super::decode(stream)) != ERROR_SUCCESS) {
return ret;
}
if ((ret = srs_amf0_read_string(stream, command_name)) != ERROR_SUCCESS) { if ((ret = srs_amf0_read_string(stream, command_name)) != ERROR_SUCCESS) {
srs_error("amf0 decode connect command_name failed. ret=%d", ret); srs_error("amf0 decode connect command_name failed. ret=%d", ret);
@ -1059,26 +1062,6 @@ SrsSetWindowAckSizePacket::~SrsSetWindowAckSizePacket()
{ {
} }
int SrsSetWindowAckSizePacket::decode(SrsStream* stream)
{
int ret = ERROR_SUCCESS;
if ((ret = super::decode(stream)) != ERROR_SUCCESS) {
return ret;
}
if (!stream->require(4)) {
ret = ERROR_RTMP_MESSAGE_DECODE;
srs_error("set window ack size failed. ret=%d", ret);
return ret;
}
ackowledgement_window_size = stream->read_4bytes();
srs_info("decode window ack size success. ack_size=%d", ackowledgement_window_size);
return ret;
}
int SrsSetWindowAckSizePacket::get_perfer_cid() int SrsSetWindowAckSizePacket::get_perfer_cid()
{ {
return RTMP_CID_ProtocolControl; return RTMP_CID_ProtocolControl;
@ -1112,3 +1095,47 @@ int SrsSetWindowAckSizePacket::encode_packet(SrsStream* stream)
return ret; return ret;
} }
SrsSetPeerBandwidthPacket::SrsSetPeerBandwidthPacket()
{
bandwidth = 0;
type = 2;
}
SrsSetPeerBandwidthPacket::~SrsSetPeerBandwidthPacket()
{
}
int SrsSetPeerBandwidthPacket::get_perfer_cid()
{
return RTMP_CID_ProtocolControl;
}
int SrsSetPeerBandwidthPacket::get_message_type()
{
return RTMP_MSG_SetPeerBandwidth;
}
int SrsSetPeerBandwidthPacket::get_size()
{
return 5;
}
int SrsSetPeerBandwidthPacket::encode_packet(SrsStream* stream)
{
int ret = ERROR_SUCCESS;
if (!stream->require(5)) {
ret = ERROR_RTMP_MESSAGE_ENCODE;
srs_error("encode set bandwidth packet failed. ret=%d", ret);
return ret;
}
stream->write_4bytes(bandwidth);
stream->write_1bytes(type);
srs_verbose("encode set bandwidth packet "
"success. bandwidth=%d, type=%d", bandwidth, type);
return ret;
}

View file

@ -256,10 +256,22 @@ public:
*/ */
class SrsPacket class SrsPacket
{ {
protected:
/**
* subpacket must override to provide the right class name.
*/
virtual const char* get_class_name()
{
return CLASS_NAME_STRING(SrsPacket);
}
public: public:
SrsPacket(); SrsPacket();
virtual ~SrsPacket(); virtual ~SrsPacket();
public: public:
/**
* subpacket must override to decode packet from stream.
* @remark never invoke the super.decode, it always failed.
*/
virtual int decode(SrsStream* stream); virtual int decode(SrsStream* stream);
public: public:
virtual int get_perfer_cid(); virtual int get_perfer_cid();
@ -283,16 +295,9 @@ protected:
virtual int get_size(); virtual int get_size();
/** /**
* subpacket can override to encode the payload to stream. * subpacket can override to encode the payload to stream.
* @remark never invoke the super.encode_packet, it always failed.
*/ */
virtual int encode_packet(SrsStream* stream); virtual int encode_packet(SrsStream* stream);
protected:
/**
* subpacket must override to provide the right class name.
*/
virtual const char* get_class_name()
{
return CLASS_NAME_STRING(SrsPacket);
}
}; };
/** /**
@ -304,6 +309,11 @@ class SrsConnectAppPacket : public SrsPacket
{ {
private: private:
typedef SrsPacket super; typedef SrsPacket super;
protected:
virtual const char* get_class_name()
{
return CLASS_NAME_STRING(SrsConnectAppPacket);
}
public: public:
std::string command_name; std::string command_name;
double transaction_id; double transaction_id;
@ -313,11 +323,6 @@ public:
virtual ~SrsConnectAppPacket(); virtual ~SrsConnectAppPacket();
public: public:
virtual int decode(SrsStream* stream); virtual int decode(SrsStream* stream);
protected:
virtual const char* get_class_name()
{
return CLASS_NAME_STRING(SrsConnectAppPacket);
}
}; };
/** /**
@ -329,13 +334,16 @@ class SrsSetWindowAckSizePacket : public SrsPacket
{ {
private: private:
typedef SrsPacket super; typedef SrsPacket super;
protected:
virtual const char* get_class_name()
{
return CLASS_NAME_STRING(SrsSetWindowAckSizePacket);
}
public: public:
int32_t ackowledgement_window_size; int32_t ackowledgement_window_size;
public: public:
SrsSetWindowAckSizePacket(); SrsSetWindowAckSizePacket();
virtual ~SrsSetWindowAckSizePacket(); virtual ~SrsSetWindowAckSizePacket();
public:
virtual int decode(SrsStream* stream);
public: public:
virtual int get_perfer_cid(); virtual int get_perfer_cid();
public: public:
@ -343,11 +351,35 @@ public:
protected: protected:
virtual int get_size(); virtual int get_size();
virtual int encode_packet(SrsStream* stream); virtual int encode_packet(SrsStream* stream);
};
/**
* 5.6. Set Peer Bandwidth (6)
* The client or the server sends this message to update the output
* bandwidth of the peer.
*/
class SrsSetPeerBandwidthPacket : public SrsPacket
{
private:
typedef SrsPacket super;
protected: protected:
virtual const char* get_class_name() virtual const char* get_class_name()
{ {
return CLASS_NAME_STRING(SrsSetWindowAckSizePacket); return CLASS_NAME_STRING(SrsSetPeerBandwidthPacket);
} }
public:
int32_t bandwidth;
int8_t type;
public:
SrsSetPeerBandwidthPacket();
virtual ~SrsSetPeerBandwidthPacket();
public:
virtual int get_perfer_cid();
public:
virtual int get_message_type();
protected:
virtual int get_size();
virtual int encode_packet(SrsStream* stream);
}; };
/** /**

View file

@ -184,3 +184,24 @@ int SrsRtmp::set_window_ack_size(int ack_size)
return ret; return ret;
} }
int SrsRtmp::set_peer_bandwidth(int bandwidth, int type)
{
int ret = ERROR_SUCCESS;
SrsMessage* msg = new SrsMessage();
SrsSetPeerBandwidthPacket* pkt = new SrsSetPeerBandwidthPacket();
pkt->bandwidth = bandwidth;
pkt->type = type;
msg->set_packet(pkt);
if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
srs_error("send set bandwidth message failed. ret=%d", ret);
return ret;
}
srs_info("send set bandwidth message "
"success. bandwidth=%d, type=%d", bandwidth, type);
return ret;
}

View file

@ -74,6 +74,11 @@ public:
virtual int handshake(); virtual int handshake();
virtual int connect_app(SrsRequest* req); virtual int connect_app(SrsRequest* req);
virtual int set_window_ack_size(int ack_size); virtual int set_window_ack_size(int ack_size);
/**
* @type: The sender can mark this message hard (0), soft (1), or dynamic (2)
* using the Limit type field.
*/
virtual int set_peer_bandwidth(int bandwidth, int type);
}; };
#endif #endif

View file

@ -78,11 +78,11 @@ void SrsStream::skip(int size)
p += size; p += size;
} }
char SrsStream::read_char() int8_t SrsStream::read_1bytes()
{ {
srs_assert(require(1)); srs_assert(require(1));
return *p++; return (int8_t)*p++;
} }
int16_t SrsStream::read_2bytes() int16_t SrsStream::read_2bytes()
@ -152,3 +152,10 @@ void SrsStream::write_4bytes(int32_t value)
*p++ = pp[0]; *p++ = pp[0];
} }
void SrsStream::write_1bytes(int8_t value)
{
srs_assert(require(1));
*p++ = value;
}

View file

@ -74,7 +74,7 @@ public:
/** /**
* get 1bytes char from stream. * get 1bytes char from stream.
*/ */
virtual char read_char(); virtual int8_t read_1bytes();
/** /**
* get 2bytes int from stream. * get 2bytes int from stream.
*/ */
@ -96,6 +96,10 @@ public:
* write 4bytes int to stream. * write 4bytes int to stream.
*/ */
virtual void write_4bytes(int32_t value); virtual void write_4bytes(int32_t value);
/**
* write 1bytes char to stream.
*/
virtual void write_1bytes(int8_t value);
}; };
#endif #endif

View file

@ -34,7 +34,7 @@ int main(int /*argc*/, char** /*argv*/){
return ret; return ret;
} }
if ((ret = server.listen(19350)) != ERROR_SUCCESS) { if ((ret = server.listen(1935)) != ERROR_SUCCESS) {
return ret; return ret;
} }