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

for #250, rename rtmp to protocol dir.

Browse source
This commit is contained in:
winlin 2015-01-23 10:07:20 +08:00
parent d8c7267cfc
commit a4ba40952a
47 changed files with 174 additions and 174 deletions
Download patch file Download diff file Expand all files Collapse all files

1894
trunk/src/protocol/srs_rtmp_amf0.cpp Normal file
View file

File diff suppressed because it is too large Load diff

840
trunk/src/protocol/srs_rtmp_amf0.hpp Normal file
View file

@ -0,0 +1,840 @@
/*
The MIT License (MIT)
Copyright (c) 2013-2015 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_RTMP_PROTOCOL_AMF0_HPP
#define SRS_RTMP_PROTOCOL_AMF0_HPP
/*
#include <srs_rtmp_amf0.hpp>
*/
#include <srs_core.hpp>
#include <string>
#include <vector>
class SrsStream;
class SrsAmf0Object;
class SrsAmf0EcmaArray;
class SrsAmf0StrictArray;
// internal objects, user should never use it.
namespace _srs_internal
{
class SrsUnSortedHashtable;
class SrsAmf0ObjectEOF;
class SrsAmf0Date;
}
/*
////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////
Usages:
1. the bytes proxy: SrsStream
// when we got some bytes from file or network,
// use SrsStream proxy to read/write bytes
// for example, read bytes from file or network.
char* bytes = ...;
// initialize the stream, proxy for bytes.
SrsStream stream;
stream.initialize(bytes);
// use stream instead.
2. directly read AMF0 any instance from stream:
SrsAmf0Any* pany = NULL;
srs_amf0_read_any(&stream, &pany);
3. use SrsAmf0Any to discovery instance from stream:
SrsAmf0Any* pany = NULL;
SrsAmf0Any::discovery(&stream, &pany);
4. directly read specified AMF0 instance value from stream:
string value;
srs_amf0_read_string(&stream, value);
5. directly read specified AMF0 instance from stream:
SrsAmf0Any* str = SrsAmf0Any::str();
str->read(&stream);
6. get value from AMF0 instance:
// parse or set by other user
SrsAmf0Any* any = ...;
if (any->is_string()) {
string str = any->to_string();
}
7. get complex object from AMF0 insance:
// parse or set by other user
SrsAmf0Any* any = ...;
if (any->is_object()) {
SrsAmf0Object* obj = any->to_object();
obj->set("width", SrsAmf0Any::number(1024));
obj->set("height", SrsAmf0Any::number(576));
}
8. serialize AMF0 instance to bytes:
// parse or set by other user
SrsAmf0Any* any = ...;
char* bytes = new char[any->total_size()];
SrsStream stream;
stream.initialize(bytes);
any->write(&stream);
@remark: for detail usage, see interfaces of each object.
@remark: all examples ignore the error process.
////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////
*/
/**
* 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
*/
class SrsAmf0Any
{
public:
char marker;
public:
SrsAmf0Any();
virtual ~SrsAmf0Any();
// type identify, user should identify the type then convert from/to value.
public:
/**
* whether current instance is an AMF0 string.
* @return true if instance is an AMF0 string; otherwise, false.
* @remark, if true, use to_string() to get its value.
*/
virtual bool is_string();
/**
* whether current instance is an AMF0 boolean.
* @return true if instance is an AMF0 boolean; otherwise, false.
* @remark, if true, use to_boolean() to get its value.
*/
virtual bool is_boolean();
/**
* whether current instance is an AMF0 number.
* @return true if instance is an AMF0 number; otherwise, false.
* @remark, if true, use to_number() to get its value.
*/
virtual bool is_number();
/**
* whether current instance is an AMF0 null.
* @return true if instance is an AMF0 null; otherwise, false.
*/
virtual bool is_null();
/**
* whether current instance is an AMF0 undefined.
* @return true if instance is an AMF0 undefined; otherwise, false.
*/
virtual bool is_undefined();
/**
* whether current instance is an AMF0 object.
* @return true if instance is an AMF0 object; otherwise, false.
* @remark, if true, use to_object() to get its value.
*/
virtual bool is_object();
/**
* whether current instance is an AMF0 object-EOF.
* @return true if instance is an AMF0 object-EOF; otherwise, false.
*/
virtual bool is_object_eof();
/**
* whether current instance is an AMF0 ecma-array.
* @return true if instance is an AMF0 ecma-array; otherwise, false.
* @remark, if true, use to_ecma_array() to get its value.
*/
virtual bool is_ecma_array();
/**
* whether current instance is an AMF0 strict-array.
* @return true if instance is an AMF0 strict-array; otherwise, false.
* @remark, if true, use to_strict_array() to get its value.
*/
virtual bool is_strict_array();
/**
* whether current instance is an AMF0 date.
* @return true if instance is an AMF0 date; otherwise, false.
* @remark, if true, use to_date() to get its value.
*/
virtual bool is_date();
/**
* whether current instance is an AMF0 object, object-EOF, ecma-array or strict-array.
*/
virtual bool is_complex_object();
// get value of instance
public:
/**
* get a string copy of instance.
* @remark assert is_string(), user must ensure the type then convert.
*/
virtual std::string to_str();
/**
* get the raw str of instance,
* user can directly set the content of str.
* @remark assert is_string(), user must ensure the type then convert.
*/
virtual const char* to_str_raw();
/**
* convert instance to amf0 boolean,
* @remark assert is_boolean(), user must ensure the type then convert.
*/
virtual bool to_boolean();
/**
* convert instance to amf0 number,
* @remark assert is_number(), user must ensure the type then convert.
*/
virtual double to_number();
/**
* convert instance to date,
* @remark assert is_date(), user must ensure the type then convert.
*/
virtual int64_t to_date();
virtual int16_t to_date_time_zone();
/**
* convert instance to amf0 object,
* @remark assert is_object(), user must ensure the type then convert.
*/
virtual SrsAmf0Object* to_object();
/**
* convert instance to ecma array,
* @remark assert is_ecma_array(), user must ensure the type then convert.
*/
virtual SrsAmf0EcmaArray* to_ecma_array();
/**
* convert instance to strict array,
* @remark assert is_strict_array(), user must ensure the type then convert.
*/
virtual SrsAmf0StrictArray* to_strict_array();
// set value of instance
public:
/**
* set the number of any when is_number() indicates true.
* user must ensure the type is a number, or assert failed.
*/
virtual void set_number(double value);
// serialize/deseriaize instance.
public:
/**
* get the size of amf0 any, including the marker size.
* the size is the bytes which instance serialized to.
*/
virtual int total_size() = 0;
/**
* read AMF0 instance from stream.
*/
virtual int read(SrsStream* stream) = 0;
/**
* write AMF0 instance to stream.
*/
virtual int write(SrsStream* stream) = 0;
/**
* copy current AMF0 instance.
*/
virtual SrsAmf0Any* copy() = 0;
/**
* human readable print
* @param pdata, output the heap data, NULL to ignore.
* @return return the *pdata for print. NULL to ignore.
* @remark user must free the data returned or output by pdata.
*/
virtual char* human_print(char** pdata, int* psize);
// create AMF0 instance.
public:
/**
* create an AMF0 string instance, set string content by value.
*/
static SrsAmf0Any* str(const char* value = NULL);
/**
* create an AMF0 boolean instance, set boolean content by value.
*/
static SrsAmf0Any* boolean(bool value = false);
/**
* create an AMF0 number instance, set number content by value.
*/
static SrsAmf0Any* number(double value = 0.0);
/**
* create an AMF0 date instance
*/
static SrsAmf0Any* date(int64_t value = 0);
/**
* create an AMF0 null instance
*/
static SrsAmf0Any* null();
/**
* create an AMF0 undefined instance
*/
static SrsAmf0Any* undefined();
/**
* create an AMF0 empty object instance
*/
static SrsAmf0Object* object();
/**
* create an AMF0 object-EOF instance
*/
static SrsAmf0Any* object_eof();
/**
* create an AMF0 empty ecma-array instance
*/
static SrsAmf0EcmaArray* ecma_array();
/**
* create an AMF0 empty strict-array instance
*/
static SrsAmf0StrictArray* strict_array();
// discovery instance from stream
public:
/**
* discovery AMF0 instance from stream
* @param ppvalue, output the discoveried AMF0 instance.
* NULL if error.
* @remark, instance is created without read from stream, user must
* use (*ppvalue)->read(stream) to get the instance.
*/
static int discovery(SrsStream* stream, SrsAmf0Any** ppvalue);
};
/**
* 2.5 Object Type
* anonymous-object-type = object-marker *(object-property)
* object-property = (UTF-8 value-type) | (UTF-8-empty object-end-marker)
*/
class SrsAmf0Object : public SrsAmf0Any
{
private:
_srs_internal::SrsUnSortedHashtable* properties;
_srs_internal::SrsAmf0ObjectEOF* eof;
private:
friend class SrsAmf0Any;
/**
* make amf0 object to private,
* use should never declare it, use SrsAmf0Any::object() to create it.
*/
SrsAmf0Object();
public:
virtual ~SrsAmf0Object();
// serialize/deserialize to/from stream.
public:
virtual int total_size();
virtual int read(SrsStream* stream);
virtual int write(SrsStream* stream);
virtual SrsAmf0Any* copy();
// properties iteration
public:
/**
* clear all propergies.
*/
virtual void clear();
/**
* get the count of properties(key:value).
*/
virtual int count();
/**
* get the property(key:value) key at index.
* @remark: max index is count().
*/
virtual std::string key_at(int index);
/**
* get the property(key:value) key raw bytes at index.
* user can directly set the key bytes.
* @remark: max index is count().
*/
virtual const char* key_raw_at(int index);
/**
* get the property(key:value) value at index.
* @remark: max index is count().
*/
virtual SrsAmf0Any* value_at(int index);
// property set/get.
public:
/**
* set the property(key:value) of object,
* @param key, string property name.
* @param value, an AMF0 instance property value.
* @remark user should never free the value, this instance will manage it.
*/
virtual void set(std::string key, SrsAmf0Any* value);
/**
* get the property(key:value) of object,
* @param name, the property name/key
* @return the property AMF0 value, NULL if not found.
* @remark user should never free the returned value, copy it if needed.
*/
virtual SrsAmf0Any* get_property(std::string name);
/**
* get the string property, ensure the property is_string().
* @return the property AMF0 value, NULL if not found, or not a string.
* @remark user should never free the returned value, copy it if needed.
*/
virtual SrsAmf0Any* ensure_property_string(std::string name);
/**
* get the number property, ensure the property is_number().
* @return the property AMF0 value, NULL if not found, or not a number.
* @remark user should never free the returned value, copy it if needed.
*/
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)
*/
class SrsAmf0EcmaArray : public SrsAmf0Any
{
private:
_srs_internal::SrsUnSortedHashtable* properties;
_srs_internal::SrsAmf0ObjectEOF* eof;
int32_t _count;
private:
friend class SrsAmf0Any;
/**
* make amf0 object to private,
* use should never declare it, use SrsAmf0Any::ecma_array() to create it.
*/
SrsAmf0EcmaArray();
public:
virtual ~SrsAmf0EcmaArray();
// serialize/deserialize to/from stream.
public:
virtual int total_size();
virtual int read(SrsStream* stream);
virtual int write(SrsStream* stream);
virtual SrsAmf0Any* copy();
// properties iteration
public:
/**
* clear all propergies.
*/
virtual void clear();
/**
* get the count of properties(key:value).
*/
virtual int count();
/**
* get the property(key:value) key at index.
* @remark: max index is count().
*/
virtual std::string key_at(int index);
/**
* get the property(key:value) key raw bytes at index.
* user can directly set the key bytes.
* @remark: max index is count().
*/
virtual const char* key_raw_at(int index);
/**
* get the property(key:value) value at index.
* @remark: max index is count().
*/
virtual SrsAmf0Any* value_at(int index);
// property set/get.
public:
/**
* set the property(key:value) of array,
* @param key, string property name.
* @param value, an AMF0 instance property value.
* @remark user should never free the value, this instance will manage it.
*/
virtual void set(std::string key, SrsAmf0Any* value);
/**
* get the property(key:value) of array,
* @param name, the property name/key
* @return the property AMF0 value, NULL if not found.
* @remark user should never free the returned value, copy it if needed.
*/
virtual SrsAmf0Any* get_property(std::string name);
/**
* get the string property, ensure the property is_string().
* @return the property AMF0 value, NULL if not found, or not a string.
* @remark user should never free the returned value, copy it if needed.
*/
virtual SrsAmf0Any* ensure_property_string(std::string name);
/**
* get the number property, ensure the property is_number().
* @return the property AMF0 value, NULL if not found, or not a number.
* @remark user should never free the returned value, copy it if needed.
*/
virtual SrsAmf0Any* ensure_property_number(std::string name);
};
/**
* 2.12 Strict Array Type
* array-count = U32
* strict-array-type = array-count *(value-type)
*/
class SrsAmf0StrictArray : public SrsAmf0Any
{
private:
std::vector<SrsAmf0Any*> properties;
int32_t _count;
private:
friend class SrsAmf0Any;
/**
* make amf0 object to private,
* use should never declare it, use SrsAmf0Any::strict_array() to create it.
*/
SrsAmf0StrictArray();
public:
virtual ~SrsAmf0StrictArray();
// serialize/deserialize to/from stream.
public:
virtual int total_size();
virtual int read(SrsStream* stream);
virtual int write(SrsStream* stream);
virtual SrsAmf0Any* copy();
// properties iteration
public:
/**
* clear all elements.
*/
virtual void clear();
/**
* get the count of elements
*/
virtual int count();
/**
* get the elements key at index.
* @remark: max index is count().
*/
virtual SrsAmf0Any* at(int index);
// property set/get.
public:
/**
* append new element to array
* @param any, an AMF0 instance property value.
* @remark user should never free the any, this instance will manage it.
*/
virtual void append(SrsAmf0Any* any);
};
/**
* the class to get amf0 object size
*/
class SrsAmf0Size
{
public:
static int utf8(std::string value);
static int str(std::string value);
static int number();
static int date();
static int null();
static int undefined();
static int boolean();
static int object(SrsAmf0Object* obj);
static int object_eof();
static int ecma_array(SrsAmf0EcmaArray* arr);
static int strict_array(SrsAmf0StrictArray* arr);
static int any(SrsAmf0Any* o);
};
/**
* read anything from stream.
* @param ppvalue, the output amf0 any elem.
* NULL if error; otherwise, never NULL and user must free it.
*/
extern int srs_amf0_read_any(SrsStream* stream, SrsAmf0Any** ppvalue);
/**
* 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);
// internal objects, user should never use it.
namespace _srs_internal
{
/**
* read amf0 string from stream.
* 2.4 String Type
* string-type = string-marker UTF-8
* @return default value is empty string.
* @remark: use SrsAmf0Any::str() to create it.
*/
class SrsAmf0String : public SrsAmf0Any
{
public:
std::string value;
private:
friend class SrsAmf0Any;
/**
* make amf0 string to private,
* use should never declare it, use SrsAmf0Any::str() to create it.
*/
SrsAmf0String(const char* _value);
public:
virtual ~SrsAmf0String();
public:
virtual int total_size();
virtual int read(SrsStream* stream);
virtual int write(SrsStream* stream);
virtual SrsAmf0Any* copy();
};
/**
* 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.
*/
class SrsAmf0Boolean : public SrsAmf0Any
{
public:
bool value;
private:
friend class SrsAmf0Any;
/**
* make amf0 boolean to private,
* use should never declare it, use SrsAmf0Any::boolean() to create it.
*/
SrsAmf0Boolean(bool _value);
public:
virtual ~SrsAmf0Boolean();
public:
virtual int total_size();
virtual int read(SrsStream* stream);
virtual int write(SrsStream* stream);
virtual SrsAmf0Any* copy();
};
/**
* read amf0 number from stream.
* 2.2 Number Type
* number-type = number-marker DOUBLE
* @return default value is 0.
*/
class SrsAmf0Number : public SrsAmf0Any
{
public:
double value;
private:
friend class SrsAmf0Any;
/**
* make amf0 number to private,
* use should never declare it, use SrsAmf0Any::number() to create it.
*/
SrsAmf0Number(double _value);
public:
virtual ~SrsAmf0Number();
public:
virtual int total_size();
virtual int read(SrsStream* stream);
virtual int write(SrsStream* stream);
virtual SrsAmf0Any* copy();
};
/**
* 2.13 Date Type
* time-zone = S16 ; reserved, not supported should be set to 0x0000
* date-type = date-marker DOUBLE time-zone
* @see: https://github.com/winlinvip/simple-rtmp-server/issues/185
*/
class SrsAmf0Date : public SrsAmf0Any
{
private:
int64_t _date_value;
int16_t _time_zone;
private:
friend class SrsAmf0Any;
/**
* make amf0 date to private,
* use should never declare it, use SrsAmf0Any::date() to create it.
*/
SrsAmf0Date(int64_t value);
public:
virtual ~SrsAmf0Date();
// serialize/deserialize to/from stream.
public:
virtual int total_size();
virtual int read(SrsStream* stream);
virtual int write(SrsStream* stream);
virtual SrsAmf0Any* copy();
public:
/**
* get the date value.
*/
virtual int64_t date();
/**
* get the time_zone.
*/
virtual int16_t time_zone();
};
/**
* read amf0 null from stream.
* 2.7 null Type
* null-type = null-marker
*/
class SrsAmf0Null : public SrsAmf0Any
{
private:
friend class SrsAmf0Any;
/**
* make amf0 null to private,
* use should never declare it, use SrsAmf0Any::null() to create it.
*/
SrsAmf0Null();
public:
virtual ~SrsAmf0Null();
public:
virtual int total_size();
virtual int read(SrsStream* stream);
virtual int write(SrsStream* stream);
virtual SrsAmf0Any* copy();
};
/**
* read amf0 undefined from stream.
* 2.8 undefined Type
* undefined-type = undefined-marker
*/
class SrsAmf0Undefined : public SrsAmf0Any
{
private:
friend class SrsAmf0Any;
/**
* make amf0 undefined to private,
* use should never declare it, use SrsAmf0Any::undefined() to create it.
*/
SrsAmf0Undefined();
public:
virtual ~SrsAmf0Undefined();
public:
virtual int total_size();
virtual int read(SrsStream* stream);
virtual int write(SrsStream* stream);
virtual SrsAmf0Any* copy();
};
/**
* 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.
*/
class SrsUnSortedHashtable
{
private:
typedef std::pair<std::string, SrsAmf0Any*> SrsAmf0ObjectPropertyType;
std::vector<SrsAmf0ObjectPropertyType> properties;
public:
SrsUnSortedHashtable();
virtual ~SrsUnSortedHashtable();
public:
virtual int count();
virtual void clear();
virtual std::string key_at(int index);
virtual const char* key_raw_at(int index);
virtual SrsAmf0Any* value_at(int index);
virtual void set(std::string key, SrsAmf0Any* value);
public:
virtual SrsAmf0Any* get_property(std::string name);
virtual SrsAmf0Any* ensure_property_string(std::string name);
virtual SrsAmf0Any* ensure_property_number(std::string name);
public:
virtual void copy(SrsUnSortedHashtable* src);
};
/**
* 2.11 Object End Type
* object-end-type = UTF-8-empty object-end-marker
* 0x00 0x00 0x09
*/
class SrsAmf0ObjectEOF : public SrsAmf0Any
{
public:
SrsAmf0ObjectEOF();
virtual ~SrsAmf0ObjectEOF();
public:
virtual int total_size();
virtual int read(SrsStream* stream);
virtual int write(SrsStream* stream);
virtual SrsAmf0Any* copy();
};
/**
* 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);
extern bool srs_amf0_is_object_eof(SrsStream* stream);
extern int srs_amf0_write_object_eof(SrsStream* stream, SrsAmf0ObjectEOF* value);
extern int srs_amf0_write_any(SrsStream* stream, SrsAmf0Any* value);
};
#endif

194
trunk/src/protocol/srs_rtmp_buffer.cpp Normal file
View file

@ -0,0 +1,194 @@
/*
The MIT License (MIT)
Copyright (c) 2013-2015 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_rtmp_buffer.hpp>
#include <srs_kernel_error.hpp>
#include <srs_kernel_log.hpp>
#include <srs_kernel_utility.hpp>
#include <srs_core_performance.hpp>
// the default recv buffer size, 128KB.
#define SRS_DEFAULT_RECV_BUFFER_SIZE 131072
// limit user-space buffer to 256KB, for 3Mbps stream delivery.
// 800*2000/8=200000B(about 195KB).
// @remark it's ok for higher stream, the buffer is ok for one chunk is 256KB.
#define SRS_MAX_SOCKET_BUFFER 262144
// the max header size,
// @see SrsProtocol::read_message_header().
#define SRS_RTMP_MAX_MESSAGE_HEADER 11
#ifdef SRS_PERF_MERGED_READ
IMergeReadHandler::IMergeReadHandler()
{
}
IMergeReadHandler::~IMergeReadHandler()
{
}
#endif
SrsFastBuffer::SrsFastBuffer()
{
#ifdef SRS_PERF_MERGED_READ
merged_read = false;
_handler = NULL;
#endif
nb_buffer = SRS_DEFAULT_RECV_BUFFER_SIZE;
buffer = new char[nb_buffer];
p = end = buffer;
}
void SrsFastBuffer::set_buffer(int buffer_size)
{
// the user-space buffer size limit to a max value.
int nb_max_buf = srs_min(buffer_size, SRS_MAX_SOCKET_BUFFER);
if (nb_max_buf < buffer_size) {
srs_warn("limit the user-space buffer from %d to %d", buffer_size, nb_max_buf);
}
// only realloc when buffer changed bigger
if (nb_max_buf <= nb_buffer) {
return;
}
int start = p - buffer;
int cap = end - p;
char* buf = new char[nb_max_buf];
if (cap > 0) {
memcpy(buf, buffer, nb_buffer);
}
srs_freep(buffer);
buffer = buf;
p = buffer + start;
end = p + cap;
}
SrsFastBuffer::~SrsFastBuffer()
{
srs_freep(buffer);
}
char SrsFastBuffer::read_1byte()
{
srs_assert(end - p >= 1);
return *p++;
}
char* SrsFastBuffer::read_slice(int size)
{
srs_assert(end - p >= size);
srs_assert(p + size > buffer);
char* ptr = p;
p += size;
// reset when consumed all.
if (p == end) {
p = end = buffer;
srs_verbose("all consumed, reset fast buffer");
}
return ptr;
}
void SrsFastBuffer::skip(int size)
{
srs_assert(end - p >= size);
srs_assert(p + size > buffer);
p += size;
}
int SrsFastBuffer::grow(ISrsBufferReader* reader, int required_size)
{
int ret = ERROR_SUCCESS;
// generally the required size is ok.
if (end - p >= required_size) {
return ret;
}
// must be positive.
srs_assert(required_size > 0);
// when read payload or there is no space to read,
// reset the buffer with exists bytes.
int max_to_read = buffer + nb_buffer - end;
if (required_size > SRS_RTMP_MAX_MESSAGE_HEADER || max_to_read < required_size) {
int nb_cap = end - p;
srs_verbose("move fast buffer %d bytes", nb_cap);
if (nb_cap < nb_buffer) {
buffer = (char*)memmove(buffer, p, nb_cap);
p = buffer;
end = p + nb_cap;
}
}
// directly check the available bytes to read in buffer.
max_to_read = buffer + nb_buffer - end;
if (max_to_read < required_size) {
ret = ERROR_READER_BUFFER_OVERFLOW;
srs_error("buffer overflow, required=%d, max=%d, ret=%d", required_size, nb_buffer, ret);
return ret;
}
// buffer is ok, read required size of bytes.
while (end - p < required_size) {
ssize_t nread;
if ((ret = reader->read(end, max_to_read, &nread)) != ERROR_SUCCESS) {
return ret;
}
#ifdef SRS_PERF_MERGED_READ
/**
* to improve read performance, merge some packets then read,
* when it on and read small bytes, we sleep to wait more data.,
* that is, we merge some data to read together.
* @see https://github.com/winlinvip/simple-rtmp-server/issues/241
*/
if (merged_read && _handler) {
_handler->on_read(nread);
}
#endif
// we just move the ptr to next.
srs_assert((int)nread > 0);
end += nread;
}
return ret;
}
#ifdef SRS_PERF_MERGED_READ
void SrsFastBuffer::set_merge_read(bool v, IMergeReadHandler* handler)
{
merged_read = v;
_handler = handler;
}
#endif

138
trunk/src/protocol/srs_rtmp_buffer.hpp Normal file
View file

@ -0,0 +1,138 @@
/*
The MIT License (MIT)
Copyright (c) 2013-2015 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_PROTOCOL_BUFFER_HPP
#define SRS_PROTOCOL_BUFFER_HPP
/*
#include <srs_rtmp_buffer.hpp>
*/
#include <srs_core.hpp>
#include <srs_rtmp_io.hpp>
#include <srs_core_performance.hpp>
#include <srs_kernel_buffer.hpp>
#ifdef SRS_PERF_MERGED_READ
/**
* to improve read performance, merge some packets then read,
* when it on and read small bytes, we sleep to wait more data.,
* that is, we merge some data to read together.
* @see https://github.com/winlinvip/simple-rtmp-server/issues/241
*/
class IMergeReadHandler
{
public:
IMergeReadHandler();
virtual ~IMergeReadHandler();
public:
/**
* when read from channel, notice the merge handler to sleep for
* some small bytes.
* @remark, it only for server-side, client srs-librtmp just ignore.
*/
virtual void on_read(ssize_t nread) = 0;
};
#endif
/**
* the buffer provices bytes cache for protocol. generally,
* protocol recv data from socket, put into buffer, decode to RTMP message.
*/
// TODO: FIXME: add utest for it.
class SrsFastBuffer
{
private:
#ifdef SRS_PERF_MERGED_READ
// the merged handler
bool merged_read;
IMergeReadHandler* _handler;
#endif
// the user-space buffer to fill by reader,
// which use fast index and reset when chunk body read ok.
// @see https://github.com/winlinvip/simple-rtmp-server/issues/248
// ptr to the current read position.
char* p;
// ptr to the content end.
char* end;
// ptr to the buffer.
// buffer <= p <= end <= buffer+nb_buffer
char* buffer;
// the max size of buffer.
int nb_buffer;
public:
SrsFastBuffer();
virtual ~SrsFastBuffer();
public:
/**
* create buffer with specifeid size.
* @param buffer the size of buffer.
* @remark when MR(SRS_PERF_MERGED_READ) disabled, always set to 8K.
* @remark when buffer changed, the previous ptr maybe invalid.
* @see https://github.com/winlinvip/simple-rtmp-server/issues/241
*/
virtual void set_buffer(int buffer_size);
public:
/**
* read 1byte from buffer, move to next bytes.
* @remark assert buffer already grow(1).
*/
virtual char read_1byte();
/**
* read a slice in size bytes, move to next bytes.
* user can use this char* ptr directly, and should never free it.
* @remark assert buffer already grow(size).
* @remark the ptr returned maybe invalid after grow(x).
*/
virtual char* read_slice(int size);
/**
* skip some bytes in buffer.
* @param size the bytes to skip. positive to next; negative to previous.
* @remark assert buffer already grow(size).
*/
virtual void skip(int size);
public:
/**
* grow buffer to the required size, loop to read from skt to fill.
* @param reader, read more bytes from reader to fill the buffer to required size.
* @param required_size, loop to fill to ensure buffer size to required.
* @return an int error code, error if required_size negative.
* @remark, we actually maybe read more than required_size, maybe 4k for example.
*/
virtual int grow(ISrsBufferReader* reader, int required_size);
public:
#ifdef SRS_PERF_MERGED_READ
/**
* to improve read performance, merge some packets then read,
* when it on and read small bytes, we sleep to wait more data.,
* that is, we merge some data to read together.
* @param v true to ename merged read.
* @param handler the handler when merge read is enabled.
* @see https://github.com/winlinvip/simple-rtmp-server/issues/241
*/
virtual void set_merge_read(bool v, IMergeReadHandler* handler);
#endif
};
#endif

1383
trunk/src/protocol/srs_rtmp_handshake.cpp Normal file
View file

File diff suppressed because it is too large Load diff

546
trunk/src/protocol/srs_rtmp_handshake.hpp Normal file
View file

@ -0,0 +1,546 @@
/*
The MIT License (MIT)
Copyright (c) 2013-2015 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_RTMP_PROTOCOL_HANDSHKAE_HPP
#define SRS_RTMP_PROTOCOL_HANDSHKAE_HPP
/*
#include <srs_rtmp_handshake.hpp>
*/
#include <srs_core.hpp>
class ISrsProtocolReaderWriter;
class SrsComplexHandshake;
class SrsHandshakeBytes;
class SrsStream;
#ifdef SRS_AUTO_SSL
// for openssl.
#include <openssl/hmac.h>
namespace _srs_internal
{
// the digest key generate size.
#define __SRS_OpensslHashSize 512
extern u_int8_t SrsGenuineFMSKey[];
extern u_int8_t SrsGenuineFPKey[];
int openssl_HMACsha256(const void* key, int key_size, const void* data, int data_size, void* digest);
int openssl_generate_key(char* public_key, int32_t size);
/**
* the DH wrapper.
*/
class SrsDH
{
private:
DH* pdh;
public:
SrsDH();
virtual ~SrsDH();
public:
/**
* initialize dh, generate the public and private key.
* @param ensure_128bytes_public_key whether ensure public key is 128bytes,
* sometimes openssl generate 127bytes public key.
* default to false to donot ensure.
*/
virtual int initialize(bool ensure_128bytes_public_key = false);
/**
* copy the public key.
* @param pkey the bytes to copy the public key.
* @param pkey_size the max public key size, output the actual public key size.
* user should never ignore this size.
* @remark, when ensure_128bytes_public_key, the size always 128.
*/
virtual int copy_public_key(char* pkey, int32_t& pkey_size);
/**
* generate and copy the shared key.
* generate the shared key with peer public key.
* @param ppkey peer public key.
* @param ppkey_size the size of ppkey.
* @param skey the computed shared key.
* @param skey_size the max shared key size, output the actual shared key size.
* user should never ignore this size.
*/
virtual int copy_shared_key(const char* ppkey, int32_t ppkey_size, char* skey, int32_t& skey_size);
private:
virtual int do_initialize();
};
/**
* the schema type.
*/
enum srs_schema_type
{
srs_schema_invalid = 2,
/**
* key-digest sequence
*/
srs_schema0 = 0,
/**
* digest-key sequence
* @remark, FMS requires the schema1(digest-key), or connect failed.
*/
//
srs_schema1 = 1,
};
/**
* 764bytes key structure
* random-data: (offset)bytes
* key-data: 128bytes
* random-data: (764-offset-128-4)bytes
* offset: 4bytes
* @see also: http://blog.csdn.net/win_lin/article/details/13006803
*/
class key_block
{
public:
// (offset)bytes
char* random0;
int random0_size;
// 128bytes
char key[128];
// (764-offset-128-4)bytes
char* random1;
int random1_size;
// 4bytes
int32_t offset;
public:
key_block();
virtual ~key_block();
public:
// parse key block from c1s1.
// if created, user must free it by srs_key_block_free
// @stream contains c1s1_key_bytes the key start bytes
int parse(SrsStream* stream);
private:
// calc the offset of key,
// the key->offset cannot be used as the offset of key.
int calc_valid_offset();
};
/**
* 764bytes digest structure
* offset: 4bytes
* random-data: (offset)bytes
* digest-data: 32bytes
* random-data: (764-4-offset-32)bytes
* @see also: http://blog.csdn.net/win_lin/article/details/13006803
*/
class digest_block
{
public:
// 4bytes
int32_t offset;
// (offset)bytes
char* random0;
int random0_size;
// 32bytes
char digest[32];
// (764-4-offset-32)bytes
char* random1;
int random1_size;
public:
digest_block();
virtual ~digest_block();
public:
// parse digest block from c1s1.
// if created, user must free it by srs_digest_block_free
// @stream contains c1s1_digest_bytes the digest start bytes
int parse(SrsStream* stream);
private:
// calc the offset of digest,
// the key->offset cannot be used as the offset of digest.
int calc_valid_offset();
};
class c1s1;
/**
* the c1s1 strategy, use schema0 or schema1.
* the template method class to defines common behaviors,
* while the concrete class to implements in schema0 or schema1.
*/
class c1s1_strategy
{
protected:
key_block key;
digest_block digest;
public:
c1s1_strategy();
virtual ~c1s1_strategy();
public:
/**
* get the scema.
*/
virtual srs_schema_type schema() = 0;
/**
* get the digest.
*/
virtual char* get_digest();
/**
* get the key.
*/
virtual char* get_key();
/**
* copy to bytes.
* @param size must be 1536.
*/
virtual int dump(c1s1* owner, char* _c1s1, int size);
/**
* server: parse the c1s1, discovery the key and digest by schema.
* use the c1_validate_digest() to valid the digest of c1.
*/
virtual int parse(char* _c1s1, int size) = 0;
public:
/**
* client: create and sign c1 by schema.
* sign the c1, generate the digest.
* calc_c1_digest(c1, schema) {
* get c1s1-joined from c1 by specified schema
* digest-data = HMACsha256(c1s1-joined, FPKey, 30)
* return digest-data;
* }
* random fill 1536bytes c1 // also fill the c1-128bytes-key
* time = time() // c1[0-3]
* version = [0x80, 0x00, 0x07, 0x02] // c1[4-7]
* schema = choose schema0 or schema1
* digest-data = calc_c1_digest(c1, schema)
* copy digest-data to c1
*/
virtual int c1_create(c1s1* owner);
/**
* server: validate the parsed c1 schema
*/
virtual int c1_validate_digest(c1s1* owner, bool& is_valid);
/**
* server: create and sign the s1 from c1.
* // decode c1 try schema0 then schema1
* c1-digest-data = get-c1-digest-data(schema0)
* if c1-digest-data equals to calc_c1_digest(c1, schema0) {
* c1-key-data = get-c1-key-data(schema0)
* schema = schema0
* } else {
* c1-digest-data = get-c1-digest-data(schema1)
* if c1-digest-data not equals to calc_c1_digest(c1, schema1) {
* switch to simple handshake.
* return
* }
* c1-key-data = get-c1-key-data(schema1)
* schema = schema1
* }
*
* // generate s1
* random fill 1536bytes s1
* time = time() // c1[0-3]
* version = [0x04, 0x05, 0x00, 0x01] // s1[4-7]
* s1-key-data=shared_key=DH_compute_key(peer_pub_key=c1-key-data)
* get c1s1-joined by specified schema
* s1-digest-data = HMACsha256(c1s1-joined, FMSKey, 36)
* copy s1-digest-data and s1-key-data to s1.
* @param c1, to get the peer_pub_key of client.
*/
virtual int s1_create(c1s1* owner, c1s1* c1);
/**
* server: validate the parsed s1 schema
*/
virtual int s1_validate_digest(c1s1* owner, bool& is_valid);
public:
/**
* calc the digest for c1
*/
virtual int calc_c1_digest(c1s1* owner, char*& c1_digest);
/**
* calc the digest for s1
*/
virtual int calc_s1_digest(c1s1* owner, char*& s1_digest);
/**
* copy whole c1s1 to bytes.
* @param size must always be 1536 with digest, and 1504 without digest.
*/
virtual int copy_to(c1s1* owner, char* bytes, int size, bool with_digest) = 0;
/**
* copy time and version to stream.
*/
virtual void copy_time_version(SrsStream* stream, c1s1* owner);
/**
* copy key to stream.
*/
virtual void copy_key(SrsStream* stream);
/**
* copy digest to stream.
*/
virtual void copy_digest(SrsStream* stream, bool with_digest);
};
/**
* c1s1 schema0
* key: 764bytes
* digest: 764bytes
*/
class c1s1_strategy_schema0 : public c1s1_strategy
{
public:
c1s1_strategy_schema0();
virtual ~c1s1_strategy_schema0();
public:
virtual srs_schema_type schema();
virtual int parse(char* _c1s1, int size);
public:
virtual int copy_to(c1s1* owner, char* bytes, int size, bool with_digest);
};
/**
* c1s1 schema1
* digest: 764bytes
* key: 764bytes
*/
class c1s1_strategy_schema1 : public c1s1_strategy
{
public:
c1s1_strategy_schema1();
virtual ~c1s1_strategy_schema1();
public:
virtual srs_schema_type schema();
virtual int parse(char* _c1s1, int size);
public:
virtual int copy_to(c1s1* owner, char* bytes, int size, bool with_digest);
};
/**
* c1s1 schema0
* time: 4bytes
* version: 4bytes
* key: 764bytes
* digest: 764bytes
* c1s1 schema1
* time: 4bytes
* version: 4bytes
* digest: 764bytes
* key: 764bytes
* @see also: http://blog.csdn.net/win_lin/article/details/13006803
*/
class c1s1
{
public:
// 4bytes
int32_t time;
// 4bytes
int32_t version;
// 764bytes+764bytes
c1s1_strategy* payload;
public:
c1s1();
virtual ~c1s1();
public:
/**
* get the scema.
*/
virtual srs_schema_type schema();
/**
* get the digest key.
*/
virtual char* get_digest();
/**
* get the key.
*/
virtual char* get_key();
public:
/**
* copy to bytes.
* @param size, must always be 1536.
*/
virtual int dump(char* _c1s1, int size);
/**
* server: parse the c1s1, discovery the key and digest by schema.
* @param size, must always be 1536.
* use the c1_validate_digest() to valid the digest of c1.
* use the s1_validate_digest() to valid the digest of s1.
*/
virtual int parse(char* _c1s1, int size, srs_schema_type _schema);
public:
/**
* client: create and sign c1 by schema.
* sign the c1, generate the digest.
* calc_c1_digest(c1, schema) {
* get c1s1-joined from c1 by specified schema
* digest-data = HMACsha256(c1s1-joined, FPKey, 30)
* return digest-data;
* }
* random fill 1536bytes c1 // also fill the c1-128bytes-key
* time = time() // c1[0-3]
* version = [0x80, 0x00, 0x07, 0x02] // c1[4-7]
* schema = choose schema0 or schema1
* digest-data = calc_c1_digest(c1, schema)
* copy digest-data to c1
*/
virtual int c1_create(srs_schema_type _schema);
/**
* server: validate the parsed c1 schema
*/
virtual int c1_validate_digest(bool& is_valid);
public:
/**
* server: create and sign the s1 from c1.
* // decode c1 try schema0 then schema1
* c1-digest-data = get-c1-digest-data(schema0)
* if c1-digest-data equals to calc_c1_digest(c1, schema0) {
* c1-key-data = get-c1-key-data(schema0)
* schema = schema0
* } else {
* c1-digest-data = get-c1-digest-data(schema1)
* if c1-digest-data not equals to calc_c1_digest(c1, schema1) {
* switch to simple handshake.
* return
* }
* c1-key-data = get-c1-key-data(schema1)
* schema = schema1
* }
*
* // generate s1
* random fill 1536bytes s1
* time = time() // c1[0-3]
* version = [0x04, 0x05, 0x00, 0x01] // s1[4-7]
* s1-key-data=shared_key=DH_compute_key(peer_pub_key=c1-key-data)
* get c1s1-joined by specified schema
* s1-digest-data = HMACsha256(c1s1-joined, FMSKey, 36)
* copy s1-digest-data and s1-key-data to s1.
*/
virtual int s1_create(c1s1* c1);
/**
* server: validate the parsed s1 schema
*/
virtual int s1_validate_digest(bool& is_valid);
};
/**
* the c2s2 complex handshake structure.
* random-data: 1504bytes
* digest-data: 32bytes
* @see also: http://blog.csdn.net/win_lin/article/details/13006803
*/
class c2s2
{
public:
char random[1504];
char digest[32];
public:
c2s2();
virtual ~c2s2();
public:
/**
* copy to bytes.
* @param size, must always be 1536.
*/
virtual int dump(char* _c2s2, int size);
/**
* parse the c2s2
* @param size, must always be 1536.
*/
virtual int parse(char* _c2s2, int size);
public:
/**
* create c2.
* random fill c2s2 1536 bytes
*
* // client generate C2, or server valid C2
* temp-key = HMACsha256(s1-digest, FPKey, 62)
* c2-digest-data = HMACsha256(c2-random-data, temp-key, 32)
*/
virtual int c2_create(c1s1* s1);
/**
* validate the c2 from client.
*/
virtual int c2_validate(c1s1* s1, bool& is_valid);
public:
/**
* create s2.
* random fill c2s2 1536 bytes
*
* // server generate S2, or client valid S2
* temp-key = HMACsha256(c1-digest, FMSKey, 68)
* s2-digest-data = HMACsha256(s2-random-data, temp-key, 32)
*/
virtual int s2_create(c1s1* c1);
/**
* validate the s2 from server.
*/
virtual int s2_validate(c1s1* c1, bool& is_valid);
};
}
#endif
/**
* simple handshake.
* user can try complex handshake first,
* rollback to simple handshake if error ERROR_RTMP_TRY_SIMPLE_HS
*/
class SrsSimpleHandshake
{
public:
SrsSimpleHandshake();
virtual ~SrsSimpleHandshake();
public:
/**
* simple handshake.
*/
virtual int handshake_with_client(SrsHandshakeBytes* hs_bytes, ISrsProtocolReaderWriter* io);
virtual int handshake_with_server(SrsHandshakeBytes* hs_bytes, ISrsProtocolReaderWriter* io);
};
/**
* rtmp complex handshake,
* @see also crtmp(crtmpserver) or librtmp,
* @see also: http://blog.csdn.net/win_lin/article/details/13006803
*/
class SrsComplexHandshake
{
public:
SrsComplexHandshake();
virtual ~SrsComplexHandshake();
public:
/**
* complex hanshake.
* @return user must:
* continue connect app if success,
* try simple handshake if error is ERROR_RTMP_TRY_SIMPLE_HS,
* otherwise, disconnect
*/
virtual int handshake_with_client(SrsHandshakeBytes* hs_bytes, ISrsProtocolReaderWriter* io);
virtual int handshake_with_server(SrsHandshakeBytes* hs_bytes, ISrsProtocolReaderWriter* io);
};
#endif

73
trunk/src/protocol/srs_rtmp_io.cpp Normal file
View file

@ -0,0 +1,73 @@
/*
The MIT License (MIT)
Copyright (c) 2013-2015 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_rtmp_io.hpp>
ISrsBufferReader::ISrsBufferReader()
{
}
ISrsBufferReader::~ISrsBufferReader()
{
}
ISrsBufferWriter::ISrsBufferWriter()
{
}
ISrsBufferWriter::~ISrsBufferWriter()
{
}
ISrsProtocolStatistic::ISrsProtocolStatistic()
{
}
ISrsProtocolStatistic::~ISrsProtocolStatistic()
{
}
ISrsProtocolReader::ISrsProtocolReader()
{
}
ISrsProtocolReader::~ISrsProtocolReader()
{
}
ISrsProtocolWriter::ISrsProtocolWriter()
{
}
ISrsProtocolWriter::~ISrsProtocolWriter()
{
}
ISrsProtocolReaderWriter::ISrsProtocolReaderWriter()
{
}
ISrsProtocolReaderWriter::~ISrsProtocolReaderWriter()
{
}

184
trunk/src/protocol/srs_rtmp_io.hpp Normal file
View file

@ -0,0 +1,184 @@
/*
The MIT License (MIT)
Copyright (c) 2013-2015 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_RTMP_PROTOCOL_IO_HPP
#define SRS_RTMP_PROTOCOL_IO_HPP
/*
#include <srs_rtmp_io.hpp>
*/
#include <srs_core.hpp>
// for srs-librtmp, @see https://github.com/winlinvip/simple-rtmp-server/issues/213
#ifndef _WIN32
#include <sys/uio.h>
#endif
/**
* the system io reader/writer architecture:
+---------------+ +--------------------+ +---------------+
| IBufferReader | | IStatistic | | IBufferWriter |
+---------------+ +--------------------+ +---------------+
| + read() | | + get_recv_bytes() | | + write() |
+------+--------+ | + get_recv_bytes() | | + writev() |
/ \ +---+--------------+-+ +-------+-------+
| / \ / \ / \
| | | |
+------+------------------+-+ +-----+----------------+--+
| IProtocolReader | | IProtocolWriter |
+---------------------------+ +-------------------------+
| + readfully() | | + set_send_timeout() |
| + set_recv_timeout() | +-------+-----------------+
+------------+--------------+ / \
/ \ |
| |
+--+-----------------------------+-+
| IProtocolReaderWriter |
+----------------------------------+
| + is_never_timeout() |
+----------------------------------+
*/
/**
* the reader for the buffer to read from whatever channel.
*/
class ISrsBufferReader
{
public:
ISrsBufferReader();
virtual ~ISrsBufferReader();
// for protocol/amf0/msg-codec
public:
virtual int read(void* buf, size_t size, ssize_t* nread) = 0;
};
/**
* the writer for the buffer to write to whatever channel.
*/
class ISrsBufferWriter
{
public:
ISrsBufferWriter();
virtual ~ISrsBufferWriter();
// for protocol
public:
/**
* write bytes over writer.
* @nwrite the actual written bytes. NULL to ignore.
*/
virtual int write(void* buf, size_t size, ssize_t* nwrite) = 0;
/**
* write iov over writer.
* @nwrite the actual written bytes. NULL to ignore.
*/
virtual int writev(const iovec *iov, int iov_size, ssize_t* nwrite) = 0;
};
/**
* get the statistic of channel.
*/
class ISrsProtocolStatistic
{
public:
ISrsProtocolStatistic();
virtual ~ISrsProtocolStatistic();
// for protocol
public:
/**
* get the total recv bytes over underlay fd.
*/
virtual int64_t get_recv_bytes() = 0;
/**
* get the total send bytes over underlay fd.
*/
virtual int64_t get_send_bytes() = 0;
};
/**
* the reader for the protocol to read from whatever channel.
*/
class ISrsProtocolReader : public virtual ISrsBufferReader, public virtual ISrsProtocolStatistic
{
public:
ISrsProtocolReader();
virtual ~ISrsProtocolReader();
// for protocol
public:
/**
* set the recv timeout in us, recv will error when timeout.
* @remark, if not set, use ST_UTIME_NO_TIMEOUT, never timeout.
*/
virtual void set_recv_timeout(int64_t timeout_us) = 0;
/**
* get the recv timeout in us.
*/
virtual int64_t get_recv_timeout() = 0;
// for handshake.
public:
/**
* read specified size bytes of data
* @param nread, the actually read size, NULL to ignore.
*/
virtual int read_fully(void* buf, size_t size, ssize_t* nread) = 0;
};
/**
* the writer for the protocol to write to whatever channel.
*/
class ISrsProtocolWriter : public virtual ISrsBufferWriter, public virtual ISrsProtocolStatistic
{
public:
ISrsProtocolWriter();
virtual ~ISrsProtocolWriter();
// for protocol
public:
/**
* set the send timeout in us, send will error when timeout.
* @remark, if not set, use ST_UTIME_NO_TIMEOUT, never timeout.
*/
virtual void set_send_timeout(int64_t timeout_us) = 0;
/**
* get the send timeout in us.
*/
virtual int64_t get_send_timeout() = 0;
};
/**
* the reader and writer.
*/
class ISrsProtocolReaderWriter : public virtual ISrsProtocolReader, public virtual ISrsProtocolWriter
{
public:
ISrsProtocolReaderWriter();
virtual ~ISrsProtocolReaderWriter();
// for protocol
public:
/**
* whether the specified timeout_us is never timeout.
*/
virtual bool is_never_timeout(int64_t timeout_us) = 0;
};
#endif

65
trunk/src/protocol/srs_rtmp_msg_array.cpp Normal file
View file

@ -0,0 +1,65 @@
/*
The MIT License (MIT)
Copyright (c) 2013-2015 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_rtmp_msg_array.hpp>
#include <srs_rtmp_stack.hpp>
SrsMessageArray::SrsMessageArray(int max_msgs)
{
srs_assert(max_msgs > 0);
msgs = new SrsSharedPtrMessage*[max_msgs];
max = max_msgs;
zero(max_msgs);
}
SrsMessageArray::~SrsMessageArray()
{
// we just free the msgs itself,
// both delete and delete[] is ok,
// for each msg in msgs is already freed by send_and_free_messages.
srs_freep(msgs);
}
void SrsMessageArray::free(int count)
{
// initialize
for (int i = 0; i < count; i++) {
SrsSharedPtrMessage* msg = msgs[i];
srs_freep(msg);
msgs[i] = NULL;
}
}
void SrsMessageArray::zero(int count)
{
// initialize
for (int i = 0; i < count; i++) {
msgs[i] = NULL;
}
}

76
trunk/src/protocol/srs_rtmp_msg_array.hpp Normal file
View file

@ -0,0 +1,76 @@
/*
The MIT License (MIT)
Copyright (c) 2013-2015 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_RTMP_PROTOCOL_MSG_ARRAY_HPP
#define SRS_RTMP_PROTOCOL_MSG_ARRAY_HPP
/*
#include <srs_rtmp_msg_array.hpp>
*/
#include <srs_core.hpp>
class SrsSharedPtrMessage;
/**
* the class to auto free the shared ptr message array.
* when need to get some messages, for instance, from Consumer queue,
* create a message array, whose msgs can used to accept the msgs,
* then send each message and set to NULL.
*
* @remark: user must free all msgs in array, for the SRS2.0 protocol stack
* provides an api to send messages, @see send_and_free_messages
*/
class SrsMessageArray
{
public:
/**
* when user already send the msg in msgs, please set to NULL,
* for instance, msg= msgs.msgs[i], msgs.msgs[i]=NULL, send(msg),
* where send(msg) will always send and free it.
*/
SrsSharedPtrMessage** msgs;
int max;
public:
/**
* create msg array, initialize array to NULL ptrs.
*/
SrsMessageArray(int max_msgs);
/**
* free the msgs not sent out(not NULL).
*/
virtual ~SrsMessageArray();
public:
/**
* free specified count of messages.
*/
virtual void free(int count);
private:
/**
* zero initialize the message array.
*/
virtual void zero(int count);
};
#endif

1516
trunk/src/protocol/srs_rtmp_sdk.cpp Normal file
View file

File diff suppressed because it is too large Load diff

541
trunk/src/protocol/srs_rtmp_sdk.hpp Normal file
View file

@ -0,0 +1,541 @@
/*
The MIT License (MIT)
Copyright (c) 2013-2015 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_RTMP_PROTOCOL_RTMP_HPP
#define SRS_RTMP_PROTOCOL_RTMP_HPP
/*
#include <srs_rtmp_sdk.hpp>
*/
#include <srs_core.hpp>
#include <string>
#include <srs_rtmp_stack.hpp>
#include <srs_core_performance.hpp>
class SrsProtocol;
class ISrsProtocolReaderWriter;
class SrsCommonMessage;
class SrsCreateStreamPacket;
class SrsFMLEStartPacket;
class SrsPublishPacket;
class SrsOnMetaDataPacket;
class SrsPlayPacket;
class SrsCommonMessage;
class SrsPacket;
class SrsAmf0Object;
class IMergeReadHandler;
/**
* the original request from client.
*/
class SrsRequest
{
public:
// client ip.
std::string ip;
public:
/**
* tcUrl: rtmp://request_vhost:port/app/stream
* support pass vhost in query string, such as:
* rtmp://ip:port/app?vhost=request_vhost/stream
* rtmp://ip:port/app...vhost...request_vhost/stream
*/
std::string tcUrl;
std::string pageUrl;
std::string swfUrl;
double objectEncoding;
// data discovery from request.
public:
// discovery from tcUrl and play/publish.
std::string schema;
// the vhost in tcUrl.
std::string vhost;
// the host in tcUrl.
std::string host;
// the port in tcUrl.
std::string port;
// the app in tcUrl, without param.
std::string app;
// the param in tcUrl(app).
std::string param;
// the stream in play/publish
std::string stream;
// for play live stream,
// used to specified the stop when exceed the duration.
// @see https://github.com/winlinvip/simple-rtmp-server/issues/45
// in ms.
double duration;
// the token in the connect request,
// used for edge traverse to origin authentication,
// @see https://github.com/winlinvip/simple-rtmp-server/issues/104
SrsAmf0Object* args;
public:
SrsRequest();
virtual ~SrsRequest();
public:
/**
* deep copy the request, for source to use it to support reload,
* for when initialize the source, the request is valid,
* when reload it, the request maybe invalid, so need to copy it.
*/
virtual SrsRequest* copy();
/**
* update the auth info of request,
* to keep the current request ptr is ok,
* for many components use the ptr of request.
*/
virtual void update_auth(SrsRequest* req);
/**
* get the stream identify, vhost/app/stream.
*/
virtual std::string get_stream_url();
/**
* strip url, user must strip when update the url.
*/
virtual void strip();
};
/**
* the response to client.
*/
class SrsResponse
{
public:
/**
* the stream id to response client createStream.
*/
int stream_id;
public:
SrsResponse();
virtual ~SrsResponse();
};
/**
* the rtmp client type.
*/
enum SrsRtmpConnType
{
SrsRtmpConnUnknown,
SrsRtmpConnPlay,
SrsRtmpConnFMLEPublish,
SrsRtmpConnFlashPublish,
};
std::string srs_client_type_string(SrsRtmpConnType type);
/**
* store the handshake bytes,
* for smart switch between complex and simple handshake.
*/
class SrsHandshakeBytes
{
public:
// [1+1536]
char* c0c1;
// [1+1536+1536]
char* s0s1s2;
// [1536]
char* c2;
public:
SrsHandshakeBytes();
virtual ~SrsHandshakeBytes();
public:
virtual int read_c0c1(ISrsProtocolReaderWriter* io);
virtual int read_s0s1s2(ISrsProtocolReaderWriter* io);
virtual int read_c2(ISrsProtocolReaderWriter* io);
virtual int create_c0c1();
virtual int create_s0s1s2(const char* c1 = NULL);
virtual int create_c2();
};
/**
* implements the client role protocol.
*/
class SrsRtmpClient
{
private:
SrsHandshakeBytes* hs_bytes;
protected:
SrsProtocol* protocol;
ISrsProtocolReaderWriter* io;
public:
SrsRtmpClient(ISrsProtocolReaderWriter* skt);
virtual ~SrsRtmpClient();
// protocol methods proxy
public:
/**
* set the recv timeout in us.
* if timeout, recv/send message return ERROR_SOCKET_TIMEOUT.
*/
virtual void set_recv_timeout(int64_t timeout_us);
/**
* set the send timeout in us.
* if timeout, recv/send message return ERROR_SOCKET_TIMEOUT.
*/
virtual void set_send_timeout(int64_t timeout_us);
/**
* get recv/send bytes.
*/
virtual int64_t get_recv_bytes();
virtual int64_t get_send_bytes();
/**
* recv a RTMP message, which is bytes oriented.
* user can use decode_message to get the decoded RTMP packet.
* @param pmsg, set the received message,
* always NULL if error,
* NULL for unknown packet but return success.
* never NULL if decode success.
* @remark, drop message when msg is empty or payload length is empty.
*/
virtual int recv_message(SrsCommonMessage** pmsg);
/**
* decode bytes oriented RTMP message to RTMP packet,
* @param ppacket, output decoded packet,
* always NULL if error, never NULL if success.
* @return error when unknown packet, error when decode failed.
*/
virtual int decode_message(SrsCommonMessage* msg, SrsPacket** ppacket);
/**
* send the RTMP message and always free it.
* user must never free or use the msg after this method,
* for it will always free the msg.
* @param msg, the msg to send out, never be NULL.
* @param stream_id, the stream id of packet to send over, 0 for control message.
*/
virtual int send_and_free_message(SrsSharedPtrMessage* msg, int stream_id);
/**
* send the RTMP message and always free it.
* user must never free or use the msg after this method,
* for it will always free the msg.
* @param msgs, the msgs to send out, never be NULL.
* @param nb_msgs, the size of msgs to send out.
* @param stream_id, the stream id of packet to send over, 0 for control message.
*/
virtual int send_and_free_messages(SrsSharedPtrMessage** msgs, int nb_msgs, int stream_id);
/**
* send the RTMP packet and always free it.
* user must never free or use the packet after this method,
* for it will always free the packet.
* @param packet, the packet to send out, never be NULL.
* @param stream_id, the stream id of packet to send over, 0 for control message.
*/
virtual int send_and_free_packet(SrsPacket* packet, int stream_id);
public:
/**
* handshake with server, try complex, then simple handshake.
*/
virtual int handshake();
/**
* only use simple handshake
*/
virtual int simple_handshake();
/**
* only use complex handshake
*/
virtual int complex_handshake();
/**
* set req to use the original request of client:
* pageUrl and swfUrl for refer antisuck.
* args for edge to origin traverse auth, @see SrsRequest.args
*/
virtual int connect_app(std::string app, std::string tc_url,
SrsRequest* req, bool debug_srs_upnode);
/**
* connect to server, get the debug srs info.
*
* @param app, the app to connect at.
* @param tc_url, the tcUrl to connect at.
* @param req, the optional req object, use the swfUrl/pageUrl if specified. NULL to ignore.
*
* SRS debug info:
* @param srs_server_ip, debug info, server ip client connected at.
* @param srs_server, server info.
* @param srs_primary, primary authors.
* @param srs_authors, authors.
* @param srs_id, int, debug info, client id in server log.
* @param srs_pid, int, debug info, server pid in log.
*/
virtual int connect_app2(
std::string app, std::string tc_url, SrsRequest* req, bool debug_srs_upnode,
std::string& srs_server_ip, std::string& srs_server, std::string& srs_primary,
std::string& srs_authors, std::string& srs_version, int& srs_id,
int& srs_pid
);
/**
* create a stream, then play/publish data over this stream.
*/
virtual int create_stream(int& stream_id);
/**
* start play stream.
*/
virtual int play(std::string stream, int stream_id);
/**
* start publish stream. use flash publish workflow:
* connect-app => create-stream => flash-publish
*/
virtual int publish(std::string stream, int stream_id);
/**
* start publish stream. use FMLE publish workflow:
* connect-app => FMLE publish
*/
virtual int fmle_publish(std::string stream, int& stream_id);
public:
/**
* expect a specified message, drop others util got specified one.
* @pmsg, user must free it. NULL if not success.
* @ppacket, store in the pmsg, user must never free it. NULL if not success.
* @remark, only when success, user can use and must free the pmsg/ppacket.
* for example:
SrsCommonMessage* msg = NULL;
SrsConnectAppResPacket* pkt = NULL;
if ((ret = srs_rtmp_expect_message<SrsConnectAppResPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
return ret;
}
// use pkt
* user should never recv message and convert it, use this method instead.
* if need to set timeout, use set timeout of SrsProtocol.
*/
template<class T>
int expect_message(SrsCommonMessage** pmsg, T** ppacket)
{
return protocol->expect_message<T>(pmsg, ppacket);
}
};
/**
* the rtmp provices rtmp-command-protocol services,
* a high level protocol, media stream oriented services,
* such as connect to vhost/app, play stream, get audio/video data.
*/
class SrsRtmpServer
{
private:
SrsHandshakeBytes* hs_bytes;
SrsProtocol* protocol;
ISrsProtocolReaderWriter* io;
public:
SrsRtmpServer(ISrsProtocolReaderWriter* skt);
virtual ~SrsRtmpServer();
// protocol methods proxy
public:
/**
* set the auto response message when recv for protocol stack.
* @param v, whether auto response message when recv message.
* @see: https://github.com/winlinvip/simple-rtmp-server/issues/217
*/
virtual void set_auto_response(bool v);
#ifdef SRS_PERF_MERGED_READ
/**
* to improve read performance, merge some packets then read,
* when it on and read small bytes, we sleep to wait more data.,
* that is, we merge some data to read together.
* @param v true to ename merged read.
* @param handler the handler when merge read is enabled.
* @see https://github.com/winlinvip/simple-rtmp-server/issues/241
*/
virtual void set_merge_read(bool v, IMergeReadHandler* handler);
/**
* create buffer with specifeid size.
* @param buffer the size of buffer.
* @remark when MR(SRS_PERF_MERGED_READ) disabled, always set to 8K.
* @remark when buffer changed, the previous ptr maybe invalid.
* @see https://github.com/winlinvip/simple-rtmp-server/issues/241
*/
virtual void set_recv_buffer(int buffer_size);
#endif
/**
* set/get the recv timeout in us.
* if timeout, recv/send message return ERROR_SOCKET_TIMEOUT.
*/
virtual void set_recv_timeout(int64_t timeout_us);
virtual int64_t get_recv_timeout();
/**
* set/get the send timeout in us.
* if timeout, recv/send message return ERROR_SOCKET_TIMEOUT.
*/
virtual void set_send_timeout(int64_t timeout_us);
virtual int64_t get_send_timeout();
/**
* get recv/send bytes.
*/
virtual int64_t get_recv_bytes();
virtual int64_t get_send_bytes();
/**
* recv a RTMP message, which is bytes oriented.
* user can use decode_message to get the decoded RTMP packet.
* @param pmsg, set the received message,
* always NULL if error,
* NULL for unknown packet but return success.
* never NULL if decode success.
* @remark, drop message when msg is empty or payload length is empty.
*/
virtual int recv_message(SrsCommonMessage** pmsg);
/**
* decode bytes oriented RTMP message to RTMP packet,
* @param ppacket, output decoded packet,
* always NULL if error, never NULL if success.
* @return error when unknown packet, error when decode failed.
*/
virtual int decode_message(SrsCommonMessage* msg, SrsPacket** ppacket);
/**
* send the RTMP message and always free it.
* user must never free or use the msg after this method,
* for it will always free the msg.
* @param msg, the msg to send out, never be NULL.
* @param stream_id, the stream id of packet to send over, 0 for control message.
*/
virtual int send_and_free_message(SrsSharedPtrMessage* msg, int stream_id);
/**
* send the RTMP message and always free it.
* user must never free or use the msg after this method,
* for it will always free the msg.
* @param msgs, the msgs to send out, never be NULL.
* @param nb_msgs, the size of msgs to send out.
* @param stream_id, the stream id of packet to send over, 0 for control message.
*
* @remark performance issue, to support 6k+ 250kbps client,
* @see https://github.com/winlinvip/simple-rtmp-server/issues/194
*/
virtual int send_and_free_messages(SrsSharedPtrMessage** msgs, int nb_msgs, int stream_id);
/**
* send the RTMP packet and always free it.
* user must never free or use the packet after this method,
* for it will always free the packet.
* @param packet, the packet to send out, never be NULL.
* @param stream_id, the stream id of packet to send over, 0 for control message.
*/
virtual int send_and_free_packet(SrsPacket* packet, int stream_id);
public:
/**
* handshake with client, try complex then simple.
*/
virtual int handshake();
/**
* do connect app with client, to discovery tcUrl.
*/
virtual int connect_app(SrsRequest* req);
/**
* set ack size to client, client will send ack-size for each ack window
*/
virtual int set_window_ack_size(int ack_size);
/**
* @type: The sender can mark this message hard (0), soft (1), or dynamic (2)
* using the Limit type field.
*/
virtual int set_peer_bandwidth(int bandwidth, int type);
/**
* @param server_ip the ip of server.
*/
virtual int response_connect_app(SrsRequest* req, const char* server_ip = NULL);
/**
* reject the connect app request.
*/
virtual void response_connect_reject(SrsRequest* req, const char* desc);
/**
* response client the onBWDone message.
*/
virtual int on_bw_done();
/**
* recv some message to identify the client.
* @stream_id, client will createStream to play or publish by flash,
* the stream_id used to response the createStream request.
* @type, output the client type.
* @stream_name, output the client publish/play stream name. @see: SrsRequest.stream
* @duration, output the play client duration. @see: SrsRequest.duration
*/
virtual int identify_client(int stream_id, SrsRtmpConnType& type, std::string& stream_name, double& duration);
/**
* set the chunk size when client type identified.
*/
virtual int set_chunk_size(int chunk_size);
/**
* when client type is play, response with packets:
* StreamBegin,
* onStatus(NetStream.Play.Reset), onStatus(NetStream.Play.Start).,
* |RtmpSampleAccess(false, false),
* onStatus(NetStream.Data.Start).
*/
virtual int start_play(int stream_id);
/**
* when client(type is play) send pause message,
* if is_pause, response the following packets:
* onStatus(NetStream.Pause.Notify)
* StreamEOF
* if not is_pause, response the following packets:
* onStatus(NetStream.Unpause.Notify)
* StreamBegin
*/
virtual int on_play_client_pause(int stream_id, bool is_pause);
/**
* when client type is publish, response with packets:
* releaseStream response
* FCPublish
* FCPublish response
* createStream response
* onFCPublish(NetStream.Publish.Start)
* onStatus(NetStream.Publish.Start)
*/
virtual int start_fmle_publish(int stream_id);
/**
* process the FMLE unpublish event.
* @unpublish_tid the unpublish request transaction id.
*/
virtual int fmle_unpublish(int stream_id, double unpublish_tid);
/**
* when client type is publish, response with packets:
* onStatus(NetStream.Publish.Start)
*/
virtual int start_flash_publish(int stream_id);
public:
/**
* expect a specified message, drop others util got specified one.
* @pmsg, user must free it. NULL if not success.
* @ppacket, store in the pmsg, user must never free it. NULL if not success.
* @remark, only when success, user can use and must free the pmsg/ppacket.
* for example:
SrsCommonMessage* msg = NULL;
SrsConnectAppResPacket* pkt = NULL;
if ((ret = srs_rtmp_expect_message<SrsConnectAppResPacket>(protocol, &msg, &pkt)) != ERROR_SUCCESS) {
return ret;
}
// use pkt
* user should never recv message and convert it, use this method instead.
* if need to set timeout, use set timeout of SrsProtocol.
*/
template<class T>
int expect_message(SrsCommonMessage** pmsg, T** ppacket)
{
return protocol->expect_message<T>(pmsg, ppacket);
}
private:
virtual int identify_create_stream_client(SrsCreateStreamPacket* req, int stream_id, SrsRtmpConnType& type, std::string& stream_name, double& duration);
virtual int identify_fmle_publish_client(SrsFMLEStartPacket* req, SrsRtmpConnType& type, std::string& stream_name);
virtual int identify_flash_publish_client(SrsPublishPacket* req, SrsRtmpConnType& type, std::string& stream_name);
private:
virtual int identify_play_client(SrsPlayPacket* req, SrsRtmpConnType& type, std::string& stream_name, double& duration);
};
#endif

4137
trunk/src/protocol/srs_rtmp_stack.cpp Normal file
View file

File diff suppressed because it is too large Load diff

1740
trunk/src/protocol/srs_rtmp_stack.hpp Normal file
View file

File diff suppressed because it is too large Load diff

289
trunk/src/protocol/srs_rtmp_utility.cpp Normal file
View file

@ -0,0 +1,289 @@
/*
The MIT License (MIT)
Copyright (c) 2013-2015 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_rtmp_utility.hpp>
#include <stdlib.h>
using namespace std;
#include <srs_kernel_log.hpp>
#include <srs_kernel_utility.hpp>
#include <srs_kernel_stream.hpp>
#include <srs_rtmp_stack.hpp>
void srs_discovery_tc_url(
string tcUrl,
string& schema, string& host, string& vhost,
string& app, string& port, std::string& param
) {
size_t pos = std::string::npos;
std::string url = tcUrl;
if ((pos = url.find("://")) != std::string::npos) {
schema = url.substr(0, pos);
url = url.substr(schema.length() + 3);
srs_info("discovery schema=%s", schema.c_str());
}
if ((pos = url.find("/")) != std::string::npos) {
host = url.substr(0, pos);
url = url.substr(host.length() + 1);
srs_info("discovery host=%s", host.c_str());
}
port = SRS_CONSTS_RTMP_DEFAULT_PORT;
if ((pos = host.find(":")) != std::string::npos) {
port = host.substr(pos + 1);
host = host.substr(0, pos);
srs_info("discovery host=%s, port=%s", host.c_str(), port.c_str());
}
app = url;
vhost = host;
srs_vhost_resolve(vhost, app, param);
}
void srs_vhost_resolve(string& vhost, string& app, string& param)
{
// get original param
size_t pos = 0;
if ((pos = app.find("?")) != std::string::npos) {
param = app.substr(pos);
}
// filter tcUrl
app = srs_string_replace(app, ",", "?");
app = srs_string_replace(app, "...", "?");
app = srs_string_replace(app, "&&", "?");
app = srs_string_replace(app, "=", "?");
if ((pos = app.find("?")) == std::string::npos) {
return;
}
std::string query = app.substr(pos + 1);
app = app.substr(0, pos);
if ((pos = query.find("vhost?")) != std::string::npos) {
query = query.substr(pos + 6);
if (!query.empty()) {
vhost = query;
}
if ((pos = vhost.find("?")) != std::string::npos) {
vhost = vhost.substr(0, pos);
}
}
}
void srs_random_generate(char* bytes, int size)
{
static bool _random_initialized = false;
if (!_random_initialized) {
srand(0);
_random_initialized = true;
srs_trace("srand initialized the random.");
}
for (int i = 0; i < size; i++) {
// the common value in [0x0f, 0xf0]
bytes[i] = 0x0f + (rand() % (256 - 0x0f - 0x0f));
}
}
string srs_generate_tc_url(string ip, string vhost, string app, string port, string param)
{
string tcUrl = "rtmp://";
if (vhost == SRS_CONSTS_RTMP_DEFAULT_VHOST) {
tcUrl += ip;
} else {
tcUrl += vhost;
}
if (port != SRS_CONSTS_RTMP_DEFAULT_PORT) {
tcUrl += ":";
tcUrl += port;
}
tcUrl += "/";
tcUrl += app;
tcUrl += param;
return tcUrl;
}
/**
* compare the memory in bytes.
*/
bool srs_bytes_equals(void* pa, void* pb, int size)
{
u_int8_t* a = (u_int8_t*)pa;
u_int8_t* b = (u_int8_t*)pb;
if (!a && !b) {
return true;
}
if (!a || !b) {
return false;
}
for(int i = 0; i < size; i++){
if(a[i] != b[i]){
return false;
}
}
return true;
}
int srs_chunk_header_c0(
int perfer_cid, u_int32_t timestamp, int32_t payload_length,
int8_t message_type, int32_t stream_id,
char* cache, int nb_cache
)
{
// to directly set the field.
char* pp = NULL;
// generate the header.
char* p = cache;
// no header.
if (nb_cache < SRS_CONSTS_RTMP_MAX_FMT0_HEADER_SIZE) {
return 0;
}
// write new chunk stream header, fmt is 0
*p++ = 0x00 | (perfer_cid & 0x3F);
// chunk message header, 11 bytes
// timestamp, 3bytes, big-endian
if (timestamp < RTMP_EXTENDED_TIMESTAMP) {
pp = (char*)&timestamp;
*p++ = pp[2];
*p++ = pp[1];
*p++ = pp[0];
} else {
*p++ = 0xFF;
*p++ = 0xFF;
*p++ = 0xFF;
}
// message_length, 3bytes, big-endian
pp = (char*)&payload_length;
*p++ = pp[2];
*p++ = pp[1];
*p++ = pp[0];
// message_type, 1bytes
*p++ = message_type;
// stream_id, 4bytes, little-endian
pp = (char*)&stream_id;
*p++ = pp[0];
*p++ = pp[1];
*p++ = pp[2];
*p++ = pp[3];
// for c0
// chunk extended timestamp header, 0 or 4 bytes, big-endian
//
// for c3:
// chunk extended timestamp header, 0 or 4 bytes, big-endian
// 6.1.3. Extended Timestamp
// This field is transmitted only when the normal time stamp in the
// chunk message header is set to 0x00ffffff. If normal time stamp is
// set to any value less than 0x00ffffff, this field MUST NOT be
// present. This field MUST NOT be present if the timestamp field is not
// present. Type 3 chunks MUST NOT have this field.
// adobe changed for Type3 chunk:
// FMLE always sendout the extended-timestamp,
// must send the extended-timestamp to FMS,
// must send the extended-timestamp to flash-player.
// @see: ngx_rtmp_prepare_message
// @see: http://blog.csdn.net/win_lin/article/details/13363699
// TODO: FIXME: extract to outer.
if (timestamp >= RTMP_EXTENDED_TIMESTAMP) {
pp = (char*)&timestamp;
*p++ = pp[3];
*p++ = pp[2];
*p++ = pp[1];
*p++ = pp[0];
}
// always has header
return p - cache;
}
int srs_chunk_header_c3(
int perfer_cid, u_int32_t timestamp,
char* cache, int nb_cache
)
{
// to directly set the field.
char* pp = NULL;
// generate the header.
char* p = cache;
// no header.
if (nb_cache < SRS_CONSTS_RTMP_MAX_FMT3_HEADER_SIZE) {
return 0;
}
// write no message header chunk stream, fmt is 3
// @remark, if perfer_cid > 0x3F, that is, use 2B/3B chunk header,
// SRS will rollback to 1B chunk header.
*p++ = 0xC0 | (perfer_cid & 0x3F);
// for c0
// chunk extended timestamp header, 0 or 4 bytes, big-endian
//
// for c3:
// chunk extended timestamp header, 0 or 4 bytes, big-endian
// 6.1.3. Extended Timestamp
// This field is transmitted only when the normal time stamp in the
// chunk message header is set to 0x00ffffff. If normal time stamp is
// set to any value less than 0x00ffffff, this field MUST NOT be
// present. This field MUST NOT be present if the timestamp field is not
// present. Type 3 chunks MUST NOT have this field.
// adobe changed for Type3 chunk:
// FMLE always sendout the extended-timestamp,
// must send the extended-timestamp to FMS,
// must send the extended-timestamp to flash-player.
// @see: ngx_rtmp_prepare_message
// @see: http://blog.csdn.net/win_lin/article/details/13363699
// TODO: FIXME: extract to outer.
if (timestamp >= RTMP_EXTENDED_TIMESTAMP) {
pp = (char*)&timestamp;
*p++ = pp[3];
*p++ = pp[2];
*p++ = pp[1];
*p++ = pp[0];
}
// always has header
return p - cache;
}

114
trunk/src/protocol/srs_rtmp_utility.hpp Normal file
View file

@ -0,0 +1,114 @@
/*
The MIT License (MIT)
Copyright (c) 2013-2015 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_RTMP_PROTOCOL_CONSTS_HPP
#define SRS_RTMP_PROTOCOL_CONSTS_HPP
/*
#include <srs_rtmp_utility.hpp>
*/
#include <srs_core.hpp>
#include <string>
#include <srs_kernel_consts.hpp>
class SrsMessageHeader;
/**
* parse the tcUrl, output the schema, host, vhost, app and port.
* @param tcUrl, the input tcUrl, for example,
* rtmp://192.168.1.10:19350/live?vhost=vhost.ossrs.net
* @param schema, for example, rtmp
* @param host, for example, 192.168.1.10
* @param vhost, for example, vhost.ossrs.net.
* vhost default to host, when user not set vhost in query of app.
* @param app, for example, live
* @param port, for example, 19350
* default to 1935 if not specified.
* param param, for example, vhost=vhost.ossrs.net
*/
extern void srs_discovery_tc_url(
std::string tcUrl,
std::string& schema, std::string& host, std::string& vhost,
std::string& app, std::string& port, std::string& param
);
/**
* resolve the vhost in query string
* @pram vhost, update the vhost if query contains the vhost.
* @param app, may contains the vhost in query string format:
* app?vhost=request_vhost
* app...vhost...request_vhost
* @param param, the query, for example, ?vhost=xxx
*/
extern void srs_vhost_resolve(std::string& vhost, std::string& app, std::string& param);
/**
* generate ramdom data for handshake.
*/
extern void srs_random_generate(char* bytes, int size);
/**
* generate the tcUrl.
* @param param, the app parameters in tcUrl. for example, ?key=xxx,vhost=xxx
* @return the tcUrl generated from ip/vhost/app/port.
* @remark when vhost equals to __defaultVhost__, use ip as vhost.
* @remark ignore port if port equals to default port 1935.
*/
extern std::string srs_generate_tc_url(
std::string ip, std::string vhost, std::string app, std::string port,
std::string param
);
/**
* compare the memory in bytes.
* @return true if completely equal; otherwise, false.
*/
extern bool srs_bytes_equals(void* pa, void* pb, int size);
/**
* generate the c0 chunk header for msg.
* @param cache, the cache to write header.
* @param nb_cache, the size of cache.
* @return the size of header. 0 if cache not enough.
*/
extern int srs_chunk_header_c0(
int perfer_cid, u_int32_t timestamp, int32_t payload_length,
int8_t message_type, int32_t stream_id,
char* cache, int nb_cache
);
/**
* generate the c3 chunk header for msg.
* @param cache, the cache to write header.
* @param nb_cache, the size of cache.
* @return the size of header. 0 if cache not enough.
*/
extern int srs_chunk_header_c3(
int perfer_cid, u_int32_t timestamp,
char* cache, int nb_cache
);
#endif