胡斌 / srs

转到一个项目
Toggle navigation Toggle navigation pinning
winlin
2c4c91d8 1 parent 145eb7bf

convert format to unix

内嵌 并排对比
正在显示 9 个修改的文件 包含 2419 行增加 和 2419 行删除

要显示太多修改。
重新载入完整差异 差异文件 邮件补丁

为保证性能只显示 9 of 9+ 个文件。

/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include <srs_core.hpp>
#include <sys/time.h>
static int64_t _srs_system_time_us_cache = 0;
int64_t srs_get_system_time_ms()
{
return _srs_system_time_us_cache / 1000;
}
void srs_update_system_time_ms()
{
timeval now;
gettimeofday(&now, NULL);
// we must convert the tv_sec/tv_usec to int64_t.
_srs_system_time_us_cache = now.tv_sec * 1000 * 1000 + now.tv_usec;
_srs_system_time_us_cache = srs_max(0, _srs_system_time_us_cache);
}
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include <srs_core.hpp>
#include <sys/time.h>
static int64_t _srs_system_time_us_cache = 0;
int64_t srs_get_system_time_ms()
{
return _srs_system_time_us_cache / 1000;
}
void srs_update_system_time_ms()
{
timeval now;
gettimeofday(&now, NULL);
// we must convert the tv_sec/tv_usec to int64_t.
_srs_system_time_us_cache = now.tv_sec * 1000 * 1000 + now.tv_usec;
_srs_system_time_us_cache = srs_max(0, _srs_system_time_us_cache);
}
... ...
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#ifndef SRS_CORE_HPP
#define SRS_CORE_HPP
/*
#include <srs_core.hpp>
*/
/**
* the core provides the common defined macros, utilities,
* user must include the srs_core.hpp before any header, or maybe
* build failed.
*/
// for int64_t print using PRId64 format.
#ifndef __STDC_FORMAT_MACROS
#define __STDC_FORMAT_MACROS
#endif
#include <inttypes.h>
#include <assert.h>
#define srs_assert(expression) assert(expression)
#include <stddef.h>
#include <sys/types.h>
// free the p and set to NULL.
// p must be a T*.
#define srs_freep(p) \
if (p) { \
delete p; \
p = NULL; \
} \
(void)0
// free the p which represents a array
#define srs_freepa(p) \
if (p) { \
delete[] p; \
p = NULL; \
} \
(void)0
// current release version
#define RTMP_SIG_SRS_VERSION "0.4.0"
// server info.
#define RTMP_SIG_SRS_KEY "srs"
#define RTMP_SIG_SRS_ROLE "origin server"
#define RTMP_SIG_SRS_NAME RTMP_SIG_SRS_KEY"(simple rtmp server)"
#define RTMP_SIG_SRS_URL "https://"RTMP_SIG_SRS_URL_SHORT
#define RTMP_SIG_SRS_URL_SHORT "github.com/winlinvip/simple-rtmp-server"
#define RTMP_SIG_SRS_WEB "http://blog.csdn.net/win_lin"
#define RTMP_SIG_SRS_EMAIL "winterserver@126.com"
#define RTMP_SIG_SRS_LICENSE "The MIT License (MIT)"
#define RTMP_SIG_SRS_COPYRIGHT "Copyright (c) 2013 winlin"
// compare
#define srs_min(a, b) (((a) < (b))? (a) : (b))
#define srs_max(a, b) (((a) < (b))? (b) : (a))
// get current system time in ms, use cache to avoid performance problem
extern int64_t srs_get_system_time_ms();
// the deamon st-thread will update it.
extern void srs_update_system_time_ms();
// signal defines.
#define SIGNAL_RELOAD SIGHUP
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#ifndef SRS_CORE_HPP
#define SRS_CORE_HPP
/*
#include <srs_core.hpp>
*/
/**
* the core provides the common defined macros, utilities,
* user must include the srs_core.hpp before any header, or maybe
* build failed.
*/
// for int64_t print using PRId64 format.
#ifndef __STDC_FORMAT_MACROS
#define __STDC_FORMAT_MACROS
#endif
#include <inttypes.h>
#include <assert.h>
#define srs_assert(expression) assert(expression)
#include <stddef.h>
#include <sys/types.h>
// free the p and set to NULL.
// p must be a T*.
#define srs_freep(p) \
if (p) { \
delete p; \
p = NULL; \
} \
(void)0
// free the p which represents a array
#define srs_freepa(p) \
if (p) { \
delete[] p; \
p = NULL; \
} \
(void)0
// current release version
#define RTMP_SIG_SRS_VERSION "0.4.0"
// server info.
#define RTMP_SIG_SRS_KEY "srs"
#define RTMP_SIG_SRS_ROLE "origin server"
#define RTMP_SIG_SRS_NAME RTMP_SIG_SRS_KEY"(simple rtmp server)"
#define RTMP_SIG_SRS_URL "https://"RTMP_SIG_SRS_URL_SHORT
#define RTMP_SIG_SRS_URL_SHORT "github.com/winlinvip/simple-rtmp-server"
#define RTMP_SIG_SRS_WEB "http://blog.csdn.net/win_lin"
#define RTMP_SIG_SRS_EMAIL "winterserver@126.com"
#define RTMP_SIG_SRS_LICENSE "The MIT License (MIT)"
#define RTMP_SIG_SRS_COPYRIGHT "Copyright (c) 2013 winlin"
// compare
#define srs_min(a, b) (((a) < (b))? (a) : (b))
#define srs_max(a, b) (((a) < (b))? (b) : (a))
// get current system time in ms, use cache to avoid performance problem
extern int64_t srs_get_system_time_ms();
// the deamon st-thread will update it.
extern void srs_update_system_time_ms();
// signal defines.
#define SIGNAL_RELOAD SIGHUP
#endif
\ No newline at end of file
... ...
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include <srs_core_amf0.hpp>
#include <utility>
#include <srs_core_log.hpp>
#include <srs_core_error.hpp>
#include <srs_core_stream.hpp>
// AMF0 marker
#define RTMP_AMF0_Number 0x00
#define RTMP_AMF0_Boolean 0x01
#define RTMP_AMF0_String 0x02
#define RTMP_AMF0_Object 0x03
#define RTMP_AMF0_MovieClip 0x04 // reserved, not supported
#define RTMP_AMF0_Null 0x05
#define RTMP_AMF0_Undefined 0x06
#define RTMP_AMF0_Reference 0x07
#define RTMP_AMF0_EcmaArray 0x08
#define RTMP_AMF0_ObjectEnd 0x09
#define RTMP_AMF0_StrictArray 0x0A
#define RTMP_AMF0_Date 0x0B
#define RTMP_AMF0_LongString 0x0C
#define RTMP_AMF0_UnSupported 0x0D
#define RTMP_AMF0_RecordSet 0x0E // reserved, not supported
#define RTMP_AMF0_XmlDocument 0x0F
#define RTMP_AMF0_TypedObject 0x10
// AVM+ object is the AMF3 object.
#define RTMP_AMF0_AVMplusObject 0x11
// origin array whos data takes the same form as LengthValueBytes
#define RTMP_AMF0_OriginStrictArray 0x20
// User defined
#define RTMP_AMF0_Invalid 0x3F
int srs_amf0_get_object_eof_size();
int srs_amf0_get_any_size(SrsAmf0Any* value);
int srs_amf0_read_object_eof(SrsStream* stream, SrsAmf0ObjectEOF*&);
int srs_amf0_write_object_eof(SrsStream* stream, SrsAmf0ObjectEOF*);
int srs_amf0_read_any(SrsStream* stream, SrsAmf0Any*& value);
int srs_amf0_write_any(SrsStream* stream, SrsAmf0Any* value);
SrsAmf0Any::SrsAmf0Any()
{
marker = RTMP_AMF0_Invalid;
}
SrsAmf0Any::~SrsAmf0Any()
{
}
bool SrsAmf0Any::is_string()
{
return marker == RTMP_AMF0_String;
}
bool SrsAmf0Any::is_boolean()
{
return marker == RTMP_AMF0_Boolean;
}
bool SrsAmf0Any::is_number()
{
return marker == RTMP_AMF0_Number;
}
bool SrsAmf0Any::is_null()
{
return marker == RTMP_AMF0_Null;
}
bool SrsAmf0Any::is_undefined()
{
return marker == RTMP_AMF0_Undefined;
}
bool SrsAmf0Any::is_object()
{
return marker == RTMP_AMF0_Object;
}
bool SrsAmf0Any::is_ecma_array()
{
return marker == RTMP_AMF0_EcmaArray;
}
bool SrsAmf0Any::is_object_eof()
{
return marker == RTMP_AMF0_ObjectEnd;
}
SrsAmf0String::SrsAmf0String(const char* _value)
{
marker = RTMP_AMF0_String;
if (_value) {
value = _value;
}
}
SrsAmf0String::~SrsAmf0String()
{
}
SrsAmf0Boolean::SrsAmf0Boolean(bool _value)
{
marker = RTMP_AMF0_Boolean;
value = _value;
}
SrsAmf0Boolean::~SrsAmf0Boolean()
{
}
SrsAmf0Number::SrsAmf0Number(double _value)
{
marker = RTMP_AMF0_Number;
value = _value;
}
SrsAmf0Number::~SrsAmf0Number()
{
}
SrsAmf0Null::SrsAmf0Null()
{
marker = RTMP_AMF0_Null;
}
SrsAmf0Null::~SrsAmf0Null()
{
}
SrsAmf0Undefined::SrsAmf0Undefined()
{
marker = RTMP_AMF0_Undefined;
}
SrsAmf0Undefined::~SrsAmf0Undefined()
{
}
SrsAmf0ObjectEOF::SrsAmf0ObjectEOF()
{
marker = RTMP_AMF0_ObjectEnd;
utf8_empty = 0x00;
}
SrsAmf0ObjectEOF::~SrsAmf0ObjectEOF()
{
}
SrsUnSortedHashtable::SrsUnSortedHashtable()
{
}
SrsUnSortedHashtable::~SrsUnSortedHashtable()
{
std::vector<SrsObjectPropertyType>::iterator it;
for (it = properties.begin(); it != properties.end(); ++it) {
SrsObjectPropertyType& elem = *it;
SrsAmf0Any* any = elem.second;
srs_freep(any);
}
properties.clear();
}
int SrsUnSortedHashtable::size()
{
return (int)properties.size();
}
void SrsUnSortedHashtable::clear()
{
properties.clear();
}
std::string SrsUnSortedHashtable::key_at(int index)
{
srs_assert(index < size());
SrsObjectPropertyType& elem = properties[index];
return elem.first;
}
SrsAmf0Any* SrsUnSortedHashtable::value_at(int index)
{
srs_assert(index < size());
SrsObjectPropertyType& elem = properties[index];
return elem.second;
}
void SrsUnSortedHashtable::set(std::string key, SrsAmf0Any* value)
{
std::vector<SrsObjectPropertyType>::iterator it;
for (it = properties.begin(); it != properties.end(); ++it) {
SrsObjectPropertyType& elem = *it;
std::string name = elem.first;
SrsAmf0Any* any = elem.second;
if (key == name) {
srs_freep(any);
properties.erase(it);
break;
}
}
properties.push_back(std::make_pair(key, value));
}
SrsAmf0Any* SrsUnSortedHashtable::get_property(std::string name)
{
std::vector<SrsObjectPropertyType>::iterator it;
for (it = properties.begin(); it != properties.end(); ++it) {
SrsObjectPropertyType& elem = *it;
std::string key = elem.first;
SrsAmf0Any* any = elem.second;
if (key == name) {
return any;
}
}
return NULL;
}
SrsAmf0Any* SrsUnSortedHashtable::ensure_property_string(std::string name)
{
SrsAmf0Any* prop = get_property(name);
if (!prop) {
return NULL;
}
if (!prop->is_string()) {
return NULL;
}
return prop;
}
SrsAmf0Any* SrsUnSortedHashtable::ensure_property_number(std::string name)
{
SrsAmf0Any* prop = get_property(name);
if (!prop) {
return NULL;
}
if (!prop->is_number()) {
return NULL;
}
return prop;
}
SrsAmf0Object::SrsAmf0Object()
{
marker = RTMP_AMF0_Object;
}
SrsAmf0Object::~SrsAmf0Object()
{
}
int SrsAmf0Object::size()
{
return properties.size();
}
std::string SrsAmf0Object::key_at(int index)
{
return properties.key_at(index);
}
SrsAmf0Any* SrsAmf0Object::value_at(int index)
{
return properties.value_at(index);
}
void SrsAmf0Object::set(std::string key, SrsAmf0Any* value)
{
properties.set(key, value);
}
SrsAmf0Any* SrsAmf0Object::get_property(std::string name)
{
return properties.get_property(name);
}
SrsAmf0Any* SrsAmf0Object::ensure_property_string(std::string name)
{
return properties.ensure_property_string(name);
}
SrsAmf0Any* SrsAmf0Object::ensure_property_number(std::string name)
{
return properties.ensure_property_number(name);
}
SrsASrsAmf0EcmaArray::SrsASrsAmf0EcmaArray()
{
marker = RTMP_AMF0_EcmaArray;
}
SrsASrsAmf0EcmaArray::~SrsASrsAmf0EcmaArray()
{
}
int SrsASrsAmf0EcmaArray::size()
{
return properties.size();
}
void SrsASrsAmf0EcmaArray::clear()
{
properties.clear();
}
std::string SrsASrsAmf0EcmaArray::key_at(int index)
{
return properties.key_at(index);
}
SrsAmf0Any* SrsASrsAmf0EcmaArray::value_at(int index)
{
return properties.value_at(index);
}
void SrsASrsAmf0EcmaArray::set(std::string key, SrsAmf0Any* value)
{
properties.set(key, value);
}
SrsAmf0Any* SrsASrsAmf0EcmaArray::get_property(std::string name)
{
return properties.get_property(name);
}
SrsAmf0Any* SrsASrsAmf0EcmaArray::ensure_property_string(std::string name)
{
return properties.ensure_property_string(name);
}
int srs_amf0_read_utf8(SrsStream* stream, std::string& value)
{
int ret = ERROR_SUCCESS;
// len
if (!stream->require(2)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read string length failed. ret=%d", ret);
return ret;
}
int16_t len = stream->read_2bytes();
srs_verbose("amf0 read string length success. len=%d", len);
// empty string
if (len <= 0) {
srs_verbose("amf0 read empty string. ret=%d", ret);
return ret;
}
// data
if (!stream->require(len)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read string data failed. ret=%d", ret);
return ret;
}
std::string str = stream->read_string(len);
// support utf8-1 only
// 1.3.1 Strings and UTF-8
// UTF8-1 = %x00-7F
for (int i = 0; i < len; i++) {
char ch = *(str.data() + i);
if ((ch & 0x80) != 0) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("ignored. only support utf8-1, 0x00-0x7F, actual is %#x. ret=%d", (int)ch, ret);
ret = ERROR_SUCCESS;
}
}
value = str;
srs_verbose("amf0 read string data success. str=%s", str.c_str());
return ret;
}
int srs_amf0_write_utf8(SrsStream* stream, std::string value)
{
int ret = ERROR_SUCCESS;
// len
if (!stream->require(2)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write string length failed. ret=%d", ret);
return ret;
}
stream->write_2bytes(value.length());
srs_verbose("amf0 write string length success. len=%d", (int)value.length());
// empty string
if (value.length() <= 0) {
srs_verbose("amf0 write empty string. ret=%d", ret);
return ret;
}
// data
if (!stream->require(value.length())) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write string data failed. ret=%d", ret);
return ret;
}
stream->write_string(value);
srs_verbose("amf0 write string data success. str=%s", value.c_str());
return ret;
}
int srs_amf0_read_string(SrsStream* stream, std::string& value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read string marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_String) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check string marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_String, ret);
return ret;
}
srs_verbose("amf0 read string marker success");
return srs_amf0_read_utf8(stream, value);
}
int srs_amf0_write_string(SrsStream* stream, std::string value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write string marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_String);
srs_verbose("amf0 write string marker success");
return srs_amf0_write_utf8(stream, value);
}
int srs_amf0_read_boolean(SrsStream* stream, bool& value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read bool marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_Boolean) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check bool marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_Boolean, ret);
return ret;
}
srs_verbose("amf0 read bool marker success");
// value
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read bool value failed. ret=%d", ret);
return ret;
}
if (stream->read_1bytes() == 0) {
value = false;
} else {
value = true;
}
srs_verbose("amf0 read bool value success. value=%d", value);
return ret;
}
int srs_amf0_write_boolean(SrsStream* stream, bool value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write bool marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_Boolean);
srs_verbose("amf0 write bool marker success");
// value
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write bool value failed. ret=%d", ret);
return ret;
}
if (value) {
stream->write_1bytes(0x01);
} else {
stream->write_1bytes(0x00);
}
srs_verbose("amf0 write bool value success. value=%d", value);
return ret;
}
int srs_amf0_read_number(SrsStream* stream, double& value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read number marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_Number) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check number marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_Number, ret);
return ret;
}
srs_verbose("amf0 read number marker success");
// value
if (!stream->require(8)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read number value failed. ret=%d", ret);
return ret;
}
int64_t temp = stream->read_8bytes();
memcpy(&value, &temp, 8);
srs_verbose("amf0 read number value success. value=%.2f", value);
return ret;
}
int srs_amf0_write_number(SrsStream* stream, double value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write number marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_Number);
srs_verbose("amf0 write number marker success");
// value
if (!stream->require(8)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write number value failed. ret=%d", ret);
return ret;
}
int64_t temp = 0x00;
memcpy(&temp, &value, 8);
stream->write_8bytes(temp);
srs_verbose("amf0 write number value success. value=%.2f", value);
return ret;
}
int srs_amf0_read_null(SrsStream* stream)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read null marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_Null) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check null marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_Null, ret);
return ret;
}
srs_verbose("amf0 read null success");
return ret;
}
int srs_amf0_write_null(SrsStream* stream)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write null marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_Null);
srs_verbose("amf0 write null marker success");
return ret;
}
int srs_amf0_read_undefined(SrsStream* stream)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read undefined marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_Undefined) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check undefined marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_Undefined, ret);
return ret;
}
srs_verbose("amf0 read undefined success");
return ret;
}
int srs_amf0_write_undefined(SrsStream* stream)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write undefined marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_Undefined);
srs_verbose("amf0 write undefined marker success");
return ret;
}
int srs_amf0_read_any(SrsStream* stream, SrsAmf0Any*& value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read any marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
srs_verbose("amf0 any marker success");
// backward the 1byte marker.
stream->skip(-1);
switch (marker) {
case RTMP_AMF0_String: {
std::string data;
if ((ret = srs_amf0_read_string(stream, data)) != ERROR_SUCCESS) {
return ret;
}
value = new SrsAmf0String();
srs_amf0_convert<SrsAmf0String>(value)->value = data;
return ret;
}
case RTMP_AMF0_Boolean: {
bool data;
if ((ret = srs_amf0_read_boolean(stream, data)) != ERROR_SUCCESS) {
return ret;
}
value = new SrsAmf0Boolean();
srs_amf0_convert<SrsAmf0Boolean>(value)->value = data;
return ret;
}
case RTMP_AMF0_Number: {
double data;
if ((ret = srs_amf0_read_number(stream, data)) != ERROR_SUCCESS) {
return ret;
}
value = new SrsAmf0Number();
srs_amf0_convert<SrsAmf0Number>(value)->value = data;
return ret;
}
case RTMP_AMF0_Null: {
stream->skip(1);
value = new SrsAmf0Null();
return ret;
}
case RTMP_AMF0_Undefined: {
stream->skip(1);
value = new SrsAmf0Undefined();
return ret;
}
case RTMP_AMF0_ObjectEnd: {
SrsAmf0ObjectEOF* p = NULL;
if ((ret = srs_amf0_read_object_eof(stream, p)) != ERROR_SUCCESS) {
return ret;
}
value = p;
return ret;
}
case RTMP_AMF0_Object: {
SrsAmf0Object* p = NULL;
if ((ret = srs_amf0_read_object(stream, p)) != ERROR_SUCCESS) {
return ret;
}
value = p;
return ret;
}
case RTMP_AMF0_EcmaArray: {
SrsASrsAmf0EcmaArray* p = NULL;
if ((ret = srs_amf0_read_ecma_array(stream, p)) != ERROR_SUCCESS) {
return ret;
}
value = p;
return ret;
}
case RTMP_AMF0_Invalid:
default: {
ret = ERROR_RTMP_AMF0_INVALID;
srs_error("invalid amf0 message type. marker=%#x, ret=%d", marker, ret);
return ret;
}
}
return ret;
}
int srs_amf0_write_any(SrsStream* stream, SrsAmf0Any* value)
{
int ret = ERROR_SUCCESS;
srs_assert(value != NULL);
switch (value->marker) {
case RTMP_AMF0_String: {
std::string data = srs_amf0_convert<SrsAmf0String>(value)->value;
return srs_amf0_write_string(stream, data);
}
case RTMP_AMF0_Boolean: {
bool data = srs_amf0_convert<SrsAmf0Boolean>(value)->value;
return srs_amf0_write_boolean(stream, data);
}
case RTMP_AMF0_Number: {
double data = srs_amf0_convert<SrsAmf0Number>(value)->value;
return srs_amf0_write_number(stream, data);
}
case RTMP_AMF0_Null: {
return srs_amf0_write_null(stream);
}
case RTMP_AMF0_Undefined: {
return srs_amf0_write_undefined(stream);
}
case RTMP_AMF0_ObjectEnd: {
SrsAmf0ObjectEOF* p = srs_amf0_convert<SrsAmf0ObjectEOF>(value);
return srs_amf0_write_object_eof(stream, p);
}
case RTMP_AMF0_Object: {
SrsAmf0Object* p = srs_amf0_convert<SrsAmf0Object>(value);
return srs_amf0_write_object(stream, p);
}
case RTMP_AMF0_EcmaArray: {
SrsASrsAmf0EcmaArray* p = srs_amf0_convert<SrsASrsAmf0EcmaArray>(value);
return srs_amf0_write_ecma_array(stream, p);
}
case RTMP_AMF0_Invalid:
default: {
ret = ERROR_RTMP_AMF0_INVALID;
srs_error("invalid amf0 message type. marker=%#x, ret=%d", value->marker, ret);
return ret;
}
}
return ret;
}
int srs_amf0_get_any_size(SrsAmf0Any* value)
{
if (!value) {
return 0;
}
int size = 0;
switch (value->marker) {
case RTMP_AMF0_String: {
SrsAmf0String* p = srs_amf0_convert<SrsAmf0String>(value);
size += srs_amf0_get_string_size(p->value);
break;
}
case RTMP_AMF0_Boolean: {
size += srs_amf0_get_boolean_size();
break;
}
case RTMP_AMF0_Number: {
size += srs_amf0_get_number_size();
break;
}
case RTMP_AMF0_Null: {
size += srs_amf0_get_null_size();
break;
}
case RTMP_AMF0_Undefined: {
size += srs_amf0_get_undefined_size();
break;
}
case RTMP_AMF0_ObjectEnd: {
size += srs_amf0_get_object_eof_size();
break;
}
case RTMP_AMF0_Object: {
SrsAmf0Object* p = srs_amf0_convert<SrsAmf0Object>(value);
size += srs_amf0_get_object_size(p);
break;
}
case RTMP_AMF0_EcmaArray: {
SrsASrsAmf0EcmaArray* p = srs_amf0_convert<SrsASrsAmf0EcmaArray>(value);
size += srs_amf0_get_ecma_array_size(p);
break;
}
default: {
// TOOD: other AMF0 types.
srs_warn("ignore unkown AMF0 type size.");
break;
}
}
return size;
}
int srs_amf0_read_object_eof(SrsStream* stream, SrsAmf0ObjectEOF*& value)
{
int ret = ERROR_SUCCESS;
// auto skip -2 to read the object eof.
srs_assert(stream->pos() >= 2);
stream->skip(-2);
// value
if (!stream->require(2)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read object eof value failed. ret=%d", ret);
return ret;
}
int16_t temp = stream->read_2bytes();
if (temp != 0x00) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read object eof value check failed. "
"must be 0x00, actual is %#x, ret=%d", temp, ret);
return ret;
}
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read object eof marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_ObjectEnd) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check object eof marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_ObjectEnd, ret);
return ret;
}
srs_verbose("amf0 read object eof marker success");
value = new SrsAmf0ObjectEOF();
srs_verbose("amf0 read object eof success");
return ret;
}
int srs_amf0_write_object_eof(SrsStream* stream, SrsAmf0ObjectEOF* value)
{
int ret = ERROR_SUCCESS;
srs_assert(value != NULL);
// value
if (!stream->require(2)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write object eof value failed. ret=%d", ret);
return ret;
}
stream->write_2bytes(0x00);
srs_verbose("amf0 write object eof value success");
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write object eof marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_ObjectEnd);
srs_verbose("amf0 read object eof success");
return ret;
}
int srs_amf0_read_object(SrsStream* stream, SrsAmf0Object*& value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read object marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_Object) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check object marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_Object, ret);
return ret;
}
srs_verbose("amf0 read object marker success");
// value
value = new SrsAmf0Object();
while (!stream->empty()) {
// property-name: utf8 string
std::string property_name;
if ((ret =srs_amf0_read_utf8(stream, property_name)) != ERROR_SUCCESS) {
srs_error("amf0 object read property name failed. ret=%d", ret);
return ret;
}
// property-value: any
SrsAmf0Any* property_value = NULL;
if ((ret = srs_amf0_read_any(stream, property_value)) != ERROR_SUCCESS) {
srs_error("amf0 object read property_value failed. "
"name=%s, ret=%d", property_name.c_str(), ret);
return ret;
}
// AMF0 Object EOF.
if (property_name.empty() || !property_value || property_value->is_object_eof()) {
if (property_value) {
srs_freep(property_value);
}
srs_info("amf0 read object EOF.");
break;
}
// add property
value->set(property_name, property_value);
}
return ret;
}
int srs_amf0_write_object(SrsStream* stream, SrsAmf0Object* value)
{
int ret = ERROR_SUCCESS;
srs_assert(value != NULL);
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write object marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_Object);
srs_verbose("amf0 write object marker success");
// value
for (int i = 0; i < value->size(); i++) {
std::string name = value->key_at(i);
SrsAmf0Any* any = value->value_at(i);
if ((ret = srs_amf0_write_utf8(stream, name)) != ERROR_SUCCESS) {
srs_error("write object property name failed. ret=%d", ret);
return ret;
}
if ((ret = srs_amf0_write_any(stream, any)) != ERROR_SUCCESS) {
srs_error("write object property value failed. ret=%d", ret);
return ret;
}
srs_verbose("write amf0 property success. name=%s", name.c_str());
}
if ((ret = srs_amf0_write_object_eof(stream, &value->eof)) != ERROR_SUCCESS) {
srs_error("write object eof failed. ret=%d", ret);
return ret;
}
srs_verbose("write amf0 object success.");
return ret;
}
int srs_amf0_read_ecma_array(SrsStream* stream, SrsASrsAmf0EcmaArray*& value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read ecma_array marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_EcmaArray) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check ecma_array marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_Object, ret);
return ret;
}
srs_verbose("amf0 read ecma_array marker success");
// count
if (!stream->require(4)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read ecma_array count failed. ret=%d", ret);
return ret;
}
int32_t count = stream->read_4bytes();
srs_verbose("amf0 read ecma_array count success. count=%d", count);
// value
value = new SrsASrsAmf0EcmaArray();
value->count = count;
while (!stream->empty()) {
// property-name: utf8 string
std::string property_name;
if ((ret =srs_amf0_read_utf8(stream, property_name)) != ERROR_SUCCESS) {
srs_error("amf0 ecma_array read property name failed. ret=%d", ret);
return ret;
}
// property-value: any
SrsAmf0Any* property_value = NULL;
if ((ret = srs_amf0_read_any(stream, property_value)) != ERROR_SUCCESS) {
srs_error("amf0 ecma_array read property_value failed. "
"name=%s, ret=%d", property_name.c_str(), ret);
return ret;
}
// AMF0 Object EOF.
if (property_name.empty() || !property_value || property_value->is_object_eof()) {
if (property_value) {
srs_freep(property_value);
}
srs_info("amf0 read ecma_array EOF.");
break;
}
// add property
value->set(property_name, property_value);
}
return ret;
}
int srs_amf0_write_ecma_array(SrsStream* stream, SrsASrsAmf0EcmaArray* value)
{
int ret = ERROR_SUCCESS;
srs_assert(value != NULL);
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write ecma_array marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_EcmaArray);
srs_verbose("amf0 write ecma_array marker success");
// count
if (!stream->require(4)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write ecma_array count failed. ret=%d", ret);
return ret;
}
stream->write_4bytes(value->count);
srs_verbose("amf0 write ecma_array count success. count=%d", value->count);
// value
for (int i = 0; i < value->size(); i++) {
std::string name = value->key_at(i);
SrsAmf0Any* any = value->value_at(i);
if ((ret = srs_amf0_write_utf8(stream, name)) != ERROR_SUCCESS) {
srs_error("write ecma_array property name failed. ret=%d", ret);
return ret;
}
if ((ret = srs_amf0_write_any(stream, any)) != ERROR_SUCCESS) {
srs_error("write ecma_array property value failed. ret=%d", ret);
return ret;
}
srs_verbose("write amf0 property success. name=%s", name.c_str());
}
if ((ret = srs_amf0_write_object_eof(stream, &value->eof)) != ERROR_SUCCESS) {
srs_error("write ecma_array eof failed. ret=%d", ret);
return ret;
}
srs_verbose("write ecma_array object success.");
return ret;
}
int srs_amf0_get_utf8_size(std::string value)
{
return 2 + value.length();
}
int srs_amf0_get_string_size(std::string value)
{
return 1 + srs_amf0_get_utf8_size(value);
}
int srs_amf0_get_number_size()
{
return 1 + 8;
}
int srs_amf0_get_null_size()
{
return 1;
}
int srs_amf0_get_undefined_size()
{
return 1;
}
int srs_amf0_get_boolean_size()
{
return 1 + 1;
}
int srs_amf0_get_object_size(SrsAmf0Object* obj)
{
if (!obj) {
return 0;
}
int size = 1;
for (int i = 0; i < obj->size(); i++){
std::string name = obj->key_at(i);
SrsAmf0Any* value = obj->value_at(i);
size += srs_amf0_get_utf8_size(name);
size += srs_amf0_get_any_size(value);
}
size += srs_amf0_get_object_eof_size();
return size;
}
int srs_amf0_get_ecma_array_size(SrsASrsAmf0EcmaArray* arr)
{
if (!arr) {
return 0;
}
int size = 1 + 4;
for (int i = 0; i < arr->size(); i++){
std::string name = arr->key_at(i);
SrsAmf0Any* value = arr->value_at(i);
size += srs_amf0_get_utf8_size(name);
size += srs_amf0_get_any_size(value);
}
size += srs_amf0_get_object_eof_size();
return size;
}
int srs_amf0_get_object_eof_size()
{
return 2 + 1;
}
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include <srs_core_amf0.hpp>
#include <utility>
#include <srs_core_log.hpp>
#include <srs_core_error.hpp>
#include <srs_core_stream.hpp>
// AMF0 marker
#define RTMP_AMF0_Number 0x00
#define RTMP_AMF0_Boolean 0x01
#define RTMP_AMF0_String 0x02
#define RTMP_AMF0_Object 0x03
#define RTMP_AMF0_MovieClip 0x04 // reserved, not supported
#define RTMP_AMF0_Null 0x05
#define RTMP_AMF0_Undefined 0x06
#define RTMP_AMF0_Reference 0x07
#define RTMP_AMF0_EcmaArray 0x08
#define RTMP_AMF0_ObjectEnd 0x09
#define RTMP_AMF0_StrictArray 0x0A
#define RTMP_AMF0_Date 0x0B
#define RTMP_AMF0_LongString 0x0C
#define RTMP_AMF0_UnSupported 0x0D
#define RTMP_AMF0_RecordSet 0x0E // reserved, not supported
#define RTMP_AMF0_XmlDocument 0x0F
#define RTMP_AMF0_TypedObject 0x10
// AVM+ object is the AMF3 object.
#define RTMP_AMF0_AVMplusObject 0x11
// origin array whos data takes the same form as LengthValueBytes
#define RTMP_AMF0_OriginStrictArray 0x20
// User defined
#define RTMP_AMF0_Invalid 0x3F
int srs_amf0_get_object_eof_size();
int srs_amf0_get_any_size(SrsAmf0Any* value);
int srs_amf0_read_object_eof(SrsStream* stream, SrsAmf0ObjectEOF*&);
int srs_amf0_write_object_eof(SrsStream* stream, SrsAmf0ObjectEOF*);
int srs_amf0_read_any(SrsStream* stream, SrsAmf0Any*& value);
int srs_amf0_write_any(SrsStream* stream, SrsAmf0Any* value);
SrsAmf0Any::SrsAmf0Any()
{
marker = RTMP_AMF0_Invalid;
}
SrsAmf0Any::~SrsAmf0Any()
{
}
bool SrsAmf0Any::is_string()
{
return marker == RTMP_AMF0_String;
}
bool SrsAmf0Any::is_boolean()
{
return marker == RTMP_AMF0_Boolean;
}
bool SrsAmf0Any::is_number()
{
return marker == RTMP_AMF0_Number;
}
bool SrsAmf0Any::is_null()
{
return marker == RTMP_AMF0_Null;
}
bool SrsAmf0Any::is_undefined()
{
return marker == RTMP_AMF0_Undefined;
}
bool SrsAmf0Any::is_object()
{
return marker == RTMP_AMF0_Object;
}
bool SrsAmf0Any::is_ecma_array()
{
return marker == RTMP_AMF0_EcmaArray;
}
bool SrsAmf0Any::is_object_eof()
{
return marker == RTMP_AMF0_ObjectEnd;
}
SrsAmf0String::SrsAmf0String(const char* _value)
{
marker = RTMP_AMF0_String;
if (_value) {
value = _value;
}
}
SrsAmf0String::~SrsAmf0String()
{
}
SrsAmf0Boolean::SrsAmf0Boolean(bool _value)
{
marker = RTMP_AMF0_Boolean;
value = _value;
}
SrsAmf0Boolean::~SrsAmf0Boolean()
{
}
SrsAmf0Number::SrsAmf0Number(double _value)
{
marker = RTMP_AMF0_Number;
value = _value;
}
SrsAmf0Number::~SrsAmf0Number()
{
}
SrsAmf0Null::SrsAmf0Null()
{
marker = RTMP_AMF0_Null;
}
SrsAmf0Null::~SrsAmf0Null()
{
}
SrsAmf0Undefined::SrsAmf0Undefined()
{
marker = RTMP_AMF0_Undefined;
}
SrsAmf0Undefined::~SrsAmf0Undefined()
{
}
SrsAmf0ObjectEOF::SrsAmf0ObjectEOF()
{
marker = RTMP_AMF0_ObjectEnd;
utf8_empty = 0x00;
}
SrsAmf0ObjectEOF::~SrsAmf0ObjectEOF()
{
}
SrsUnSortedHashtable::SrsUnSortedHashtable()
{
}
SrsUnSortedHashtable::~SrsUnSortedHashtable()
{
std::vector<SrsObjectPropertyType>::iterator it;
for (it = properties.begin(); it != properties.end(); ++it) {
SrsObjectPropertyType& elem = *it;
SrsAmf0Any* any = elem.second;
srs_freep(any);
}
properties.clear();
}
int SrsUnSortedHashtable::size()
{
return (int)properties.size();
}
void SrsUnSortedHashtable::clear()
{
properties.clear();
}
std::string SrsUnSortedHashtable::key_at(int index)
{
srs_assert(index < size());
SrsObjectPropertyType& elem = properties[index];
return elem.first;
}
SrsAmf0Any* SrsUnSortedHashtable::value_at(int index)
{
srs_assert(index < size());
SrsObjectPropertyType& elem = properties[index];
return elem.second;
}
void SrsUnSortedHashtable::set(std::string key, SrsAmf0Any* value)
{
std::vector<SrsObjectPropertyType>::iterator it;
for (it = properties.begin(); it != properties.end(); ++it) {
SrsObjectPropertyType& elem = *it;
std::string name = elem.first;
SrsAmf0Any* any = elem.second;
if (key == name) {
srs_freep(any);
properties.erase(it);
break;
}
}
properties.push_back(std::make_pair(key, value));
}
SrsAmf0Any* SrsUnSortedHashtable::get_property(std::string name)
{
std::vector<SrsObjectPropertyType>::iterator it;
for (it = properties.begin(); it != properties.end(); ++it) {
SrsObjectPropertyType& elem = *it;
std::string key = elem.first;
SrsAmf0Any* any = elem.second;
if (key == name) {
return any;
}
}
return NULL;
}
SrsAmf0Any* SrsUnSortedHashtable::ensure_property_string(std::string name)
{
SrsAmf0Any* prop = get_property(name);
if (!prop) {
return NULL;
}
if (!prop->is_string()) {
return NULL;
}
return prop;
}
SrsAmf0Any* SrsUnSortedHashtable::ensure_property_number(std::string name)
{
SrsAmf0Any* prop = get_property(name);
if (!prop) {
return NULL;
}
if (!prop->is_number()) {
return NULL;
}
return prop;
}
SrsAmf0Object::SrsAmf0Object()
{
marker = RTMP_AMF0_Object;
}
SrsAmf0Object::~SrsAmf0Object()
{
}
int SrsAmf0Object::size()
{
return properties.size();
}
std::string SrsAmf0Object::key_at(int index)
{
return properties.key_at(index);
}
SrsAmf0Any* SrsAmf0Object::value_at(int index)
{
return properties.value_at(index);
}
void SrsAmf0Object::set(std::string key, SrsAmf0Any* value)
{
properties.set(key, value);
}
SrsAmf0Any* SrsAmf0Object::get_property(std::string name)
{
return properties.get_property(name);
}
SrsAmf0Any* SrsAmf0Object::ensure_property_string(std::string name)
{
return properties.ensure_property_string(name);
}
SrsAmf0Any* SrsAmf0Object::ensure_property_number(std::string name)
{
return properties.ensure_property_number(name);
}
SrsASrsAmf0EcmaArray::SrsASrsAmf0EcmaArray()
{
marker = RTMP_AMF0_EcmaArray;
}
SrsASrsAmf0EcmaArray::~SrsASrsAmf0EcmaArray()
{
}
int SrsASrsAmf0EcmaArray::size()
{
return properties.size();
}
void SrsASrsAmf0EcmaArray::clear()
{
properties.clear();
}
std::string SrsASrsAmf0EcmaArray::key_at(int index)
{
return properties.key_at(index);
}
SrsAmf0Any* SrsASrsAmf0EcmaArray::value_at(int index)
{
return properties.value_at(index);
}
void SrsASrsAmf0EcmaArray::set(std::string key, SrsAmf0Any* value)
{
properties.set(key, value);
}
SrsAmf0Any* SrsASrsAmf0EcmaArray::get_property(std::string name)
{
return properties.get_property(name);
}
SrsAmf0Any* SrsASrsAmf0EcmaArray::ensure_property_string(std::string name)
{
return properties.ensure_property_string(name);
}
int srs_amf0_read_utf8(SrsStream* stream, std::string& value)
{
int ret = ERROR_SUCCESS;
// len
if (!stream->require(2)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read string length failed. ret=%d", ret);
return ret;
}
int16_t len = stream->read_2bytes();
srs_verbose("amf0 read string length success. len=%d", len);
// empty string
if (len <= 0) {
srs_verbose("amf0 read empty string. ret=%d", ret);
return ret;
}
// data
if (!stream->require(len)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read string data failed. ret=%d", ret);
return ret;
}
std::string str = stream->read_string(len);
// support utf8-1 only
// 1.3.1 Strings and UTF-8
// UTF8-1 = %x00-7F
for (int i = 0; i < len; i++) {
char ch = *(str.data() + i);
if ((ch & 0x80) != 0) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("ignored. only support utf8-1, 0x00-0x7F, actual is %#x. ret=%d", (int)ch, ret);
ret = ERROR_SUCCESS;
}
}
value = str;
srs_verbose("amf0 read string data success. str=%s", str.c_str());
return ret;
}
int srs_amf0_write_utf8(SrsStream* stream, std::string value)
{
int ret = ERROR_SUCCESS;
// len
if (!stream->require(2)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write string length failed. ret=%d", ret);
return ret;
}
stream->write_2bytes(value.length());
srs_verbose("amf0 write string length success. len=%d", (int)value.length());
// empty string
if (value.length() <= 0) {
srs_verbose("amf0 write empty string. ret=%d", ret);
return ret;
}
// data
if (!stream->require(value.length())) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write string data failed. ret=%d", ret);
return ret;
}
stream->write_string(value);
srs_verbose("amf0 write string data success. str=%s", value.c_str());
return ret;
}
int srs_amf0_read_string(SrsStream* stream, std::string& value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read string marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_String) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check string marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_String, ret);
return ret;
}
srs_verbose("amf0 read string marker success");
return srs_amf0_read_utf8(stream, value);
}
int srs_amf0_write_string(SrsStream* stream, std::string value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write string marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_String);
srs_verbose("amf0 write string marker success");
return srs_amf0_write_utf8(stream, value);
}
int srs_amf0_read_boolean(SrsStream* stream, bool& value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read bool marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_Boolean) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check bool marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_Boolean, ret);
return ret;
}
srs_verbose("amf0 read bool marker success");
// value
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read bool value failed. ret=%d", ret);
return ret;
}
if (stream->read_1bytes() == 0) {
value = false;
} else {
value = true;
}
srs_verbose("amf0 read bool value success. value=%d", value);
return ret;
}
int srs_amf0_write_boolean(SrsStream* stream, bool value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write bool marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_Boolean);
srs_verbose("amf0 write bool marker success");
// value
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write bool value failed. ret=%d", ret);
return ret;
}
if (value) {
stream->write_1bytes(0x01);
} else {
stream->write_1bytes(0x00);
}
srs_verbose("amf0 write bool value success. value=%d", value);
return ret;
}
int srs_amf0_read_number(SrsStream* stream, double& value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read number marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_Number) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check number marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_Number, ret);
return ret;
}
srs_verbose("amf0 read number marker success");
// value
if (!stream->require(8)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read number value failed. ret=%d", ret);
return ret;
}
int64_t temp = stream->read_8bytes();
memcpy(&value, &temp, 8);
srs_verbose("amf0 read number value success. value=%.2f", value);
return ret;
}
int srs_amf0_write_number(SrsStream* stream, double value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write number marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_Number);
srs_verbose("amf0 write number marker success");
// value
if (!stream->require(8)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write number value failed. ret=%d", ret);
return ret;
}
int64_t temp = 0x00;
memcpy(&temp, &value, 8);
stream->write_8bytes(temp);
srs_verbose("amf0 write number value success. value=%.2f", value);
return ret;
}
int srs_amf0_read_null(SrsStream* stream)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read null marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_Null) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check null marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_Null, ret);
return ret;
}
srs_verbose("amf0 read null success");
return ret;
}
int srs_amf0_write_null(SrsStream* stream)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write null marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_Null);
srs_verbose("amf0 write null marker success");
return ret;
}
int srs_amf0_read_undefined(SrsStream* stream)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read undefined marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_Undefined) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check undefined marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_Undefined, ret);
return ret;
}
srs_verbose("amf0 read undefined success");
return ret;
}
int srs_amf0_write_undefined(SrsStream* stream)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write undefined marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_Undefined);
srs_verbose("amf0 write undefined marker success");
return ret;
}
int srs_amf0_read_any(SrsStream* stream, SrsAmf0Any*& value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read any marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
srs_verbose("amf0 any marker success");
// backward the 1byte marker.
stream->skip(-1);
switch (marker) {
case RTMP_AMF0_String: {
std::string data;
if ((ret = srs_amf0_read_string(stream, data)) != ERROR_SUCCESS) {
return ret;
}
value = new SrsAmf0String();
srs_amf0_convert<SrsAmf0String>(value)->value = data;
return ret;
}
case RTMP_AMF0_Boolean: {
bool data;
if ((ret = srs_amf0_read_boolean(stream, data)) != ERROR_SUCCESS) {
return ret;
}
value = new SrsAmf0Boolean();
srs_amf0_convert<SrsAmf0Boolean>(value)->value = data;
return ret;
}
case RTMP_AMF0_Number: {
double data;
if ((ret = srs_amf0_read_number(stream, data)) != ERROR_SUCCESS) {
return ret;
}
value = new SrsAmf0Number();
srs_amf0_convert<SrsAmf0Number>(value)->value = data;
return ret;
}
case RTMP_AMF0_Null: {
stream->skip(1);
value = new SrsAmf0Null();
return ret;
}
case RTMP_AMF0_Undefined: {
stream->skip(1);
value = new SrsAmf0Undefined();
return ret;
}
case RTMP_AMF0_ObjectEnd: {
SrsAmf0ObjectEOF* p = NULL;
if ((ret = srs_amf0_read_object_eof(stream, p)) != ERROR_SUCCESS) {
return ret;
}
value = p;
return ret;
}
case RTMP_AMF0_Object: {
SrsAmf0Object* p = NULL;
if ((ret = srs_amf0_read_object(stream, p)) != ERROR_SUCCESS) {
return ret;
}
value = p;
return ret;
}
case RTMP_AMF0_EcmaArray: {
SrsASrsAmf0EcmaArray* p = NULL;
if ((ret = srs_amf0_read_ecma_array(stream, p)) != ERROR_SUCCESS) {
return ret;
}
value = p;
return ret;
}
case RTMP_AMF0_Invalid:
default: {
ret = ERROR_RTMP_AMF0_INVALID;
srs_error("invalid amf0 message type. marker=%#x, ret=%d", marker, ret);
return ret;
}
}
return ret;
}
int srs_amf0_write_any(SrsStream* stream, SrsAmf0Any* value)
{
int ret = ERROR_SUCCESS;
srs_assert(value != NULL);
switch (value->marker) {
case RTMP_AMF0_String: {
std::string data = srs_amf0_convert<SrsAmf0String>(value)->value;
return srs_amf0_write_string(stream, data);
}
case RTMP_AMF0_Boolean: {
bool data = srs_amf0_convert<SrsAmf0Boolean>(value)->value;
return srs_amf0_write_boolean(stream, data);
}
case RTMP_AMF0_Number: {
double data = srs_amf0_convert<SrsAmf0Number>(value)->value;
return srs_amf0_write_number(stream, data);
}
case RTMP_AMF0_Null: {
return srs_amf0_write_null(stream);
}
case RTMP_AMF0_Undefined: {
return srs_amf0_write_undefined(stream);
}
case RTMP_AMF0_ObjectEnd: {
SrsAmf0ObjectEOF* p = srs_amf0_convert<SrsAmf0ObjectEOF>(value);
return srs_amf0_write_object_eof(stream, p);
}
case RTMP_AMF0_Object: {
SrsAmf0Object* p = srs_amf0_convert<SrsAmf0Object>(value);
return srs_amf0_write_object(stream, p);
}
case RTMP_AMF0_EcmaArray: {
SrsASrsAmf0EcmaArray* p = srs_amf0_convert<SrsASrsAmf0EcmaArray>(value);
return srs_amf0_write_ecma_array(stream, p);
}
case RTMP_AMF0_Invalid:
default: {
ret = ERROR_RTMP_AMF0_INVALID;
srs_error("invalid amf0 message type. marker=%#x, ret=%d", value->marker, ret);
return ret;
}
}
return ret;
}
int srs_amf0_get_any_size(SrsAmf0Any* value)
{
if (!value) {
return 0;
}
int size = 0;
switch (value->marker) {
case RTMP_AMF0_String: {
SrsAmf0String* p = srs_amf0_convert<SrsAmf0String>(value);
size += srs_amf0_get_string_size(p->value);
break;
}
case RTMP_AMF0_Boolean: {
size += srs_amf0_get_boolean_size();
break;
}
case RTMP_AMF0_Number: {
size += srs_amf0_get_number_size();
break;
}
case RTMP_AMF0_Null: {
size += srs_amf0_get_null_size();
break;
}
case RTMP_AMF0_Undefined: {
size += srs_amf0_get_undefined_size();
break;
}
case RTMP_AMF0_ObjectEnd: {
size += srs_amf0_get_object_eof_size();
break;
}
case RTMP_AMF0_Object: {
SrsAmf0Object* p = srs_amf0_convert<SrsAmf0Object>(value);
size += srs_amf0_get_object_size(p);
break;
}
case RTMP_AMF0_EcmaArray: {
SrsASrsAmf0EcmaArray* p = srs_amf0_convert<SrsASrsAmf0EcmaArray>(value);
size += srs_amf0_get_ecma_array_size(p);
break;
}
default: {
// TOOD: other AMF0 types.
srs_warn("ignore unkown AMF0 type size.");
break;
}
}
return size;
}
int srs_amf0_read_object_eof(SrsStream* stream, SrsAmf0ObjectEOF*& value)
{
int ret = ERROR_SUCCESS;
// auto skip -2 to read the object eof.
srs_assert(stream->pos() >= 2);
stream->skip(-2);
// value
if (!stream->require(2)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read object eof value failed. ret=%d", ret);
return ret;
}
int16_t temp = stream->read_2bytes();
if (temp != 0x00) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read object eof value check failed. "
"must be 0x00, actual is %#x, ret=%d", temp, ret);
return ret;
}
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read object eof marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_ObjectEnd) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check object eof marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_ObjectEnd, ret);
return ret;
}
srs_verbose("amf0 read object eof marker success");
value = new SrsAmf0ObjectEOF();
srs_verbose("amf0 read object eof success");
return ret;
}
int srs_amf0_write_object_eof(SrsStream* stream, SrsAmf0ObjectEOF* value)
{
int ret = ERROR_SUCCESS;
srs_assert(value != NULL);
// value
if (!stream->require(2)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write object eof value failed. ret=%d", ret);
return ret;
}
stream->write_2bytes(0x00);
srs_verbose("amf0 write object eof value success");
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write object eof marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_ObjectEnd);
srs_verbose("amf0 read object eof success");
return ret;
}
int srs_amf0_read_object(SrsStream* stream, SrsAmf0Object*& value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read object marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_Object) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check object marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_Object, ret);
return ret;
}
srs_verbose("amf0 read object marker success");
// value
value = new SrsAmf0Object();
while (!stream->empty()) {
// property-name: utf8 string
std::string property_name;
if ((ret =srs_amf0_read_utf8(stream, property_name)) != ERROR_SUCCESS) {
srs_error("amf0 object read property name failed. ret=%d", ret);
return ret;
}
// property-value: any
SrsAmf0Any* property_value = NULL;
if ((ret = srs_amf0_read_any(stream, property_value)) != ERROR_SUCCESS) {
srs_error("amf0 object read property_value failed. "
"name=%s, ret=%d", property_name.c_str(), ret);
return ret;
}
// AMF0 Object EOF.
if (property_name.empty() || !property_value || property_value->is_object_eof()) {
if (property_value) {
srs_freep(property_value);
}
srs_info("amf0 read object EOF.");
break;
}
// add property
value->set(property_name, property_value);
}
return ret;
}
int srs_amf0_write_object(SrsStream* stream, SrsAmf0Object* value)
{
int ret = ERROR_SUCCESS;
srs_assert(value != NULL);
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write object marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_Object);
srs_verbose("amf0 write object marker success");
// value
for (int i = 0; i < value->size(); i++) {
std::string name = value->key_at(i);
SrsAmf0Any* any = value->value_at(i);
if ((ret = srs_amf0_write_utf8(stream, name)) != ERROR_SUCCESS) {
srs_error("write object property name failed. ret=%d", ret);
return ret;
}
if ((ret = srs_amf0_write_any(stream, any)) != ERROR_SUCCESS) {
srs_error("write object property value failed. ret=%d", ret);
return ret;
}
srs_verbose("write amf0 property success. name=%s", name.c_str());
}
if ((ret = srs_amf0_write_object_eof(stream, &value->eof)) != ERROR_SUCCESS) {
srs_error("write object eof failed. ret=%d", ret);
return ret;
}
srs_verbose("write amf0 object success.");
return ret;
}
int srs_amf0_read_ecma_array(SrsStream* stream, SrsASrsAmf0EcmaArray*& value)
{
int ret = ERROR_SUCCESS;
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read ecma_array marker failed. ret=%d", ret);
return ret;
}
char marker = stream->read_1bytes();
if (marker != RTMP_AMF0_EcmaArray) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 check ecma_array marker failed. "
"marker=%#x, required=%#x, ret=%d", marker, RTMP_AMF0_Object, ret);
return ret;
}
srs_verbose("amf0 read ecma_array marker success");
// count
if (!stream->require(4)) {
ret = ERROR_RTMP_AMF0_DECODE;
srs_error("amf0 read ecma_array count failed. ret=%d", ret);
return ret;
}
int32_t count = stream->read_4bytes();
srs_verbose("amf0 read ecma_array count success. count=%d", count);
// value
value = new SrsASrsAmf0EcmaArray();
value->count = count;
while (!stream->empty()) {
// property-name: utf8 string
std::string property_name;
if ((ret =srs_amf0_read_utf8(stream, property_name)) != ERROR_SUCCESS) {
srs_error("amf0 ecma_array read property name failed. ret=%d", ret);
return ret;
}
// property-value: any
SrsAmf0Any* property_value = NULL;
if ((ret = srs_amf0_read_any(stream, property_value)) != ERROR_SUCCESS) {
srs_error("amf0 ecma_array read property_value failed. "
"name=%s, ret=%d", property_name.c_str(), ret);
return ret;
}
// AMF0 Object EOF.
if (property_name.empty() || !property_value || property_value->is_object_eof()) {
if (property_value) {
srs_freep(property_value);
}
srs_info("amf0 read ecma_array EOF.");
break;
}
// add property
value->set(property_name, property_value);
}
return ret;
}
int srs_amf0_write_ecma_array(SrsStream* stream, SrsASrsAmf0EcmaArray* value)
{
int ret = ERROR_SUCCESS;
srs_assert(value != NULL);
// marker
if (!stream->require(1)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write ecma_array marker failed. ret=%d", ret);
return ret;
}
stream->write_1bytes(RTMP_AMF0_EcmaArray);
srs_verbose("amf0 write ecma_array marker success");
// count
if (!stream->require(4)) {
ret = ERROR_RTMP_AMF0_ENCODE;
srs_error("amf0 write ecma_array count failed. ret=%d", ret);
return ret;
}
stream->write_4bytes(value->count);
srs_verbose("amf0 write ecma_array count success. count=%d", value->count);
// value
for (int i = 0; i < value->size(); i++) {
std::string name = value->key_at(i);
SrsAmf0Any* any = value->value_at(i);
if ((ret = srs_amf0_write_utf8(stream, name)) != ERROR_SUCCESS) {
srs_error("write ecma_array property name failed. ret=%d", ret);
return ret;
}
if ((ret = srs_amf0_write_any(stream, any)) != ERROR_SUCCESS) {
srs_error("write ecma_array property value failed. ret=%d", ret);
return ret;
}
srs_verbose("write amf0 property success. name=%s", name.c_str());
}
if ((ret = srs_amf0_write_object_eof(stream, &value->eof)) != ERROR_SUCCESS) {
srs_error("write ecma_array eof failed. ret=%d", ret);
return ret;
}
srs_verbose("write ecma_array object success.");
return ret;
}
int srs_amf0_get_utf8_size(std::string value)
{
return 2 + value.length();
}
int srs_amf0_get_string_size(std::string value)
{
return 1 + srs_amf0_get_utf8_size(value);
}
int srs_amf0_get_number_size()
{
return 1 + 8;
}
int srs_amf0_get_null_size()
{
return 1;
}
int srs_amf0_get_undefined_size()
{
return 1;
}
int srs_amf0_get_boolean_size()
{
return 1 + 1;
}
int srs_amf0_get_object_size(SrsAmf0Object* obj)
{
if (!obj) {
return 0;
}
int size = 1;
for (int i = 0; i < obj->size(); i++){
std::string name = obj->key_at(i);
SrsAmf0Any* value = obj->value_at(i);
size += srs_amf0_get_utf8_size(name);
size += srs_amf0_get_any_size(value);
}
size += srs_amf0_get_object_eof_size();
return size;
}
int srs_amf0_get_ecma_array_size(SrsASrsAmf0EcmaArray* arr)
{
if (!arr) {
return 0;
}
int size = 1 + 4;
for (int i = 0; i < arr->size(); i++){
std::string name = arr->key_at(i);
SrsAmf0Any* value = arr->value_at(i);
size += srs_amf0_get_utf8_size(name);
size += srs_amf0_get_any_size(value);
}
size += srs_amf0_get_object_eof_size();
return size;
}
int srs_amf0_get_object_eof_size()
{
return 2 + 1;
}
... ...
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#ifndef SRS_CORE_AMF0_HPP
#define SRS_CORE_AMF0_HPP
/*
#include <srs_core_amf0.hpp>
*/
#include <srs_core.hpp>
#include <string>
#include <vector>
class SrsStream;
class SrsAmf0Object;
/**
* any amf0 value.
* 2.1 Types Overview
* value-type = number-type | boolean-type | string-type | object-type
* | null-marker | undefined-marker | reference-type | ecma-array-type
* | strict-array-type | date-type | long-string-type | xml-document-type
* | typed-object-type
*/
struct SrsAmf0Any
{
char marker;
SrsAmf0Any();
virtual ~SrsAmf0Any();
virtual bool is_string();
virtual bool is_boolean();
virtual bool is_number();
virtual bool is_null();
virtual bool is_undefined();
virtual bool is_object();
virtual bool is_object_eof();
virtual bool is_ecma_array();
};
/**
* read amf0 string from stream.
* 2.4 String Type
* string-type = string-marker UTF-8
* @return default value is empty string.
*/
struct SrsAmf0String : public SrsAmf0Any
{
std::string value;
SrsAmf0String(const char* _value = NULL);
virtual ~SrsAmf0String();
};
/**
* read amf0 boolean from stream.
* 2.4 String Type
* boolean-type = boolean-marker U8
* 0 is false, <> 0 is true
* @return default value is false.
*/
struct SrsAmf0Boolean : public SrsAmf0Any
{
bool value;
SrsAmf0Boolean(bool _value = false);
virtual ~SrsAmf0Boolean();
};
/**
* read amf0 number from stream.
* 2.2 Number Type
* number-type = number-marker DOUBLE
* @return default value is 0.
*/
struct SrsAmf0Number : public SrsAmf0Any
{
double value;
SrsAmf0Number(double _value = 0.0);
virtual ~SrsAmf0Number();
};
/**
* read amf0 null from stream.
* 2.7 null Type
* null-type = null-marker
*/
struct SrsAmf0Null : public SrsAmf0Any
{
SrsAmf0Null();
virtual ~SrsAmf0Null();
};
/**
* read amf0 undefined from stream.
* 2.8 undefined Type
* undefined-type = undefined-marker
*/
struct SrsAmf0Undefined : public SrsAmf0Any
{
SrsAmf0Undefined();
virtual ~SrsAmf0Undefined();
};
/**
* 2.11 Object End Type
* object-end-type = UTF-8-empty object-end-marker
* 0x00 0x00 0x09
*/
struct SrsAmf0ObjectEOF : public SrsAmf0Any
{
int16_t utf8_empty;
SrsAmf0ObjectEOF();
virtual ~SrsAmf0ObjectEOF();
};
/**
* to ensure in inserted order.
* for the FMLE will crash when AMF0Object is not ordered by inserted,
* if ordered in map, the string compare order, the FMLE will creash when
* get the response of connect app.
*/
struct SrsUnSortedHashtable
{
private:
typedef std::pair<std::string, SrsAmf0Any*> SrsObjectPropertyType;
std::vector<SrsObjectPropertyType> properties;
public:
SrsUnSortedHashtable();
virtual ~SrsUnSortedHashtable();
virtual int size();
virtual void clear();
virtual std::string key_at(int index);
virtual SrsAmf0Any* value_at(int index);
virtual void set(std::string key, SrsAmf0Any* value);
virtual SrsAmf0Any* get_property(std::string name);
virtual SrsAmf0Any* ensure_property_string(std::string name);
virtual SrsAmf0Any* ensure_property_number(std::string name);
};
/**
* 2.5 Object Type
* anonymous-object-type = object-marker *(object-property)
* object-property = (UTF-8 value-type) | (UTF-8-empty object-end-marker)
*/
struct SrsAmf0Object : public SrsAmf0Any
{
private:
SrsUnSortedHashtable properties;
public:
SrsAmf0ObjectEOF eof;
SrsAmf0Object();
virtual ~SrsAmf0Object();
virtual int size();
virtual std::string key_at(int index);
virtual SrsAmf0Any* value_at(int index);
virtual void set(std::string key, SrsAmf0Any* value);
virtual SrsAmf0Any* get_property(std::string name);
virtual SrsAmf0Any* ensure_property_string(std::string name);
virtual SrsAmf0Any* ensure_property_number(std::string name);
};
/**
* 2.10 ECMA Array Type
* ecma-array-type = associative-count *(object-property)
* associative-count = U32
* object-property = (UTF-8 value-type) | (UTF-8-empty object-end-marker)
*/
struct SrsASrsAmf0EcmaArray : public SrsAmf0Any
{
private:
SrsUnSortedHashtable properties;
public:
int32_t count;
SrsAmf0ObjectEOF eof;
SrsASrsAmf0EcmaArray();
virtual ~SrsASrsAmf0EcmaArray();
virtual int size();
virtual void clear();
virtual std::string key_at(int index);
virtual SrsAmf0Any* value_at(int index);
virtual void set(std::string key, SrsAmf0Any* value);
virtual SrsAmf0Any* get_property(std::string name);
virtual SrsAmf0Any* ensure_property_string(std::string name);
};
/**
* read amf0 utf8 string from stream.
* 1.3.1 Strings and UTF-8
* UTF-8 = U16 *(UTF8-char)
* UTF8-char = UTF8-1 | UTF8-2 | UTF8-3 | UTF8-4
* UTF8-1 = %x00-7F
* @remark only support UTF8-1 char.
*/
extern int srs_amf0_read_utf8(SrsStream* stream, std::string& value);
extern int srs_amf0_write_utf8(SrsStream* stream, std::string value);
/**
* read amf0 string from stream.
* 2.4 String Type
* string-type = string-marker UTF-8
*/
extern int srs_amf0_read_string(SrsStream* stream, std::string& value);
extern int srs_amf0_write_string(SrsStream* stream, std::string value);
/**
* read amf0 boolean from stream.
* 2.4 String Type
* boolean-type = boolean-marker U8
* 0 is false, <> 0 is true
*/
extern int srs_amf0_read_boolean(SrsStream* stream, bool& value);
extern int srs_amf0_write_boolean(SrsStream* stream, bool value);
/**
* read amf0 number from stream.
* 2.2 Number Type
* number-type = number-marker DOUBLE
*/
extern int srs_amf0_read_number(SrsStream* stream, double& value);
extern int srs_amf0_write_number(SrsStream* stream, double value);
/**
* read amf0 null from stream.
* 2.7 null Type
* null-type = null-marker
*/
extern int srs_amf0_read_null(SrsStream* stream);
extern int srs_amf0_write_null(SrsStream* stream);
/**
* read amf0 undefined from stream.
* 2.8 undefined Type
* undefined-type = undefined-marker
*/
extern int srs_amf0_read_undefined(SrsStream* stream);
extern int srs_amf0_write_undefined(SrsStream* stream);
extern int srs_amf0_read_any(SrsStream* stream, SrsAmf0Any*& value);
/**
* read amf0 object from stream.
* 2.5 Object Type
* anonymous-object-type = object-marker *(object-property)
* object-property = (UTF-8 value-type) | (UTF-8-empty object-end-marker)
*/
extern int srs_amf0_read_object(SrsStream* stream, SrsAmf0Object*& value);
extern int srs_amf0_write_object(SrsStream* stream, SrsAmf0Object* value);
/**
* read amf0 object from stream.
* 2.10 ECMA Array Type
* ecma-array-type = associative-count *(object-property)
* associative-count = U32
* object-property = (UTF-8 value-type) | (UTF-8-empty object-end-marker)
*/
extern int srs_amf0_read_ecma_array(SrsStream* stream, SrsASrsAmf0EcmaArray*& value);
extern int srs_amf0_write_ecma_array(SrsStream* stream, SrsASrsAmf0EcmaArray* value);
/**
* get amf0 objects size.
*/
extern int srs_amf0_get_utf8_size(std::string value);
extern int srs_amf0_get_string_size(std::string value);
extern int srs_amf0_get_number_size();
extern int srs_amf0_get_null_size();
extern int srs_amf0_get_undefined_size();
extern int srs_amf0_get_boolean_size();
extern int srs_amf0_get_object_size(SrsAmf0Object* obj);
extern int srs_amf0_get_ecma_array_size(SrsASrsAmf0EcmaArray* arr);
/**
* convert the any to specified object.
* @return T*, the converted object. never NULL.
* @remark, user must ensure the current object type,
* or the covert will cause assert failed.
*/
template<class T>
T* srs_amf0_convert(SrsAmf0Any* any)
{
T* p = dynamic_cast<T*>(any);
srs_assert(p != NULL);
return p;
}
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#ifndef SRS_CORE_AMF0_HPP
#define SRS_CORE_AMF0_HPP
/*
#include <srs_core_amf0.hpp>
*/
#include <srs_core.hpp>
#include <string>
#include <vector>
class SrsStream;
class SrsAmf0Object;
/**
* any amf0 value.
* 2.1 Types Overview
* value-type = number-type | boolean-type | string-type | object-type
* | null-marker | undefined-marker | reference-type | ecma-array-type
* | strict-array-type | date-type | long-string-type | xml-document-type
* | typed-object-type
*/
struct SrsAmf0Any
{
char marker;
SrsAmf0Any();
virtual ~SrsAmf0Any();
virtual bool is_string();
virtual bool is_boolean();
virtual bool is_number();
virtual bool is_null();
virtual bool is_undefined();
virtual bool is_object();
virtual bool is_object_eof();
virtual bool is_ecma_array();
};
/**
* read amf0 string from stream.
* 2.4 String Type
* string-type = string-marker UTF-8
* @return default value is empty string.
*/
struct SrsAmf0String : public SrsAmf0Any
{
std::string value;
SrsAmf0String(const char* _value = NULL);
virtual ~SrsAmf0String();
};
/**
* read amf0 boolean from stream.
* 2.4 String Type
* boolean-type = boolean-marker U8
* 0 is false, <> 0 is true
* @return default value is false.
*/
struct SrsAmf0Boolean : public SrsAmf0Any
{
bool value;
SrsAmf0Boolean(bool _value = false);
virtual ~SrsAmf0Boolean();
};
/**
* read amf0 number from stream.
* 2.2 Number Type
* number-type = number-marker DOUBLE
* @return default value is 0.
*/
struct SrsAmf0Number : public SrsAmf0Any
{
double value;
SrsAmf0Number(double _value = 0.0);
virtual ~SrsAmf0Number();
};
/**
* read amf0 null from stream.
* 2.7 null Type
* null-type = null-marker
*/
struct SrsAmf0Null : public SrsAmf0Any
{
SrsAmf0Null();
virtual ~SrsAmf0Null();
};
/**
* read amf0 undefined from stream.
* 2.8 undefined Type
* undefined-type = undefined-marker
*/
struct SrsAmf0Undefined : public SrsAmf0Any
{
SrsAmf0Undefined();
virtual ~SrsAmf0Undefined();
};
/**
* 2.11 Object End Type
* object-end-type = UTF-8-empty object-end-marker
* 0x00 0x00 0x09
*/
struct SrsAmf0ObjectEOF : public SrsAmf0Any
{
int16_t utf8_empty;
SrsAmf0ObjectEOF();
virtual ~SrsAmf0ObjectEOF();
};
/**
* to ensure in inserted order.
* for the FMLE will crash when AMF0Object is not ordered by inserted,
* if ordered in map, the string compare order, the FMLE will creash when
* get the response of connect app.
*/
struct SrsUnSortedHashtable
{
private:
typedef std::pair<std::string, SrsAmf0Any*> SrsObjectPropertyType;
std::vector<SrsObjectPropertyType> properties;
public:
SrsUnSortedHashtable();
virtual ~SrsUnSortedHashtable();
virtual int size();
virtual void clear();
virtual std::string key_at(int index);
virtual SrsAmf0Any* value_at(int index);
virtual void set(std::string key, SrsAmf0Any* value);
virtual SrsAmf0Any* get_property(std::string name);
virtual SrsAmf0Any* ensure_property_string(std::string name);
virtual SrsAmf0Any* ensure_property_number(std::string name);
};
/**
* 2.5 Object Type
* anonymous-object-type = object-marker *(object-property)
* object-property = (UTF-8 value-type) | (UTF-8-empty object-end-marker)
*/
struct SrsAmf0Object : public SrsAmf0Any
{
private:
SrsUnSortedHashtable properties;
public:
SrsAmf0ObjectEOF eof;
SrsAmf0Object();
virtual ~SrsAmf0Object();
virtual int size();
virtual std::string key_at(int index);
virtual SrsAmf0Any* value_at(int index);
virtual void set(std::string key, SrsAmf0Any* value);
virtual SrsAmf0Any* get_property(std::string name);
virtual SrsAmf0Any* ensure_property_string(std::string name);
virtual SrsAmf0Any* ensure_property_number(std::string name);
};
/**
* 2.10 ECMA Array Type
* ecma-array-type = associative-count *(object-property)
* associative-count = U32
* object-property = (UTF-8 value-type) | (UTF-8-empty object-end-marker)
*/
struct SrsASrsAmf0EcmaArray : public SrsAmf0Any
{
private:
SrsUnSortedHashtable properties;
public:
int32_t count;
SrsAmf0ObjectEOF eof;
SrsASrsAmf0EcmaArray();
virtual ~SrsASrsAmf0EcmaArray();
virtual int size();
virtual void clear();
virtual std::string key_at(int index);
virtual SrsAmf0Any* value_at(int index);
virtual void set(std::string key, SrsAmf0Any* value);
virtual SrsAmf0Any* get_property(std::string name);
virtual SrsAmf0Any* ensure_property_string(std::string name);
};
/**
* read amf0 utf8 string from stream.
* 1.3.1 Strings and UTF-8
* UTF-8 = U16 *(UTF8-char)
* UTF8-char = UTF8-1 | UTF8-2 | UTF8-3 | UTF8-4
* UTF8-1 = %x00-7F
* @remark only support UTF8-1 char.
*/
extern int srs_amf0_read_utf8(SrsStream* stream, std::string& value);
extern int srs_amf0_write_utf8(SrsStream* stream, std::string value);
/**
* read amf0 string from stream.
* 2.4 String Type
* string-type = string-marker UTF-8
*/
extern int srs_amf0_read_string(SrsStream* stream, std::string& value);
extern int srs_amf0_write_string(SrsStream* stream, std::string value);
/**
* read amf0 boolean from stream.
* 2.4 String Type
* boolean-type = boolean-marker U8
* 0 is false, <> 0 is true
*/
extern int srs_amf0_read_boolean(SrsStream* stream, bool& value);
extern int srs_amf0_write_boolean(SrsStream* stream, bool value);
/**
* read amf0 number from stream.
* 2.2 Number Type
* number-type = number-marker DOUBLE
*/
extern int srs_amf0_read_number(SrsStream* stream, double& value);
extern int srs_amf0_write_number(SrsStream* stream, double value);
/**
* read amf0 null from stream.
* 2.7 null Type
* null-type = null-marker
*/
extern int srs_amf0_read_null(SrsStream* stream);
extern int srs_amf0_write_null(SrsStream* stream);
/**
* read amf0 undefined from stream.
* 2.8 undefined Type
* undefined-type = undefined-marker
*/
extern int srs_amf0_read_undefined(SrsStream* stream);
extern int srs_amf0_write_undefined(SrsStream* stream);
extern int srs_amf0_read_any(SrsStream* stream, SrsAmf0Any*& value);
/**
* read amf0 object from stream.
* 2.5 Object Type
* anonymous-object-type = object-marker *(object-property)
* object-property = (UTF-8 value-type) | (UTF-8-empty object-end-marker)
*/
extern int srs_amf0_read_object(SrsStream* stream, SrsAmf0Object*& value);
extern int srs_amf0_write_object(SrsStream* stream, SrsAmf0Object* value);
/**
* read amf0 object from stream.
* 2.10 ECMA Array Type
* ecma-array-type = associative-count *(object-property)
* associative-count = U32
* object-property = (UTF-8 value-type) | (UTF-8-empty object-end-marker)
*/
extern int srs_amf0_read_ecma_array(SrsStream* stream, SrsASrsAmf0EcmaArray*& value);
extern int srs_amf0_write_ecma_array(SrsStream* stream, SrsASrsAmf0EcmaArray* value);
/**
* get amf0 objects size.
*/
extern int srs_amf0_get_utf8_size(std::string value);
extern int srs_amf0_get_string_size(std::string value);
extern int srs_amf0_get_number_size();
extern int srs_amf0_get_null_size();
extern int srs_amf0_get_undefined_size();
extern int srs_amf0_get_boolean_size();
extern int srs_amf0_get_object_size(SrsAmf0Object* obj);
extern int srs_amf0_get_ecma_array_size(SrsASrsAmf0EcmaArray* arr);
/**
* convert the any to specified object.
* @return T*, the converted object. never NULL.
* @remark, user must ensure the current object type,
* or the covert will cause assert failed.
*/
template<class T>
T* srs_amf0_convert(SrsAmf0Any* any)
{
T* p = dynamic_cast<T*>(any);
srs_assert(p != NULL);
return p;
}
#endif
\ No newline at end of file
... ...
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include <srs_core_autofree.hpp>
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include <srs_core_autofree.hpp>
... ...
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#ifndef SRS_CORE_AUTO_FREE_HPP
#define SRS_CORE_AUTO_FREE_HPP
/*
#include <srs_core_autofree.hpp>
*/
#include <srs_core.hpp>
/**
* auto free the instance in the current scope.
*/
#define SrsAutoFree(className, instance, is_array) \
__SrsAutoFree<className> _auto_free_##instance(&instance, is_array)
template<class T>
class __SrsAutoFree
{
private:
T** ptr;
bool is_array;
public:
/**
* auto delete the ptr.
* @is_array a bool value indicates whether the ptr is a array.
*/
__SrsAutoFree(T** _ptr, bool _is_array){
ptr = _ptr;
is_array = _is_array;
}
virtual ~__SrsAutoFree(){
if (ptr == NULL || *ptr == NULL) {
return;
}
if (is_array) {
delete[] *ptr;
} else {
delete *ptr;
}
*ptr = NULL;
}
};
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#ifndef SRS_CORE_AUTO_FREE_HPP
#define SRS_CORE_AUTO_FREE_HPP
/*
#include <srs_core_autofree.hpp>
*/
#include <srs_core.hpp>
/**
* auto free the instance in the current scope.
*/
#define SrsAutoFree(className, instance, is_array) \
__SrsAutoFree<className> _auto_free_##instance(&instance, is_array)
template<class T>
class __SrsAutoFree
{
private:
T** ptr;
bool is_array;
public:
/**
* auto delete the ptr.
* @is_array a bool value indicates whether the ptr is a array.
*/
__SrsAutoFree(T** _ptr, bool _is_array){
ptr = _ptr;
is_array = _is_array;
}
virtual ~__SrsAutoFree(){
if (ptr == NULL || *ptr == NULL) {
return;
}
if (is_array) {
delete[] *ptr;
} else {
delete *ptr;
}
*ptr = NULL;
}
};
#endif
\ No newline at end of file
... ...
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include <srs_core_buffer.hpp>
#include <srs_core_error.hpp>
#include <srs_core_socket.hpp>
#include <srs_core_log.hpp>
#define SOCKET_READ_SIZE 4096
SrsBuffer::SrsBuffer()
{
}
SrsBuffer::~SrsBuffer()
{
}
int SrsBuffer::size()
{
return (int)data.size();
}
char* SrsBuffer::bytes()
{
return &data.at(0);
}
void SrsBuffer::erase(int size)
{
data.erase(data.begin(), data.begin() + size);
}
void SrsBuffer::append(char* bytes, int size)
{
data.insert(data.end(), bytes, bytes + size);
}
int SrsBuffer::ensure_buffer_bytes(SrsSocket* skt, int required_size)
{
int ret = ERROR_SUCCESS;
if (required_size < 0) {
ret = ERROR_SYSTEM_SIZE_NEGATIVE;
srs_error("size is negative. size=%d, ret=%d", required_size, ret);
return ret;
}
while (size() < required_size) {
char buffer[SOCKET_READ_SIZE];
ssize_t nread;
if ((ret = skt->read(buffer, SOCKET_READ_SIZE, &nread)) != ERROR_SUCCESS) {
return ret;
}
srs_assert((int)nread > 0);
append(buffer, (int)nread);
}
return ret;
}
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include <srs_core_buffer.hpp>
#include <srs_core_error.hpp>
#include <srs_core_socket.hpp>
#include <srs_core_log.hpp>
#define SOCKET_READ_SIZE 4096
SrsBuffer::SrsBuffer()
{
}
SrsBuffer::~SrsBuffer()
{
}
int SrsBuffer::size()
{
return (int)data.size();
}
char* SrsBuffer::bytes()
{
return &data.at(0);
}
void SrsBuffer::erase(int size)
{
data.erase(data.begin(), data.begin() + size);
}
void SrsBuffer::append(char* bytes, int size)
{
data.insert(data.end(), bytes, bytes + size);
}
int SrsBuffer::ensure_buffer_bytes(SrsSocket* skt, int required_size)
{
int ret = ERROR_SUCCESS;
if (required_size < 0) {
ret = ERROR_SYSTEM_SIZE_NEGATIVE;
srs_error("size is negative. size=%d, ret=%d", required_size, ret);
return ret;
}
while (size() < required_size) {
char buffer[SOCKET_READ_SIZE];
ssize_t nread;
if ((ret = skt->read(buffer, SOCKET_READ_SIZE, &nread)) != ERROR_SUCCESS) {
return ret;
}
srs_assert((int)nread > 0);
append(buffer, (int)nread);
}
return ret;
}
... ...
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#ifndef SRS_CORE_BUFFER_HPP
#define SRS_CORE_BUFFER_HPP
/*
#include <srs_core_buffer.hpp>
*/
#include <srs_core.hpp>
#include <vector>
class SrsSocket;
/**
* the buffer provices bytes cache for protocol. generally,
* protocol recv data from socket, put into buffer, decode to RTMP message.
* protocol encode RTMP message to bytes, put into buffer, send to socket.
*/
class SrsBuffer
{
private:
std::vector<char> data;
public:
SrsBuffer();
virtual ~SrsBuffer();
public:
virtual int size();
virtual char* bytes();
virtual void erase(int size);
private:
virtual void append(char* bytes, int size);
public:
virtual int ensure_buffer_bytes(SrsSocket* skt, int required_size);
};
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#ifndef SRS_CORE_BUFFER_HPP
#define SRS_CORE_BUFFER_HPP
/*
#include <srs_core_buffer.hpp>
*/
#include <srs_core.hpp>
#include <vector>
class SrsSocket;
/**
* the buffer provices bytes cache for protocol. generally,
* protocol recv data from socket, put into buffer, decode to RTMP message.
* protocol encode RTMP message to bytes, put into buffer, send to socket.
*/
class SrsBuffer
{
private:
std::vector<char> data;
public:
SrsBuffer();
virtual ~SrsBuffer();
public:
virtual int size();
virtual char* bytes();
virtual void erase(int size);
private:
virtual void append(char* bytes, int size);
public:
virtual int ensure_buffer_bytes(SrsSocket* skt, int required_size);
};
#endif
\ No newline at end of file
... ...
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include <srs_core_client.hpp>
#include <arpa/inet.h>
#include <stdlib.h>
#include <srs_core_error.hpp>
#include <srs_core_log.hpp>
#include <srs_core_rtmp.hpp>
#include <srs_core_protocol.hpp>
#include <srs_core_autofree.hpp>
#include <srs_core_source.hpp>
#include <srs_core_server.hpp>
#include <srs_core_pithy_print.hpp>
#include <srs_core_config.hpp>
#include <srs_core_refer.hpp>
#define SRS_PULSE_TIMEOUT_MS 100
#define SRS_SEND_TIMEOUT_MS 5000000L
#define SRS_RECV_TIMEOUT_MS SRS_SEND_TIMEOUT_MS
SrsClient::SrsClient(SrsServer* srs_server, st_netfd_t client_stfd)
: SrsConnection(srs_server, client_stfd)
{
ip = NULL;
req = new SrsRequest();
res = new SrsResponse();
rtmp = new SrsRtmp(client_stfd);
refer = new SrsRefer();
}
SrsClient::~SrsClient()
{
srs_freepa(ip);
srs_freep(req);
srs_freep(res);
srs_freep(rtmp);
srs_freep(refer);
}
int SrsClient::do_cycle()
{
int ret = ERROR_SUCCESS;
if ((ret = get_peer_ip()) != ERROR_SUCCESS) {
srs_error("get peer ip failed. ret=%d", ret);
return ret;
}
srs_trace("get peer ip success. ip=%s, send_to=%"PRId64", recv_to=%"PRId64"",
ip, SRS_SEND_TIMEOUT_MS, SRS_RECV_TIMEOUT_MS);
rtmp->set_recv_timeout(SRS_RECV_TIMEOUT_MS * 1000);
rtmp->set_send_timeout(SRS_SEND_TIMEOUT_MS * 1000);
if ((ret = rtmp->handshake()) != ERROR_SUCCESS) {
srs_error("rtmp handshake failed. ret=%d", ret);
return ret;
}
srs_verbose("rtmp handshake success");
if ((ret = rtmp->connect_app(req)) != ERROR_SUCCESS) {
srs_error("rtmp connect vhost/app failed. ret=%d", ret);
return ret;
}
srs_verbose("rtmp connect app success");
if ((ret = check_vhost()) != ERROR_SUCCESS) {
srs_error("check vhost failed. ret=%d", ret);
return ret;
}
srs_verbose("check vhost success.");
srs_trace("rtmp connect app success. "
"tcUrl=%s, pageUrl=%s, swfUrl=%s, schema=%s, vhost=%s, port=%s, app=%s",
req->tcUrl.c_str(), req->pageUrl.c_str(), req->swfUrl.c_str(),
req->schema.c_str(), req->vhost.c_str(), req->port.c_str(),
req->app.c_str());
if ((ret = refer->check(req->pageUrl, config->get_refer(req->vhost))) != ERROR_SUCCESS) {
srs_error("check refer failed. ret=%d", ret);
return ret;
}
srs_verbose("check refer success.");
if ((ret = rtmp->set_window_ack_size(2.5 * 1000 * 1000)) != ERROR_SUCCESS) {
srs_error("set window acknowledgement size failed. ret=%d", ret);
return ret;
}
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");
if ((ret = rtmp->response_connect_app(req)) != ERROR_SUCCESS) {
srs_error("response connect app failed. ret=%d", ret);
return ret;
}
srs_verbose("response connect app success");
if ((ret = rtmp->on_bw_done()) != ERROR_SUCCESS) {
srs_error("on_bw_done failed. ret=%d", ret);
return ret;
}
srs_verbose("on_bw_done success");
SrsClientType type;
if ((ret = rtmp->identify_client(res->stream_id, type, req->stream)) != ERROR_SUCCESS) {
srs_error("identify client failed. ret=%d", ret);
return ret;
}
req->strip();
srs_trace("identify client success. type=%d, stream_name=%s", type, req->stream.c_str());
int chunk_size = 4096;
SrsConfDirective* conf = config->get_chunk_size();
if (conf && !conf->arg0().empty()) {
chunk_size = ::atoi(conf->arg0().c_str());
}
if ((ret = rtmp->set_chunk_size(chunk_size)) != ERROR_SUCCESS) {
srs_error("set chunk_size=%d failed. ret=%d", chunk_size, ret);
return ret;
}
srs_trace("set chunk_size=%d success", chunk_size);
// find a source to publish.
SrsSource* source = SrsSource::find(req->get_stream_url());
srs_assert(source != NULL);
bool enabled_cache = true;
conf = config->get_gop_cache(req->vhost);
if (conf && conf->arg0() == "off") {
enabled_cache = false;
}
source->set_cache(enabled_cache);
srs_info("source found, url=%s, enabled_cache=%d", req->get_stream_url().c_str(), enabled_cache);
switch (type) {
case SrsClientPlay: {
srs_verbose("start to play stream %s.", req->stream.c_str());
if ((ret = rtmp->start_play(res->stream_id)) != ERROR_SUCCESS) {
srs_error("start to play stream failed. ret=%d", ret);
return ret;
}
srs_info("start to play stream %s success", req->stream.c_str());
return playing(source);
}
case SrsClientFMLEPublish: {
srs_verbose("FMLE start to publish stream %s.", req->stream.c_str());
if ((ret = rtmp->start_fmle_publish(res->stream_id)) != ERROR_SUCCESS) {
srs_error("start to publish stream failed. ret=%d", ret);
return ret;
}
srs_info("start to publish stream %s success", req->stream.c_str());
ret = publish(source, true);
source->on_unpublish();
return ret;
}
case SrsClientFlashPublish: {
srs_verbose("flash start to publish stream %s.", req->stream.c_str());
if ((ret = rtmp->start_flash_publish(res->stream_id)) != ERROR_SUCCESS) {
srs_error("flash start to publish stream failed. ret=%d", ret);
return ret;
}
srs_info("flash start to publish stream %s success", req->stream.c_str());
ret = publish(source, false);
source->on_unpublish();
return ret;
}
default: {
ret = ERROR_SYSTEM_CLIENT_INVALID;
srs_info("invalid client type=%d. ret=%d", type, ret);
return ret;
}
}
return ret;
}
int SrsClient::check_vhost()
{
int ret = ERROR_SUCCESS;
srs_assert(req != NULL);
SrsConfDirective* vhost = config->get_vhost(req->vhost);
if (vhost == NULL) {
ret = ERROR_RTMP_VHOST_NOT_FOUND;
srs_error("vhost %s not found. ret=%d", req->vhost.c_str(), ret);
return ret;
}
SrsConfDirective* conf = NULL;
if ((conf = vhost->get(RTMP_VHOST_ENABLED)) != NULL && conf->arg0() != "on") {
ret = ERROR_RTMP_VHOST_NOT_FOUND;
srs_error("vhost %s disabled. ret=%d", req->vhost.c_str(), ret);
return ret;
}
if (req->vhost != vhost->arg0()) {
srs_trace("vhost change from %s to %s", req->vhost.c_str(), vhost->arg0().c_str());
req->vhost = vhost->arg0();
}
return ret;
}
int SrsClient::playing(SrsSource* source)
{
int ret = ERROR_SUCCESS;
if ((ret = refer->check(req->pageUrl, config->get_refer_play(req->vhost))) != ERROR_SUCCESS) {
srs_error("check play_refer failed. ret=%d", ret);
return ret;
}
srs_verbose("check play_refer success.");
SrsConsumer* consumer = NULL;
if ((ret = source->create_consumer(consumer)) != ERROR_SUCCESS) {
srs_error("create consumer failed. ret=%d", ret);
return ret;
}
srs_assert(consumer != NULL);
SrsAutoFree(SrsConsumer, consumer, false);
srs_verbose("consumer created success.");
rtmp->set_recv_timeout(SRS_PULSE_TIMEOUT_MS * 1000);
SrsPithyPrint pithy_print(SRS_STAGE_PLAY_USER);
while (true) {
pithy_print.elapse(SRS_PULSE_TIMEOUT_MS);
// switch to other st-threads.
st_usleep(0);
// read from client.
int ctl_msg_ret = ERROR_SUCCESS;
if (true) {
SrsCommonMessage* msg = NULL;
ctl_msg_ret = ret = rtmp->recv_message(&msg);
srs_verbose("play loop recv message. ret=%d", ret);
if (ret != ERROR_SUCCESS && ret != ERROR_SOCKET_TIMEOUT) {
srs_error("recv client control message failed. ret=%d", ret);
return ret;
}
if ((ret = process_play_control_msg(consumer, msg)) != ERROR_SUCCESS) {
srs_error("process play control message failed. ret=%d", ret);
return ret;
}
}
// get messages from consumer.
SrsSharedPtrMessage** msgs = NULL;
int count = 0;
if ((ret = consumer->get_packets(0, msgs, count)) != ERROR_SUCCESS) {
srs_error("get messages from consumer failed. ret=%d", ret);
return ret;
}
// reportable
if (pithy_print.can_print()) {
srs_trace("-> clock=%u, time=%"PRId64", cmr=%d, msgs=%d, obytes=%"PRId64", ibytes=%"PRId64", okbps=%d, ikbps=%d",
(int)(srs_get_system_time_ms()/1000), pithy_print.get_age(), ctl_msg_ret, count, rtmp->get_send_bytes(), rtmp->get_recv_bytes(), rtmp->get_send_kbps(), rtmp->get_recv_kbps());
}
if (count <= 0) {
srs_verbose("no packets in queue.");
continue;
}
SrsAutoFree(SrsSharedPtrMessage*, msgs, true);
// sendout messages
for (int i = 0; i < count; i++) {
SrsSharedPtrMessage* msg = msgs[i];
// the send_message will free the msg,
// so set the msgs[i] to NULL.
msgs[i] = NULL;
if ((ret = rtmp->send_message(msg)) != ERROR_SUCCESS) {
srs_error("send message to client failed. ret=%d", ret);
return ret;
}
}
}
return ret;
}
int SrsClient::publish(SrsSource* source, bool is_fmle)
{
int ret = ERROR_SUCCESS;
if ((ret = refer->check(req->pageUrl, config->get_refer_publish(req->vhost))) != ERROR_SUCCESS) {
srs_error("check publish_refer failed. ret=%d", ret);
return ret;
}
srs_verbose("check publish_refer success.");
SrsPithyPrint pithy_print(SRS_STAGE_PUBLISH_USER);
while (true) {
// switch to other st-threads.
st_usleep(0);
SrsCommonMessage* msg = NULL;
if ((ret = rtmp->recv_message(&msg)) != ERROR_SUCCESS) {
srs_error("recv identify client message failed. ret=%d", ret);
return ret;
}
SrsAutoFree(SrsCommonMessage, msg, false);
pithy_print.set_age(msg->header.timestamp);
// reportable
if (pithy_print.can_print()) {
srs_trace("<- clock=%u, time=%"PRId64", obytes=%"PRId64", ibytes=%"PRId64", okbps=%d, ikbps=%d",
(int)(srs_get_system_time_ms()/1000), pithy_print.get_age(), rtmp->get_send_bytes(), rtmp->get_recv_bytes(), rtmp->get_send_kbps(), rtmp->get_recv_kbps());
}
// process audio packet
if (msg->header.is_audio() && ((ret = source->on_audio(msg)) != ERROR_SUCCESS)) {
srs_error("process audio message failed. ret=%d", ret);
return ret;
}
// process video packet
if (msg->header.is_video() && ((ret = source->on_video(msg)) != ERROR_SUCCESS)) {
srs_error("process video message failed. ret=%d", ret);
return ret;
}
// process onMetaData
if (msg->header.is_amf0_data() || msg->header.is_amf3_data()) {
if ((ret = msg->decode_packet()) != ERROR_SUCCESS) {
srs_error("decode onMetaData message failed. ret=%d", ret);
return ret;
}
SrsPacket* pkt = msg->get_packet();
if (dynamic_cast<SrsOnMetaDataPacket*>(pkt)) {
SrsOnMetaDataPacket* metadata = dynamic_cast<SrsOnMetaDataPacket*>(pkt);
if ((ret = source->on_meta_data(msg, metadata)) != ERROR_SUCCESS) {
srs_error("process onMetaData message failed. ret=%d", ret);
return ret;
}
srs_trace("process onMetaData message success.");
continue;
}
srs_trace("ignore AMF0/AMF3 data message.");
continue;
}
// process UnPublish event.
if (msg->header.is_amf0_command() || msg->header.is_amf3_command()) {
if ((ret = msg->decode_packet()) != ERROR_SUCCESS) {
srs_error("decode unpublish message failed. ret=%d", ret);
return ret;
}
// flash unpublish.
if (!is_fmle) {
srs_trace("flash publish finished.");
return ret;
}
SrsPacket* pkt = msg->get_packet();
if (dynamic_cast<SrsFMLEStartPacket*>(pkt)) {
SrsFMLEStartPacket* unpublish = dynamic_cast<SrsFMLEStartPacket*>(pkt);
return rtmp->fmle_unpublish(res->stream_id, unpublish->transaction_id);
}
srs_trace("ignore AMF0/AMF3 command message.");
continue;
}
}
return ret;
}
int SrsClient::get_peer_ip()
{
int ret = ERROR_SUCCESS;
int fd = st_netfd_fileno(stfd);
// discovery client information
sockaddr_in addr;
socklen_t addrlen = sizeof(addr);
if (getpeername(fd, (sockaddr*)&addr, &addrlen) == -1) {
ret = ERROR_SOCKET_GET_PEER_NAME;
srs_error("discovery client information failed. ret=%d", ret);
return ret;
}
srs_verbose("get peer name success.");
// ip v4 or v6
char buf[INET6_ADDRSTRLEN];
memset(buf, 0, sizeof(buf));
if ((inet_ntop(addr.sin_family, &addr.sin_addr, buf, sizeof(buf))) == NULL) {
ret = ERROR_SOCKET_GET_PEER_IP;
srs_error("convert client information failed. ret=%d", ret);
return ret;
}
srs_verbose("get peer ip of client ip=%s, fd=%d", buf, fd);
ip = new char[strlen(buf) + 1];
strcpy(ip, buf);
srs_verbose("get peer ip success. ip=%s, fd=%d", ip, fd);
return ret;
}
int SrsClient::process_play_control_msg(SrsConsumer* consumer, SrsCommonMessage* msg)
{
int ret = ERROR_SUCCESS;
if (!msg) {
srs_verbose("ignore all empty message.");
return ret;
}
SrsAutoFree(SrsCommonMessage, msg, false);
if (!msg->header.is_amf0_command() && !msg->header.is_amf3_command()) {
srs_info("ignore all message except amf0/amf3 command.");
return ret;
}
if ((ret = msg->decode_packet()) != ERROR_SUCCESS) {
srs_error("decode the amf0/amf3 command packet failed. ret=%d", ret);
return ret;
}
srs_info("decode the amf0/amf3 command packet success.");
SrsPausePacket* pause = dynamic_cast<SrsPausePacket*>(msg->get_packet());
if (!pause) {
srs_info("ignore all amf0/amf3 command except pause.");
return ret;
}
if ((ret = rtmp->on_play_client_pause(res->stream_id, pause->is_pause)) != ERROR_SUCCESS) {
srs_error("rtmp process play client pause failed. ret=%d", ret);
return ret;
}
if ((ret = consumer->on_play_client_pause(pause->is_pause)) != ERROR_SUCCESS) {
srs_error("consumer process play client pause failed. ret=%d", ret);
return ret;
}
srs_info("process pause success, is_pause=%d, time=%d.", pause->is_pause, pause->time_ms);
return ret;
}
/*
The MIT License (MIT)
Copyright (c) 2013 winlin
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include <srs_core_client.hpp>
#include <arpa/inet.h>
#include <stdlib.h>
#include <srs_core_error.hpp>
#include <srs_core_log.hpp>
#include <srs_core_rtmp.hpp>
#include <srs_core_protocol.hpp>
#include <srs_core_autofree.hpp>
#include <srs_core_source.hpp>
#include <srs_core_server.hpp>
#include <srs_core_pithy_print.hpp>
#include <srs_core_config.hpp>
#include <srs_core_refer.hpp>
#define SRS_PULSE_TIMEOUT_MS 100
#define SRS_SEND_TIMEOUT_MS 5000000L
#define SRS_RECV_TIMEOUT_MS SRS_SEND_TIMEOUT_MS
SrsClient::SrsClient(SrsServer* srs_server, st_netfd_t client_stfd)
: SrsConnection(srs_server, client_stfd)
{
ip = NULL;
req = new SrsRequest();
res = new SrsResponse();
rtmp = new SrsRtmp(client_stfd);
refer = new SrsRefer();
}
SrsClient::~SrsClient()
{
srs_freepa(ip);
srs_freep(req);
srs_freep(res);
srs_freep(rtmp);
srs_freep(refer);
}
int SrsClient::do_cycle()
{
int ret = ERROR_SUCCESS;
if ((ret = get_peer_ip()) != ERROR_SUCCESS) {
srs_error("get peer ip failed. ret=%d", ret);
return ret;
}
srs_trace("get peer ip success. ip=%s, send_to=%"PRId64", recv_to=%"PRId64"",
ip, SRS_SEND_TIMEOUT_MS, SRS_RECV_TIMEOUT_MS);
rtmp->set_recv_timeout(SRS_RECV_TIMEOUT_MS * 1000);
rtmp->set_send_timeout(SRS_SEND_TIMEOUT_MS * 1000);
if ((ret = rtmp->handshake()) != ERROR_SUCCESS) {
srs_error("rtmp handshake failed. ret=%d", ret);
return ret;
}
srs_verbose("rtmp handshake success");
if ((ret = rtmp->connect_app(req)) != ERROR_SUCCESS) {
srs_error("rtmp connect vhost/app failed. ret=%d", ret);
return ret;
}
srs_verbose("rtmp connect app success");
if ((ret = check_vhost()) != ERROR_SUCCESS) {
srs_error("check vhost failed. ret=%d", ret);
return ret;
}
srs_verbose("check vhost success.");
srs_trace("rtmp connect app success. "
"tcUrl=%s, pageUrl=%s, swfUrl=%s, schema=%s, vhost=%s, port=%s, app=%s",
req->tcUrl.c_str(), req->pageUrl.c_str(), req->swfUrl.c_str(),
req->schema.c_str(), req->vhost.c_str(), req->port.c_str(),
req->app.c_str());
if ((ret = refer->check(req->pageUrl, config->get_refer(req->vhost))) != ERROR_SUCCESS) {
srs_error("check refer failed. ret=%d", ret);
return ret;
}
srs_verbose("check refer success.");
if ((ret = rtmp->set_window_ack_size(2.5 * 1000 * 1000)) != ERROR_SUCCESS) {
srs_error("set window acknowledgement size failed. ret=%d", ret);
return ret;
}
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");
if ((ret = rtmp->response_connect_app(req)) != ERROR_SUCCESS) {
srs_error("response connect app failed. ret=%d", ret);
return ret;
}
srs_verbose("response connect app success");
if ((ret = rtmp->on_bw_done()) != ERROR_SUCCESS) {
srs_error("on_bw_done failed. ret=%d", ret);
return ret;
}
srs_verbose("on_bw_done success");
SrsClientType type;
if ((ret = rtmp->identify_client(res->stream_id, type, req->stream)) != ERROR_SUCCESS) {
srs_error("identify client failed. ret=%d", ret);
return ret;
}
req->strip();
srs_trace("identify client success. type=%d, stream_name=%s", type, req->stream.c_str());
int chunk_size = 4096;
SrsConfDirective* conf = config->get_chunk_size();
if (conf && !conf->arg0().empty()) {
chunk_size = ::atoi(conf->arg0().c_str());
}
if ((ret = rtmp->set_chunk_size(chunk_size)) != ERROR_SUCCESS) {
srs_error("set chunk_size=%d failed. ret=%d", chunk_size, ret);
return ret;
}
srs_trace("set chunk_size=%d success", chunk_size);
// find a source to publish.
SrsSource* source = SrsSource::find(req->get_stream_url());
srs_assert(source != NULL);
bool enabled_cache = true;
conf = config->get_gop_cache(req->vhost);
if (conf && conf->arg0() == "off") {
enabled_cache = false;
}
source->set_cache(enabled_cache);
srs_info("source found, url=%s, enabled_cache=%d", req->get_stream_url().c_str(), enabled_cache);
switch (type) {
case SrsClientPlay: {
srs_verbose("start to play stream %s.", req->stream.c_str());
if ((ret = rtmp->start_play(res->stream_id)) != ERROR_SUCCESS) {
srs_error("start to play stream failed. ret=%d", ret);
return ret;
}
srs_info("start to play stream %s success", req->stream.c_str());
return playing(source);
}
case SrsClientFMLEPublish: {
srs_verbose("FMLE start to publish stream %s.", req->stream.c_str());
if ((ret = rtmp->start_fmle_publish(res->stream_id)) != ERROR_SUCCESS) {
srs_error("start to publish stream failed. ret=%d", ret);
return ret;
}
srs_info("start to publish stream %s success", req->stream.c_str());
ret = publish(source, true);
source->on_unpublish();
return ret;
}
case SrsClientFlashPublish: {
srs_verbose("flash start to publish stream %s.", req->stream.c_str());
if ((ret = rtmp->start_flash_publish(res->stream_id)) != ERROR_SUCCESS) {
srs_error("flash start to publish stream failed. ret=%d", ret);
return ret;
}
srs_info("flash start to publish stream %s success", req->stream.c_str());
ret = publish(source, false);
source->on_unpublish();
return ret;
}
default: {
ret = ERROR_SYSTEM_CLIENT_INVALID;
srs_info("invalid client type=%d. ret=%d", type, ret);
return ret;
}
}
return ret;
}
int SrsClient::check_vhost()
{
int ret = ERROR_SUCCESS;
srs_assert(req != NULL);
SrsConfDirective* vhost = config->get_vhost(req->vhost);
if (vhost == NULL) {
ret = ERROR_RTMP_VHOST_NOT_FOUND;
srs_error("vhost %s not found. ret=%d", req->vhost.c_str(), ret);
return ret;
}
SrsConfDirective* conf = NULL;
if ((conf = vhost->get(RTMP_VHOST_ENABLED)) != NULL && conf->arg0() != "on") {
ret = ERROR_RTMP_VHOST_NOT_FOUND;
srs_error("vhost %s disabled. ret=%d", req->vhost.c_str(), ret);
return ret;
}
if (req->vhost != vhost->arg0()) {
srs_trace("vhost change from %s to %s", req->vhost.c_str(), vhost->arg0().c_str());
req->vhost = vhost->arg0();
}
return ret;
}
int SrsClient::playing(SrsSource* source)
{
int ret = ERROR_SUCCESS;
if ((ret = refer->check(req->pageUrl, config->get_refer_play(req->vhost))) != ERROR_SUCCESS) {
srs_error("check play_refer failed. ret=%d", ret);
return ret;
}
srs_verbose("check play_refer success.");
SrsConsumer* consumer = NULL;
if ((ret = source->create_consumer(consumer)) != ERROR_SUCCESS) {
srs_error("create consumer failed. ret=%d", ret);
return ret;
}
srs_assert(consumer != NULL);
SrsAutoFree(SrsConsumer, consumer, false);
srs_verbose("consumer created success.");
rtmp->set_recv_timeout(SRS_PULSE_TIMEOUT_MS * 1000);
SrsPithyPrint pithy_print(SRS_STAGE_PLAY_USER);
while (true) {
pithy_print.elapse(SRS_PULSE_TIMEOUT_MS);
// switch to other st-threads.
st_usleep(0);
// read from client.
int ctl_msg_ret = ERROR_SUCCESS;
if (true) {
SrsCommonMessage* msg = NULL;
ctl_msg_ret = ret = rtmp->recv_message(&msg);
srs_verbose("play loop recv message. ret=%d", ret);
if (ret != ERROR_SUCCESS && ret != ERROR_SOCKET_TIMEOUT) {
srs_error("recv client control message failed. ret=%d", ret);
return ret;
}
if ((ret = process_play_control_msg(consumer, msg)) != ERROR_SUCCESS) {
srs_error("process play control message failed. ret=%d", ret);
return ret;
}
}
// get messages from consumer.
SrsSharedPtrMessage** msgs = NULL;
int count = 0;
if ((ret = consumer->get_packets(0, msgs, count)) != ERROR_SUCCESS) {
srs_error("get messages from consumer failed. ret=%d", ret);
return ret;
}
// reportable
if (pithy_print.can_print()) {
srs_trace("-> clock=%u, time=%"PRId64", cmr=%d, msgs=%d, obytes=%"PRId64", ibytes=%"PRId64", okbps=%d, ikbps=%d",
(int)(srs_get_system_time_ms()/1000), pithy_print.get_age(), ctl_msg_ret, count, rtmp->get_send_bytes(), rtmp->get_recv_bytes(), rtmp->get_send_kbps(), rtmp->get_recv_kbps());
}
if (count <= 0) {
srs_verbose("no packets in queue.");
continue;
}
SrsAutoFree(SrsSharedPtrMessage*, msgs, true);
// sendout messages
for (int i = 0; i < count; i++) {
SrsSharedPtrMessage* msg = msgs[i];
// the send_message will free the msg,
// so set the msgs[i] to NULL.
msgs[i] = NULL;
if ((ret = rtmp->send_message(msg)) != ERROR_SUCCESS) {
srs_error("send message to client failed. ret=%d", ret);
return ret;
}
}
}
return ret;
}
int SrsClient::publish(SrsSource* source, bool is_fmle)
{
int ret = ERROR_SUCCESS;
if ((ret = refer->check(req->pageUrl, config->get_refer_publish(req->vhost))) != ERROR_SUCCESS) {
srs_error("check publish_refer failed. ret=%d", ret);
return ret;
}
srs_verbose("check publish_refer success.");
SrsPithyPrint pithy_print(SRS_STAGE_PUBLISH_USER);
while (true) {
// switch to other st-threads.
st_usleep(0);
SrsCommonMessage* msg = NULL;
if ((ret = rtmp->recv_message(&msg)) != ERROR_SUCCESS) {
srs_error("recv identify client message failed. ret=%d", ret);
return ret;
}
SrsAutoFree(SrsCommonMessage, msg, false);
pithy_print.set_age(msg->header.timestamp);
// reportable
if (pithy_print.can_print()) {
srs_trace("<- clock=%u, time=%"PRId64", obytes=%"PRId64", ibytes=%"PRId64", okbps=%d, ikbps=%d",
(int)(srs_get_system_time_ms()/1000), pithy_print.get_age(), rtmp->get_send_bytes(), rtmp->get_recv_bytes(), rtmp->get_send_kbps(), rtmp->get_recv_kbps());
}
// process audio packet
if (msg->header.is_audio() && ((ret = source->on_audio(msg)) != ERROR_SUCCESS)) {
srs_error("process audio message failed. ret=%d", ret);
return ret;
}
// process video packet
if (msg->header.is_video() && ((ret = source->on_video(msg)) != ERROR_SUCCESS)) {
srs_error("process video message failed. ret=%d", ret);
return ret;
}
// process onMetaData
if (msg->header.is_amf0_data() || msg->header.is_amf3_data()) {
if ((ret = msg->decode_packet()) != ERROR_SUCCESS) {
srs_error("decode onMetaData message failed. ret=%d", ret);
return ret;
}
SrsPacket* pkt = msg->get_packet();
if (dynamic_cast<SrsOnMetaDataPacket*>(pkt)) {
SrsOnMetaDataPacket* metadata = dynamic_cast<SrsOnMetaDataPacket*>(pkt);
if ((ret = source->on_meta_data(msg, metadata)) != ERROR_SUCCESS) {
srs_error("process onMetaData message failed. ret=%d", ret);
return ret;
}
srs_trace("process onMetaData message success.");
continue;
}
srs_trace("ignore AMF0/AMF3 data message.");
continue;
}
// process UnPublish event.
if (msg->header.is_amf0_command() || msg->header.is_amf3_command()) {
if ((ret = msg->decode_packet()) != ERROR_SUCCESS) {
srs_error("decode unpublish message failed. ret=%d", ret);
return ret;
}
// flash unpublish.
if (!is_fmle) {
srs_trace("flash publish finished.");
return ret;
}
SrsPacket* pkt = msg->get_packet();
if (dynamic_cast<SrsFMLEStartPacket*>(pkt)) {
SrsFMLEStartPacket* unpublish = dynamic_cast<SrsFMLEStartPacket*>(pkt);
return rtmp->fmle_unpublish(res->stream_id, unpublish->transaction_id);
}
srs_trace("ignore AMF0/AMF3 command message.");
continue;
}
}
return ret;
}
int SrsClient::get_peer_ip()
{
int ret = ERROR_SUCCESS;
int fd = st_netfd_fileno(stfd);
// discovery client information
sockaddr_in addr;
socklen_t addrlen = sizeof(addr);
if (getpeername(fd, (sockaddr*)&addr, &addrlen) == -1) {
ret = ERROR_SOCKET_GET_PEER_NAME;
srs_error("discovery client information failed. ret=%d", ret);
return ret;
}
srs_verbose("get peer name success.");
// ip v4 or v6
char buf[INET6_ADDRSTRLEN];
memset(buf, 0, sizeof(buf));
if ((inet_ntop(addr.sin_family, &addr.sin_addr, buf, sizeof(buf))) == NULL) {
ret = ERROR_SOCKET_GET_PEER_IP;
srs_error("convert client information failed. ret=%d", ret);
return ret;
}
srs_verbose("get peer ip of client ip=%s, fd=%d", buf, fd);
ip = new char[strlen(buf) + 1];
strcpy(ip, buf);
srs_verbose("get peer ip success. ip=%s, fd=%d", ip, fd);
return ret;
}
int SrsClient::process_play_control_msg(SrsConsumer* consumer, SrsCommonMessage* msg)
{
int ret = ERROR_SUCCESS;
if (!msg) {
srs_verbose("ignore all empty message.");
return ret;
}
SrsAutoFree(SrsCommonMessage, msg, false);
if (!msg->header.is_amf0_command() && !msg->header.is_amf3_command()) {
srs_info("ignore all message except amf0/amf3 command.");
return ret;
}
if ((ret = msg->decode_packet()) != ERROR_SUCCESS) {
srs_error("decode the amf0/amf3 command packet failed. ret=%d", ret);
return ret;
}
srs_info("decode the amf0/amf3 command packet success.");
SrsPausePacket* pause = dynamic_cast<SrsPausePacket*>(msg->get_packet());
if (!pause) {
srs_info("ignore all amf0/amf3 command except pause.");
return ret;
}
if ((ret = rtmp->on_play_client_pause(res->stream_id, pause->is_pause)) != ERROR_SUCCESS) {
srs_error("rtmp process play client pause failed. ret=%d", ret);
return ret;
}
if ((ret = consumer->on_play_client_pause(pause->is_pause)) != ERROR_SUCCESS) {
srs_error("consumer process play client pause failed. ret=%d", ret);
return ret;
}
srs_info("process pause success, is_pause=%d, time=%d.", pause->is_pause, pause->time_ms);
return ret;
}
... ...