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

merged master

Browse source
This commit is contained in:
wenjiegit 2013-12-13 02:41:31 +08:00
commit c87e55a2cc
45 changed files with 2155 additions and 1144 deletions
Download patch file Download diff file Expand all files Collapse all files

14
trunk/src/core/srs_core.cpp
View file

@ -111,3 +111,17 @@ void srs_vhost_resolve(std::string& vhost, std::string& app)
}
}
}
void srs_close_stfd(st_netfd_t& stfd)
{
if (stfd) {
int fd = st_netfd_fileno(stfd);
st_netfd_close(stfd);
stfd = NULL;
// st does not close it sometimes,
// close it manually.
close(fd);
}
}

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

@ -46,6 +46,8 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <stddef.h>
#include <sys/types.h>
#include <st.h>
// generated by configure.
#include <srs_auto_headers.hpp>
@ -102,4 +104,7 @@ extern std::string srs_dns_resolve(std::string host);
// app...vhost...request_vhost
extern void srs_vhost_resolve(std::string& vhost, std::string& app);
// close the netfd, and close the underlayer fd.
extern void srs_close_stfd(st_netfd_t& stfd);
#endif

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

@ -26,6 +26,8 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <arpa/inet.h>
#include <stdlib.h>
using namespace std;
#include <srs_core_error.hpp>
#include <srs_core_log.hpp>
#include <srs_core_rtmp.hpp>
@ -55,10 +57,14 @@ SrsClient::SrsClient(SrsServer* srs_server, st_netfd_t client_stfd)
#ifdef SRS_HTTP
http_hooks = new SrsHttpHooks();
#endif
config->subscribe(this);
}
SrsClient::~SrsClient()
{
config->unsubscribe(this);
srs_freepa(ip);
srs_freep(req);
srs_freep(res);
@ -113,6 +119,23 @@ int SrsClient::do_cycle()
return ret;
}
int SrsClient::on_reload_vhost_removed(string vhost)
{
int ret = ERROR_SUCCESS;
if (req->vhost != vhost) {
return ret;
}
// if the vhost connected is removed, disconnect the client.
srs_trace("vhost %s removed/disabled, close client url=%s",
vhost.c_str(), req->get_stream_url().c_str());
srs_close_stfd(stfd);
return ret;
}
int SrsClient::service_cycle()
{
@ -180,11 +203,7 @@ int SrsClient::service_cycle()
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(req->vhost);
if (conf && !conf->arg0().empty()) {
chunk_size = ::atoi(conf->arg0().c_str());
}
int chunk_size = config->get_chunk_size();
if ((ret = rtmp->set_chunk_size(chunk_size)) != ERROR_SUCCESS) {
srs_error("set chunk_size=%d failed. ret=%d", chunk_size, ret);
return ret;
@ -192,7 +211,7 @@ int SrsClient::service_cycle()
srs_trace("set chunk_size=%d success", chunk_size);
// find a source to publish.
SrsSource* source = SrsSource::find(req->get_stream_url());
SrsSource* source = SrsSource::find(req);
srs_assert(source != NULL);
// check publish available.
@ -205,14 +224,9 @@ int SrsClient::service_cycle()
return ret;
}
bool enabled_cache = true;
conf = config->get_gop_cache(req->vhost);
if (conf && conf->arg0() == "off") {
enabled_cache = false;
}
source->set_cache(enabled_cache);
bool enabled_cache = config->get_gop_cache(req->vhost);
srs_info("source found, url=%s, enabled_cache=%d", req->get_stream_url().c_str(), enabled_cache);
source->set_cache(enabled_cache);
switch (type) {
case SrsClientPlay: {

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

@ -31,6 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <srs_core.hpp>
#include <srs_core_conn.hpp>
#include <srs_core_reload.hpp>
class SrsRtmp;
class SrsRequest;
@ -46,7 +47,7 @@ class SrsHttpHooks;
/**
* the client provides the main logic control for RTMP clients.
*/
class SrsClient : public SrsConnection
class SrsClient : public SrsConnection, public ISrsReloadHandler
{
private:
char* ip;
@ -62,6 +63,9 @@ public:
virtual ~SrsClient();
protected:
virtual int do_cycle();
// interface ISrsReloadHandler
public:
virtual int on_reload_vhost_removed(std::string vhost);
private:
// when valid and connected to vhost/app, service the client.
virtual int service_cycle();

713
trunk/src/core/srs_core_config.cpp
View file

File diff suppressed because it is too large Load diff

146
trunk/src/core/srs_core_config.hpp
View file

@ -48,25 +48,20 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#define SRS_CONF_DEFAULT_AAC_SYNC 100
// in ms, for HLS aac flush the audio
#define SRS_CONF_DEFAULT_AAC_DELAY 300
// in seconds, the live queue length.
#define SRS_CONF_DEFAULT_QUEUE_LENGTH 30
// in seconds, the paused queue length.
#define SRS_CONF_DEFAULT_PAUSED_LENGTH 10
class SrsFileBuffer
{
public:
int fd;
int line;
// start of buffer.
char* start;
// end of buffer.
char* end;
// current consumed position.
char* pos;
// last available position.
char* last;
SrsFileBuffer();
virtual ~SrsFileBuffer();
virtual int open(const char* filename);
};
#define SRS_CONF_DEFAULT_CHUNK_SIZE 4096
#define SRS_STAGE_PLAY_USER_INTERVAL_MS 1300
#define SRS_STAGE_PUBLISH_USER_INTERVAL_MS 1100
#define SRS_STAGE_FORWARDER_INTERVAL_MS 2000
#define SRS_STAGE_ENCODER_INTERVAL_MS 2000
#define SRS_STAGE_HLS_INTERVAL_MS 2000
class SrsFileBuffer;
class SrsConfDirective
{
@ -83,13 +78,13 @@ public:
std::string arg2();
SrsConfDirective* at(int index);
SrsConfDirective* get(std::string _name);
SrsConfDirective* get(std::string _name, std::string _arg0);
public:
virtual int parse(const char* filename);
public:
enum SrsDirectiveType{parse_file, parse_block};
virtual int parse_conf(SrsFileBuffer* buffer, SrsDirectiveType type);
virtual int read_token(SrsFileBuffer* buffer, std::vector<std::string>& args);
virtual int refill_buffer(SrsFileBuffer* buffer, bool d_quoted, bool s_quoted, int startline, char*& pstart);
};
/**
@ -106,71 +101,76 @@ private:
bool show_version;
std::string config_file;
SrsConfDirective* root;
std::vector<SrsReloadHandler*> subscribes;
std::vector<ISrsReloadHandler*> subscribes;
public:
SrsConfig();
virtual ~SrsConfig();
public:
virtual int reload();
virtual void subscribe(SrsReloadHandler* handler);
virtual void unsubscribe(SrsReloadHandler* handler);
virtual void subscribe(ISrsReloadHandler* handler);
virtual void unsubscribe(ISrsReloadHandler* handler);
public:
virtual int parse_options(int argc, char** argv);
public:
virtual SrsConfDirective* get_vhost(const std::string &vhost);
virtual bool get_vhost_enabled(const std::string& vhost);
virtual SrsConfDirective* get_vhost_on_connect(const std::string& vhost);
virtual SrsConfDirective* get_vhost_on_close(const std::string& vhost);
virtual SrsConfDirective* get_vhost_on_publish(const std::string& vhost);
virtual SrsConfDirective* get_vhost_on_unpublish(const std::string& vhost);
virtual SrsConfDirective* get_vhost_on_play(const std::string& vhost);
virtual SrsConfDirective* get_vhost_on_stop(const std::string& vhost);
virtual SrsConfDirective* get_transcode(const std::string& vhost, const std::string& scope);
virtual bool get_transcode_enabled(SrsConfDirective* transcode);
virtual std::string get_transcode_ffmpeg(SrsConfDirective* transcode);
virtual void get_transcode_engines(SrsConfDirective* transcode, std::vector<SrsConfDirective*>& engines);
virtual bool get_engine_enabled(SrsConfDirective* engine);
virtual std::string get_engine_vcodec(SrsConfDirective* engine);
virtual int get_engine_vbitrate(SrsConfDirective* engine);
virtual double get_engine_vfps(SrsConfDirective* engine);
virtual int get_engine_vwidth(SrsConfDirective* engine);
virtual int get_engine_vheight(SrsConfDirective* engine);
virtual int get_engine_vthreads(SrsConfDirective* engine);
virtual std::string get_engine_vprofile(SrsConfDirective* engine);
virtual std::string get_engine_vpreset(SrsConfDirective* engine);
virtual void get_engine_vparams(SrsConfDirective* engine, std::vector<std::string>& vparams);
virtual void get_engine_vfilter(SrsConfDirective* engine, std::vector<std::string>& vfilter);
virtual std::string get_engine_acodec(SrsConfDirective* engine);
virtual int get_engine_abitrate(SrsConfDirective* engine);
virtual int get_engine_asample_rate(SrsConfDirective* engine);
virtual int get_engine_achannels(SrsConfDirective* engine);
virtual void get_engine_aparams(SrsConfDirective* engine, std::vector<std::string>& aparams);
virtual std::string get_engine_output(SrsConfDirective* engine);
virtual std::string get_log_dir();
virtual int get_max_connections();
virtual SrsConfDirective* get_gop_cache(const std::string& vhost);
virtual SrsConfDirective* get_forward(const std::string& vhost);
virtual SrsConfDirective* get_hls(const std::string &vhost);
virtual bool get_hls_enabled(const std::string& vhost);
virtual SrsConfDirective* get_hls_path(const std::string& vhost);
virtual SrsConfDirective* get_hls_fragment(const std::string& vhost);
virtual SrsConfDirective* get_hls_window(const std::string& vhost);
virtual SrsConfDirective* get_refer(const std::string& vhost);
virtual SrsConfDirective* get_refer_play(const std::string& vhost);
virtual SrsConfDirective* get_refer_publish(const std::string& vhost);
virtual SrsConfDirective* get_listen();
virtual SrsConfDirective* get_chunk_size(const std::string &vhost);
virtual SrsConfDirective* get_pithy_print_publish();
virtual SrsConfDirective* get_pithy_print_forwarder();
virtual SrsConfDirective* get_pithy_print_encoder();
virtual SrsConfDirective* get_pithy_print_hls();
virtual SrsConfDirective* get_pithy_print_play();
virtual bool get_bw_check_enabled(const std::string &vhost, const std::string &key);
virtual void get_bw_check_settings(const std::string &vhost, int64_t &interval_ms, int &play_kbps, int &pub_kbps);
private:
virtual int parse_file(const char* filename);
virtual int parse_argv(int& i, char** argv);
virtual void print_help(char** argv);
public:
virtual SrsConfDirective* get_vhost(std::string vhost);
virtual bool get_vhost_enabled(std::string vhost);
virtual bool get_vhost_enabled(SrsConfDirective* vhost);
virtual SrsConfDirective* get_vhost_on_connect(std::string vhost);
virtual SrsConfDirective* get_vhost_on_close(std::string vhost);
virtual SrsConfDirective* get_vhost_on_publish(std::string vhost);
virtual SrsConfDirective* get_vhost_on_unpublish(std::string vhost);
virtual SrsConfDirective* get_vhost_on_play(std::string vhost);
virtual SrsConfDirective* get_vhost_on_stop(std::string vhost);
virtual SrsConfDirective* get_transcode(std::string vhost, std::string scope);
virtual bool get_transcode_enabled(SrsConfDirective* transcode);
virtual std::string get_transcode_ffmpeg(SrsConfDirective* transcode);
virtual void get_transcode_engines(SrsConfDirective* transcode, std::vector<SrsConfDirective*>& engines);
virtual bool get_engine_enabled(SrsConfDirective* engine);
virtual std::string get_engine_vcodec(SrsConfDirective* engine);
virtual int get_engine_vbitrate(SrsConfDirective* engine);
virtual double get_engine_vfps(SrsConfDirective* engine);
virtual int get_engine_vwidth(SrsConfDirective* engine);
virtual int get_engine_vheight(SrsConfDirective* engine);
virtual int get_engine_vthreads(SrsConfDirective* engine);
virtual std::string get_engine_vprofile(SrsConfDirective* engine);
virtual std::string get_engine_vpreset(SrsConfDirective* engine);
virtual void get_engine_vparams(SrsConfDirective* engine, std::vector<std::string>& vparams);
virtual void get_engine_vfilter(SrsConfDirective* engine, std::vector<std::string>& vfilter);
virtual std::string get_engine_acodec(SrsConfDirective* engine);
virtual int get_engine_abitrate(SrsConfDirective* engine);
virtual int get_engine_asample_rate(SrsConfDirective* engine);
virtual int get_engine_achannels(SrsConfDirective* engine);
virtual void get_engine_aparams(SrsConfDirective* engine, std::vector<std::string>& aparams);
virtual std::string get_engine_output(SrsConfDirective* engine);
virtual std::string get_log_dir();
virtual int get_max_connections();
virtual bool get_gop_cache(std::string vhost);
virtual double get_queue_length(std::string vhost);
virtual SrsConfDirective* get_forward(std::string vhost);
private:
virtual SrsConfDirective* get_hls(std::string vhost);
public:
virtual bool get_hls_enabled(std::string vhost);
virtual std::string get_hls_path(std::string vhost);
virtual double get_hls_fragment(std::string vhost);
virtual double get_hls_window(std::string vhost);
virtual SrsConfDirective* get_refer(std::string vhost);
virtual SrsConfDirective* get_refer_play(std::string vhost);
virtual SrsConfDirective* get_refer_publish(std::string vhost);
virtual SrsConfDirective* get_listen();
virtual int get_chunk_size();
virtual int get_pithy_print_publish();
virtual int get_pithy_print_forwarder();
virtual int get_pithy_print_encoder();
virtual int get_pithy_print_hls();
virtual int get_pithy_print_play();
virtual bool get_bw_check_enabled(const std::string &vhost, const std::string &key);
virtual void get_bw_check_settings(const std::string &vhost, int64_t &interval_ms, int &play_kbps, int &pub_kbps);
};
/**

10
trunk/src/core/srs_core_conn.cpp
View file

@ -36,15 +36,7 @@ SrsConnection::SrsConnection(SrsServer* srs_server, st_netfd_t client_stfd)
SrsConnection::~SrsConnection()
{
if (stfd) {
int fd = st_netfd_fileno(stfd);
st_netfd_close(stfd);
stfd = NULL;
// st does not close it sometimes,
// close it manually.
close(fd);
}
srs_close_stfd(stfd);
}
int SrsConnection::start()

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

@ -30,8 +30,6 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <srs_core.hpp>
#include <st.h>
class SrsServer;
class SrsConnection
{

226
trunk/src/core/srs_core_encoder.cpp
View file

@ -483,13 +483,103 @@ void SrsFFMPEG::stop()
SrsEncoder::SrsEncoder()
{
tid = NULL;
loop = false;
pthread = new SrsThread(this, SRS_ENCODER_SLEEP_MS);
pithy_print = new SrsPithyPrint(SRS_STAGE_ENCODER);
}
SrsEncoder::~SrsEncoder()
{
on_unpublish();
srs_freep(pthread);
}
int SrsEncoder::on_publish(SrsRequest* req)
{
int ret = ERROR_SUCCESS;
ret = parse_scope_engines(req);
// ignore the loop encoder
if (ret == ERROR_ENCODER_LOOP) {
clear_engines();
ret = ERROR_SUCCESS;
}
// return for error or no engine.
if (ret != ERROR_SUCCESS || ffmpegs.empty()) {
return ret;
}
// start thread to run all encoding engines.
if ((ret = pthread->start()) != ERROR_SUCCESS) {
srs_error("st_thread_create failed. ret=%d", ret);
return ret;
}
return ret;
}
void SrsEncoder::on_unpublish()
{
pthread->stop();
clear_engines();
}
int SrsEncoder::cycle()
{
int ret = ERROR_SUCCESS;
std::vector<SrsFFMPEG*>::iterator it;
for (it = ffmpegs.begin(); it != ffmpegs.end(); ++it) {
SrsFFMPEG* ffmpeg = *it;
// start all ffmpegs.
if ((ret = ffmpeg->start()) != ERROR_SUCCESS) {
srs_error("ffmpeg start failed. ret=%d", ret);
return ret;
}
// check ffmpeg status.
if ((ret = ffmpeg->cycle()) != ERROR_SUCCESS) {
srs_error("ffmpeg cycle failed. ret=%d", ret);
return ret;
}
}
// pithy print
encoder();
pithy_print->elapse(SRS_ENCODER_SLEEP_MS);
return ret;
}
void SrsEncoder::on_leave_loop()
{
// kill ffmpeg when finished and it alive
std::vector<SrsFFMPEG*>::iterator it;
for (it = ffmpegs.begin(); it != ffmpegs.end(); ++it) {
SrsFFMPEG* ffmpeg = *it;
ffmpeg->stop();
}
}
void SrsEncoder::clear_engines()
{
std::vector<SrsFFMPEG*>::iterator it;
for (it = ffmpegs.begin(); it != ffmpegs.end(); ++it) {
SrsFFMPEG* ffmpeg = *it;
srs_freep(ffmpeg);
}
ffmpegs.clear();
}
SrsFFMPEG* SrsEncoder::at(int index)
{
return ffmpegs[index];
}
int SrsEncoder::parse_scope_engines(SrsRequest* req)
@ -531,60 +621,6 @@ int SrsEncoder::parse_scope_engines(SrsRequest* req)
return ret;
}
int SrsEncoder::on_publish(SrsRequest* req)
{
int ret = ERROR_SUCCESS;
ret = parse_scope_engines(req);
// ignore the loop encoder
if (ret == ERROR_ENCODER_LOOP) {
ret = ERROR_SUCCESS;
}
// return for error or no engine.
if (ret != ERROR_SUCCESS || ffmpegs.empty()) {
return ret;
}
// start thread to run all encoding engines.
srs_assert(!tid);
if((tid = st_thread_create(encoder_thread, this, 1, 0)) == NULL) {
ret = ERROR_ST_CREATE_FORWARD_THREAD;
srs_error("st_thread_create failed. ret=%d", ret);
return ret;
}
return ret;
}
void SrsEncoder::on_unpublish()
{
if (tid) {
loop = false;
st_thread_interrupt(tid);
st_thread_join(tid, NULL);
tid = NULL;
}
clear_engines();
}
void SrsEncoder::clear_engines()
{
std::vector<SrsFFMPEG*>::iterator it;
for (it = ffmpegs.begin(); it != ffmpegs.end(); ++it) {
SrsFFMPEG* ffmpeg = *it;
srs_freep(ffmpeg);
}
ffmpegs.clear();
}
SrsFFMPEG* SrsEncoder::at(int index)
{
return ffmpegs[index];
}
int SrsEncoder::parse_transcode(SrsRequest* req, SrsConfDirective* conf)
{
int ret = ERROR_SUCCESS;
@ -631,7 +667,6 @@ int SrsEncoder::parse_transcode(SrsRequest* req, SrsConfDirective* conf)
// if got a loop, donot transcode the whole stream.
if (ret == ERROR_ENCODER_LOOP) {
clear_engines();
break;
}
@ -646,85 +681,14 @@ int SrsEncoder::parse_transcode(SrsRequest* req, SrsConfDirective* conf)
return ret;
}
int SrsEncoder::cycle()
{
int ret = ERROR_SUCCESS;
std::vector<SrsFFMPEG*>::iterator it;
for (it = ffmpegs.begin(); it != ffmpegs.end(); ++it) {
SrsFFMPEG* ffmpeg = *it;
// start all ffmpegs.
if ((ret = ffmpeg->start()) != ERROR_SUCCESS) {
srs_error("ffmpeg start failed. ret=%d", ret);
return ret;
}
// check ffmpeg status.
if ((ret = ffmpeg->cycle()) != ERROR_SUCCESS) {
srs_error("ffmpeg cycle failed. ret=%d", ret);
return ret;
}
}
return ret;
}
void SrsEncoder::encoder_cycle()
{
int ret = ERROR_SUCCESS;
log_context->generate_id();
srs_trace("encoder cycle start");
SrsPithyPrint pithy_print(SRS_STAGE_ENCODER);
while (loop) {
if ((ret = cycle()) != ERROR_SUCCESS) {
srs_warn("encoder cycle failed, ignored and retry, ret=%d", ret);
} else {
srs_info("encoder cycle success, retry");
}
if (!loop) {
break;
}
encoder(&pithy_print);
pithy_print.elapse(SRS_ENCODER_SLEEP_MS);
st_usleep(SRS_ENCODER_SLEEP_MS * 1000);
}
// kill ffmpeg when finished and it alive
std::vector<SrsFFMPEG*>::iterator it;
for (it = ffmpegs.begin(); it != ffmpegs.end(); ++it) {
SrsFFMPEG* ffmpeg = *it;
ffmpeg->stop();
}
srs_trace("encoder cycle finished");
}
void SrsEncoder::encoder(SrsPithyPrint* pithy_print)
void SrsEncoder::encoder()
{
// reportable
if (pithy_print->can_print()) {
srs_trace("-> time=%"PRId64", encoders=%d",
pithy_print->get_age(), (int)ffmpegs.size());
// TODO: FIXME: show more info.
srs_trace("-> time=%"PRId64", encoders=%d", pithy_print->get_age(), (int)ffmpegs.size());
}
}
void* SrsEncoder::encoder_thread(void* arg)
{
SrsEncoder* obj = (SrsEncoder*)arg;
srs_assert(obj != NULL);
obj->loop = true;
obj->encoder_cycle();
return NULL;
}
#endif

19
trunk/src/core/srs_core_encoder.hpp
View file

@ -32,7 +32,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <string>
#include <vector>
#include <st.h>
#include <srs_core_thread.hpp>
class SrsConfDirective;
class SrsRequest;
@ -85,28 +85,29 @@ public:
* the encoder for a stream,
* may use multiple ffmpegs to transcode the specified stream.
*/
class SrsEncoder
class SrsEncoder : public ISrsThreadHandler
{
private:
std::vector<SrsFFMPEG*> ffmpegs;
private:
st_thread_t tid;
bool loop;
SrsThread* pthread;
SrsPithyPrint* pithy_print;
public:
SrsEncoder();
virtual ~SrsEncoder();
public:
virtual int on_publish(SrsRequest* req);
virtual void on_unpublish();
// interface ISrsThreadHandler.
public:
virtual int cycle();
virtual void on_leave_loop();
private:
virtual int parse_scope_engines(SrsRequest* req);
virtual void clear_engines();
virtual SrsFFMPEG* at(int index);
virtual int parse_scope_engines(SrsRequest* req);
virtual int parse_transcode(SrsRequest* req, SrsConfDirective* conf);
virtual int cycle();
virtual void encoder_cycle();
virtual void encoder(SrsPithyPrint* pithy_print);
static void* encoder_thread(void* arg);
virtual void encoder();
};
#endif

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

@ -37,8 +37,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#define ERROR_ST_OPEN_SOCKET 102
#define ERROR_ST_CREATE_LISTEN_THREAD 103
#define ERROR_ST_CREATE_CYCLE_THREAD 104
#define ERROR_ST_CREATE_FORWARD_THREAD 105
#define ERROR_ST_CONNECT 106
#define ERROR_ST_CONNECT 105
#define ERROR_SOCKET_CREATE 200
#define ERROR_SOCKET_SETREUSE 201
@ -85,9 +84,8 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#define ERROR_SYSTEM_CONFIG_EOF 409
#define ERROR_SYSTEM_STREAM_BUSY 410
#define ERROR_SYSTEM_IP_INVALID 411
#define ERROR_SYSTEM_CONFIG_TOO_LARGE 412
#define ERROR_SYSTEM_FORWARD_LOOP 413
#define ERROR_SYSTEM_WAITPID 414
#define ERROR_SYSTEM_FORWARD_LOOP 412
#define ERROR_SYSTEM_WAITPID 413
// see librtmp.
// failed when open ssl create the dh

277
trunk/src/core/srs_core_forward.cpp
View file

@ -35,32 +35,39 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <srs_core_pithy_print.hpp>
#include <srs_core_rtmp.hpp>
#include <srs_core_config.hpp>
#include <srs_core_source.hpp>
#include <srs_core_autofree.hpp>
#define SRS_PULSE_TIMEOUT_MS 100
#define SRS_FORWARDER_SLEEP_MS 2000
#define SRS_SEND_TIMEOUT_US 3000000L
#define SRS_RECV_TIMEOUT_US SRS_SEND_TIMEOUT_US
SrsForwarder::SrsForwarder()
SrsForwarder::SrsForwarder(SrsSource* _source)
{
source = _source;
client = NULL;
stfd = NULL;
stream_id = 0;
tid = NULL;
loop = false;
pthread = new SrsThread(this, SRS_FORWARDER_SLEEP_MS);
queue = new SrsMessageQueue();
jitter = new SrsRtmpJitter();
}
SrsForwarder::~SrsForwarder()
{
on_unpublish();
std::vector<SrsSharedPtrMessage*>::iterator it;
for (it = msgs.begin(); it != msgs.end(); ++it) {
SrsSharedPtrMessage* msg = *it;
srs_freep(msg);
}
msgs.clear();
srs_freep(pthread);
srs_freep(queue);
srs_freep(jitter);
}
void SrsForwarder::set_queue_size(double queue_size)
{
queue->set_queue_size(queue_size);
}
int SrsForwarder::on_publish(SrsRequest* req, std::string forward_server)
@ -110,41 +117,19 @@ int SrsForwarder::on_publish(SrsRequest* req, std::string forward_server)
source_ep.c_str(), dest_ep.c_str(), tc_url.c_str(),
stream_name.c_str());
// TODO: seems bug when republish and reforward.
// start forward
if ((ret = open_socket()) != ERROR_SUCCESS) {
return ret;
}
srs_assert(!tid);
if((tid = st_thread_create(forward_thread, this, 1, 0)) == NULL){
ret = ERROR_ST_CREATE_FORWARD_THREAD;
srs_error("st_thread_create failed. ret=%d", ret);
if ((ret = pthread->start()) != ERROR_SUCCESS) {
srs_error("start srs thread failed. ret=%d", ret);
return ret;
}
}
return ret;
}
void SrsForwarder::on_unpublish()
{
if (tid) {
loop = false;
st_thread_interrupt(tid);
st_thread_join(tid, NULL);
tid = NULL;
}
pthread->stop();
if (stfd) {
int fd = st_netfd_fileno(stfd);
st_netfd_close(stfd);
stfd = NULL;
// st does not close it sometimes,
// close it manually.
close(fd);
}
close_underlayer_socket();
srs_freep(client);
}
@ -153,7 +138,14 @@ int SrsForwarder::on_meta_data(SrsSharedPtrMessage* metadata)
{
int ret = ERROR_SUCCESS;
msgs.push_back(metadata);
if ((ret = jitter->correct(metadata, 0, 0)) != ERROR_SUCCESS) {
srs_freep(metadata);
return ret;
}
if ((ret = queue->enqueue(metadata)) != ERROR_SUCCESS) {
return ret;
}
return ret;
}
@ -162,7 +154,14 @@ int SrsForwarder::on_audio(SrsSharedPtrMessage* msg)
{
int ret = ERROR_SUCCESS;
msgs.push_back(msg);
if ((ret = jitter->correct(msg, 0, 0)) != ERROR_SUCCESS) {
srs_freep(msg);
return ret;
}
if ((ret = queue->enqueue(msg)) != ERROR_SUCCESS) {
return ret;
}
return ret;
}
@ -171,15 +170,74 @@ int SrsForwarder::on_video(SrsSharedPtrMessage* msg)
{
int ret = ERROR_SUCCESS;
msgs.push_back(msg);
if ((ret = jitter->correct(msg, 0, 0)) != ERROR_SUCCESS) {
srs_freep(msg);
return ret;
}
if ((ret = queue->enqueue(msg)) != ERROR_SUCCESS) {
return ret;
}
return ret;
}
int SrsForwarder::open_socket()
int SrsForwarder::cycle()
{
int ret = ERROR_SUCCESS;
if ((ret = connect_server()) != ERROR_SUCCESS) {
return ret;
}
srs_assert(client);
client->set_recv_timeout(SRS_RECV_TIMEOUT_US);
client->set_send_timeout(SRS_SEND_TIMEOUT_US);
if ((ret = client->handshake()) != ERROR_SUCCESS) {
srs_error("handshake with server failed. ret=%d", ret);
return ret;
}
if ((ret = client->connect_app(app, tc_url)) != ERROR_SUCCESS) {
srs_error("connect with server failed, tcUrl=%s. ret=%d", tc_url.c_str(), ret);
return ret;
}
if ((ret = client->create_stream(stream_id)) != ERROR_SUCCESS) {
srs_error("connect with server failed, stream_id=%d. ret=%d", stream_id, ret);
return ret;
}
if ((ret = client->publish(stream_name, stream_id)) != ERROR_SUCCESS) {
srs_error("connect with server failed, stream_name=%s, stream_id=%d. ret=%d",
stream_name.c_str(), stream_id, ret);
return ret;
}
if ((ret = source->on_forwarder_start(this)) != ERROR_SUCCESS) {
srs_error("callback the source to feed the sequence header failed. ret=%d", ret);
return ret;
}
if ((ret = forward()) != ERROR_SUCCESS) {
return ret;
}
return ret;
}
void SrsForwarder::close_underlayer_socket()
{
srs_close_stfd(stfd);
}
int SrsForwarder::connect_server()
{
int ret = ERROR_SUCCESS;
// reopen
close_underlayer_socket();
// open socket.
srs_trace("forward stream=%s, tcUrl=%s to server=%s, port=%d",
stream_name.c_str(), tc_url.c_str(), server.c_str(), port);
@ -190,6 +248,7 @@ int SrsForwarder::open_socket()
return ret;
}
srs_assert(!stfd);
stfd = st_netfd_open_socket(sock);
if(stfd == NULL){
ret = ERROR_ST_OPEN_SOCKET;
@ -200,13 +259,7 @@ int SrsForwarder::open_socket()
srs_freep(client);
client = new SrsRtmpClient(stfd);
return ret;
}
int SrsForwarder::connect_server()
{
int ret = ERROR_SUCCESS;
// connect to server.
std::string ip = srs_dns_resolve(server);
if (ip.empty()) {
ret = ERROR_SYSTEM_IP_INVALID;
@ -229,43 +282,6 @@ int SrsForwarder::connect_server()
return ret;
}
int SrsForwarder::cycle()
{
int ret = ERROR_SUCCESS;
client->set_recv_timeout(SRS_RECV_TIMEOUT_US);
client->set_send_timeout(SRS_SEND_TIMEOUT_US);
if ((ret = connect_server()) != ERROR_SUCCESS) {
return ret;
}
srs_assert(client);
if ((ret = client->handshake()) != ERROR_SUCCESS) {
srs_error("handshake with server failed. ret=%d", ret);
return ret;
}
if ((ret = client->connect_app(app, tc_url)) != ERROR_SUCCESS) {
srs_error("connect with server failed, tcUrl=%s. ret=%d", tc_url.c_str(), ret);
return ret;
}
if ((ret = client->create_stream(stream_id)) != ERROR_SUCCESS) {
srs_error("connect with server failed, stream_id=%d. ret=%d", stream_id, ret);
return ret;
}
if ((ret = client->publish(stream_name, stream_id)) != ERROR_SUCCESS) {
srs_error("connect with server failed, stream_name=%s, stream_id=%d. ret=%d",
stream_name.c_str(), stream_id, ret);
return ret;
}
if ((ret = forward()) != ERROR_SUCCESS) {
return ret;
}
return ret;
}
int SrsForwarder::forward()
{
int ret = ERROR_SUCCESS;
@ -274,9 +290,7 @@ int SrsForwarder::forward()
SrsPithyPrint pithy_print(SRS_STAGE_FORWARDER);
while (loop) {
pithy_print.elapse(SRS_PULSE_TIMEOUT_MS);
while (true) {
// switch to other st-threads.
st_usleep(0);
@ -292,91 +306,42 @@ int SrsForwarder::forward()
}
}
// forward all messages.
int count = 0;
SrsSharedPtrMessage** msgs = NULL;
if ((ret = queue->get_packets(0, msgs, count)) != ERROR_SUCCESS) {
srs_error("get message to forward failed. ret=%d", ret);
return ret;
}
// ignore when no messages.
int count = (int)msgs.size();
if (msgs.empty()) {
if (count <= 0) {
srs_verbose("no packets to forward.");
continue;
}
SrsAutoFree(SrsSharedPtrMessage*, msgs, true);
// reportable
// pithy print
pithy_print.elapse(SRS_PULSE_TIMEOUT_MS);
if (pithy_print.can_print()) {
srs_trace("-> time=%"PRId64", msgs=%d, obytes=%"PRId64", ibytes=%"PRId64", okbps=%d, ikbps=%d",
pithy_print.get_age(), count, client->get_send_bytes(), client->get_recv_bytes(), client->get_send_kbps(), client->get_recv_kbps());
}
// all msgs to forward.
int i = 0;
for (i = 0; i < count; i++) {
for (int i = 0; i < count; i++) {
SrsSharedPtrMessage* msg = msgs[i];
msgs[i] = NULL;
// we erased the sendout messages, the msg must not be NULL.
srs_assert(msg);
ret = client->send_message(msg);
if (ret != ERROR_SUCCESS) {
srs_assert(msg);
msgs[i] = NULL;
if ((ret = client->send_message(msg)) != ERROR_SUCCESS) {
srs_error("forwarder send message to server failed. ret=%d", ret);
// convert the index to count when error.
i++;
break;
return ret;
}
}
// clear sendout mesages.
if (i < count) {
srs_warn("clear forwarded msg, total=%d, forwarded=%d, ret=%d", count, i, ret);
} else {
srs_info("clear forwarded msg, total=%d, forwarded=%d, ret=%d", count, i, ret);
}
msgs.erase(msgs.begin(), msgs.begin() + i);
if (ret != ERROR_SUCCESS) {
break;
}
}
return ret;
}
void SrsForwarder::forward_cycle()
{
int ret = ERROR_SUCCESS;
log_context->generate_id();
srs_trace("forward cycle start");
while (loop) {
if ((ret = cycle()) != ERROR_SUCCESS) {
srs_warn("forward cycle failed, ignored and retry, ret=%d", ret);
} else {
srs_info("forward cycle success, retry");
}
if (!loop) {
break;
}
st_usleep(SRS_FORWARDER_SLEEP_MS * 1000);
if ((ret = open_socket()) != ERROR_SUCCESS) {
srs_warn("forward cycle reopen failed, ignored and retry, ret=%d", ret);
} else {
srs_info("forward cycle reopen success");
}
}
srs_trace("forward cycle finished");
}
void* SrsForwarder::forward_thread(void* arg)
{
SrsForwarder* obj = (SrsForwarder*)arg;
srs_assert(obj != NULL);
obj->loop = true;
obj->forward_cycle();
return NULL;
}

30
trunk/src/core/srs_core_forward.hpp
View file

@ -30,19 +30,21 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <srs_core.hpp>
#include <string>
#include <vector>
#include <st.h>
#include <srs_core_thread.hpp>
class SrsSharedPtrMessage;
class SrsOnMetaDataPacket;
class SrsMessageQueue;
class SrsRtmpJitter;
class SrsRtmpClient;
class SrsRequest;
class SrsSource;
/**
* forward the stream to other servers.
*/
class SrsForwarder
class SrsForwarder : public ISrsThreadHandler
{
private:
std::string app;
@ -53,28 +55,30 @@ private:
int port;
private:
st_netfd_t stfd;
st_thread_t tid;
bool loop;
SrsThread* pthread;
private:
SrsSource* source;
SrsRtmpClient* client;
std::vector<SrsSharedPtrMessage*> msgs;
SrsRtmpJitter* jitter;
SrsMessageQueue* queue;
public:
SrsForwarder();
SrsForwarder(SrsSource* _source);
virtual ~SrsForwarder();
public:
virtual void set_queue_size(double queue_size);
public:
virtual int on_publish(SrsRequest* req, std::string forward_server);
virtual void on_unpublish();
virtual int on_meta_data(SrsSharedPtrMessage* metadata);
virtual int on_audio(SrsSharedPtrMessage* msg);
virtual int on_video(SrsSharedPtrMessage* msg);
private:
virtual int open_socket();
virtual int connect_server();
private:
// interface ISrsThreadHandler.
public:
virtual int cycle();
private:
virtual void close_underlayer_socket();
virtual int connect_server();
virtual int forward();
virtual void forward_cycle();
static void* forward_thread(void* arg);
};
#endif

57
trunk/src/core/srs_core_hls.cpp
View file

@ -1109,10 +1109,11 @@ int SrsTSCache::cache_video(SrsCodec* codec, SrsCodecSample* sample)
return ret;
}
SrsHls::SrsHls()
SrsHls::SrsHls(SrsSource* _source)
{
hls_enabled = false;
source = _source;
codec = new SrsCodec();
sample = new SrsCodecSample();
jitter = new SrsRtmpJitter();
@ -1148,7 +1149,6 @@ int SrsHls::on_publish(SrsRequest* req)
std::string stream = req->stream;
std::string app = req->app;
// TODO: support reload.
if (!config->get_hls_enabled(vhost)) {
return ret;
}
@ -1156,30 +1156,11 @@ int SrsHls::on_publish(SrsRequest* req)
// if enabled, open the muxer.
hls_enabled = true;
// TODO: subscribe the reload event.
int hls_fragment = 0;
int hls_window = 0;
SrsConfDirective* conf = NULL;
if ((conf = config->get_hls_fragment(vhost)) != NULL && !conf->arg0().empty()) {
hls_fragment = ::atoi(conf->arg0().c_str());
}
if (hls_fragment <= 0) {
hls_fragment = SRS_CONF_DEFAULT_HLS_FRAGMENT;
}
if ((conf = config->get_hls_window(vhost)) != NULL && !conf->arg0().empty()) {
hls_window = ::atoi(conf->arg0().c_str());
}
if (hls_window <= 0) {
hls_window = SRS_CONF_DEFAULT_HLS_WINDOW;
}
int hls_fragment = config->get_hls_fragment(vhost);
int hls_window = config->get_hls_window(vhost);
// get the hls path config
std::string hls_path = SRS_CONF_DEFAULT_HLS_PATH;
if ((conf = config->get_hls_path(vhost)) != NULL) {
hls_path = conf->arg0();
}
std::string hls_path = config->get_hls_path(vhost);
// open muxer
if ((ret = muxer->update_config(app, stream, hls_path, hls_fragment, hls_window)) != ERROR_SUCCESS) {
@ -1191,6 +1172,12 @@ int SrsHls::on_publish(SrsRequest* req)
srs_error("m3u8 muxer open segment failed. ret=%d", ret);
return ret;
}
// notice the source to get the cached sequence header.
if ((ret = source->on_hls_start()) != ERROR_SUCCESS) {
srs_error("callback source hls start failed. ret=%d", ret);
return ret;
}
return ret;
}
@ -1215,16 +1202,16 @@ void SrsHls::on_unpublish()
hls_enabled = false;
}
int SrsHls::on_meta_data(SrsOnMetaDataPacket* metadata)
int SrsHls::on_meta_data(SrsAmf0Object* metadata)
{
int ret = ERROR_SUCCESS;
if (!metadata || !metadata->metadata) {
if (!metadata) {
srs_trace("no metadata persent, hls ignored it.");
return ret;
}
SrsAmf0Object* obj = metadata->metadata;
SrsAmf0Object* obj = metadata;
if (obj->size() <= 0) {
srs_trace("no metadata persent, hls ignored it.");
return ret;
@ -1273,7 +1260,6 @@ int SrsHls::on_audio(SrsSharedPtrMessage* audio)
SrsAutoFree(SrsSharedPtrMessage, audio, false);
// TODO: maybe donot need to demux the aac?
if (!hls_enabled) {
return ret;
}
@ -1293,14 +1279,13 @@ int SrsHls::on_audio(SrsSharedPtrMessage* audio)
return ret;
}
int64_t corrected_time = 0;
if ((ret = jitter->correct(audio, 0, 0, &corrected_time)) != ERROR_SUCCESS) {
if ((ret = jitter->correct(audio, 0, 0)) != ERROR_SUCCESS) {
srs_error("rtmp jitter correct audio failed. ret=%d", ret);
return ret;
}
// the pts calc from rtmp/flv header.
int64_t pts = corrected_time * 90;
int64_t pts = audio->header.timestamp * 90;
if ((ret = ts_cache->write_audio(codec, muxer, pts, sample)) != ERROR_SUCCESS) {
srs_error("ts cache write audio failed. ret=%d", ret);
@ -1316,7 +1301,6 @@ int SrsHls::on_video(SrsSharedPtrMessage* video)
SrsAutoFree(SrsSharedPtrMessage, video, false);
// TODO: maybe donot need to demux the avc?
if (!hls_enabled) {
return ret;
}
@ -1337,24 +1321,23 @@ int SrsHls::on_video(SrsSharedPtrMessage* video)
return ret;
}
int64_t corrected_time = 0;
if ((ret = jitter->correct(video, 0, 0, &corrected_time)) != ERROR_SUCCESS) {
if ((ret = jitter->correct(video, 0, 0)) != ERROR_SUCCESS) {
srs_error("rtmp jitter correct video failed. ret=%d", ret);
return ret;
}
int64_t dts = corrected_time * 90;
int64_t dts = video->header.timestamp * 90;
if ((ret = ts_cache->write_video(codec, muxer, dts, sample)) != ERROR_SUCCESS) {
srs_error("ts cache write video failed. ret=%d", ret);
return ret;
}
_mpegts();
hls_mux();
return ret;
}
void SrsHls::_mpegts()
void SrsHls::hls_mux()
{
// reportable
if (pithy_print->can_print()) {

27
trunk/src/core/srs_core_hls.hpp
View file

@ -34,16 +34,17 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <string>
#include <vector>
class SrsOnMetaDataPacket;
class SrsSharedPtrMessage;
class SrsCodecSample;
class SrsCodecBuffer;
class SrsMpegtsFrame;
class SrsAmf0Object;
class SrsRtmpJitter;
class SrsTSMuxer;
class SrsCodec;
class SrsRequest;
class SrsPithyPrint;
class SrsSource;
/**
* jitter correct for audio,
@ -207,21 +208,39 @@ private:
SrsTSCache* ts_cache;
private:
bool hls_enabled;
SrsSource* source;
SrsCodec* codec;
SrsCodecSample* sample;
SrsRtmpJitter* jitter;
SrsPithyPrint* pithy_print;
public:
SrsHls();
SrsHls(SrsSource* _source);
virtual ~SrsHls();
public:
/**
* publish stream event, continue to write the m3u8,
* for the muxer object not destroyed.
*/
virtual int on_publish(SrsRequest* req);
/**
* the unpublish event, only close the muxer, donot destroy the
* muxer, for when we continue to publish, the m3u8 will continue.
*/
virtual void on_unpublish();
virtual int on_meta_data(SrsOnMetaDataPacket* metadata);
/**
* get some information from metadata, it's optinal.
*/
virtual int on_meta_data(SrsAmf0Object* metadata);
/**
* mux the audio packets to ts.
*/
virtual int on_audio(SrsSharedPtrMessage* audio);
/**
* mux the video packets to ts.
*/
virtual int on_video(SrsSharedPtrMessage* video);
private:
virtual void _mpegts();
virtual void hls_mux();
};
#endif

10
trunk/src/core/srs_core_http.cpp
View file

@ -184,15 +184,7 @@ void SrsHttpClient::disconnect()
{
connected = false;
if (stfd) {
int fd = st_netfd_fileno(stfd);
st_netfd_close(stfd);
stfd = NULL;
// st does not close it sometimes,
// close it manually.
::close(fd);
}
srs_close_stfd(stfd);
}
int SrsHttpClient::connect(SrsHttpUri* uri)

1
trunk/src/core/srs_core_http.hpp
View file

@ -36,7 +36,6 @@ class SrsSocket;
#include <string>
#include <st.h>
#include <http_parser.h>
#define SRS_HTTP_HEADER_BUFFER 1024

2
trunk/src/core/srs_core_log.cpp
View file

@ -29,8 +29,6 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <string>
#include <map>
#include <st.h>
ILogContext::ILogContext()
{
}

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

@ -32,13 +32,8 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <srs_core_error.hpp>
#define SRS_STAGE_DEFAULT_INTERVAL_MS 1200
#define SRS_STAGE_PLAY_USER_INTERVAL_MS 1300
#define SRS_STAGE_PUBLISH_USER_INTERVAL_MS 1100
#define SRS_STAGE_FORWARDER_INTERVAL_MS 2000
#define SRS_STAGE_ENCODER_INTERVAL_MS 2000
#define SRS_STAGE_HLS_INTERVAL_MS 2000
struct SrsStageInfo : public SrsReloadHandler
struct SrsStageInfo : public ISrsReloadHandler
{
int stage_id;
int pithy_print_time_ms;
@ -61,43 +56,23 @@ struct SrsStageInfo : public SrsReloadHandler
{
switch (stage_id) {
case SRS_STAGE_PLAY_USER: {
pithy_print_time_ms = SRS_STAGE_PLAY_USER_INTERVAL_MS;
SrsConfDirective* conf = config->get_pithy_print_play();
if (conf && !conf->arg0().empty()) {
pithy_print_time_ms = ::atoi(conf->arg0().c_str());
}
pithy_print_time_ms = config->get_pithy_print_play();
break;
}
case SRS_STAGE_PUBLISH_USER: {
pithy_print_time_ms = SRS_STAGE_PUBLISH_USER_INTERVAL_MS;
SrsConfDirective* conf = config->get_pithy_print_publish();
if (conf && !conf->arg0().empty()) {
pithy_print_time_ms = ::atoi(conf->arg0().c_str());
}
pithy_print_time_ms = config->get_pithy_print_publish();
break;
}
case SRS_STAGE_FORWARDER: {
pithy_print_time_ms = SRS_STAGE_FORWARDER_INTERVAL_MS;
SrsConfDirective* conf = config->get_pithy_print_forwarder();
if (conf && !conf->arg0().empty()) {
pithy_print_time_ms = ::atoi(conf->arg0().c_str());
}
pithy_print_time_ms = config->get_pithy_print_forwarder();
break;
}
case SRS_STAGE_ENCODER: {
pithy_print_time_ms = SRS_STAGE_ENCODER_INTERVAL_MS;
SrsConfDirective* conf = config->get_pithy_print_encoder();
if (conf && !conf->arg0().empty()) {
pithy_print_time_ms = ::atoi(conf->arg0().c_str());
}
pithy_print_time_ms = config->get_pithy_print_encoder();
break;
}
case SRS_STAGE_HLS: {
pithy_print_time_ms = SRS_STAGE_HLS_INTERVAL_MS;
SrsConfDirective* conf = config->get_pithy_print_hls();
if (conf && !conf->arg0().empty()) {
pithy_print_time_ms = ::atoi(conf->arg0().c_str());
}
pithy_print_time_ms = config->get_pithy_print_hls();
break;
}
default: {

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

@ -307,6 +307,11 @@ void SrsProtocol::set_send_timeout(int64_t timeout_us)
return skt->set_send_timeout(timeout_us);
}
int64_t SrsProtocol::get_send_timeout()
{
return skt->get_send_timeout();
}
int64_t SrsProtocol::get_recv_bytes()
{
return skt->get_recv_bytes();
@ -349,7 +354,7 @@ int SrsProtocol::recv_message(SrsCommonMessage** pmsg)
}
if (msg->size <= 0 || msg->header.payload_length <= 0) {
srs_trace("ignore empty message(type=%d, size=%d, time=%d, sid=%d).",
srs_trace("ignore empty message(type=%d, size=%d, time=%"PRId64", sid=%d).",
msg->header.message_type, msg->header.payload_length,
msg->header.timestamp, msg->header.stream_id);
srs_freep(msg);
@ -400,12 +405,13 @@ int SrsProtocol::send_message(ISrsMessage* msg)
// chunk message header, 11 bytes
// timestamp, 3bytes, big-endian
if (msg->header.timestamp >= RTMP_EXTENDED_TIMESTAMP) {
u_int32_t timestamp = (u_int32_t)msg->header.timestamp;
if (timestamp >= RTMP_EXTENDED_TIMESTAMP) {
*pheader++ = 0xFF;
*pheader++ = 0xFF;
*pheader++ = 0xFF;
} else {
pp = (char*)&msg->header.timestamp;
pp = (char*)&timestamp;
*pheader++ = pp[2];
*pheader++ = pp[1];
*pheader++ = pp[0];
@ -428,8 +434,8 @@ int SrsProtocol::send_message(ISrsMessage* msg)
*pheader++ = pp[3];
// chunk extended timestamp header, 0 or 4 bytes, big-endian
if(msg->header.timestamp >= RTMP_EXTENDED_TIMESTAMP){
pp = (char*)&msg->header.timestamp;
if(timestamp >= RTMP_EXTENDED_TIMESTAMP){
pp = (char*)&timestamp;
*pheader++ = pp[3];
*pheader++ = pp[2];
*pheader++ = pp[1];
@ -456,8 +462,9 @@ int SrsProtocol::send_message(ISrsMessage* msg)
// must send the extended-timestamp to flash-player.
// @see: ngx_rtmp_prepare_message
// @see: http://blog.csdn.net/win_lin/article/details/13363699
if(msg->header.timestamp >= RTMP_EXTENDED_TIMESTAMP){
pp = (char*)&msg->header.timestamp;
u_int32_t timestamp = (u_int32_t)msg->header.timestamp;
if(timestamp >= RTMP_EXTENDED_TIMESTAMP){
pp = (char*)&timestamp;
*pheader++ = pp[3];
*pheader++ = pp[2];
*pheader++ = pp[1];
@ -697,7 +704,7 @@ int SrsProtocol::recv_interlaced_message(SrsCommonMessage** pmsg)
srs_verbose("cache new chunk stream: fmt=%d, cid=%d", fmt, cid);
} else {
chunk = chunk_streams[cid];
srs_verbose("cached chunk stream: fmt=%d, cid=%d, size=%d, message(type=%d, size=%d, time=%d, sid=%d)",
srs_verbose("cached chunk stream: fmt=%d, cid=%d, size=%d, message(type=%d, size=%d, time=%"PRId64", sid=%d)",
chunk->fmt, chunk->cid, (chunk->msg? chunk->msg->size : 0), chunk->header.message_type, chunk->header.payload_length,
chunk->header.timestamp, chunk->header.stream_id);
}
@ -711,7 +718,7 @@ int SrsProtocol::recv_interlaced_message(SrsCommonMessage** pmsg)
return ret;
}
srs_verbose("read message header success. "
"fmt=%d, mh_size=%d, ext_time=%d, size=%d, message(type=%d, size=%d, time=%d, sid=%d)",
"fmt=%d, mh_size=%d, ext_time=%d, size=%d, message(type=%d, size=%d, time=%"PRId64", sid=%d)",
fmt, mh_size, chunk->extended_timestamp, (chunk->msg? chunk->msg->size : 0), chunk->header.message_type,
chunk->header.payload_length, chunk->header.timestamp, chunk->header.stream_id);
@ -733,14 +740,14 @@ int SrsProtocol::recv_interlaced_message(SrsCommonMessage** pmsg)
// not got an entire RTMP message, try next chunk.
if (!msg) {
srs_verbose("get partial message success. chunk_payload_size=%d, size=%d, message(type=%d, size=%d, time=%d, sid=%d)",
srs_verbose("get partial message success. chunk_payload_size=%d, size=%d, message(type=%d, size=%d, time=%"PRId64", sid=%d)",
payload_size, (msg? msg->size : (chunk->msg? chunk->msg->size : 0)), chunk->header.message_type, chunk->header.payload_length,
chunk->header.timestamp, chunk->header.stream_id);
return ret;
}
*pmsg = msg;
srs_info("get entire message success. chunk_payload_size=%d, size=%d, message(type=%d, size=%d, time=%d, sid=%d)",
srs_info("get entire message success. chunk_payload_size=%d, size=%d, message(type=%d, size=%d, time=%"PRId64", sid=%d)",
payload_size, (msg? msg->size : (chunk->msg? chunk->msg->size : 0)), chunk->header.message_type, chunk->header.payload_length,
chunk->header.timestamp, chunk->header.stream_id);
@ -947,16 +954,16 @@ int SrsProtocol::read_message_header(SrsChunkStream* chunk, char fmt, int bh_siz
pp[1] = *p++;
pp[2] = *p++;
pp[3] = *p++;
srs_verbose("header read completed. fmt=%d, mh_size=%d, ext_time=%d, time=%d, payload=%d, type=%d, sid=%d",
srs_verbose("header read completed. fmt=%d, mh_size=%d, ext_time=%d, time=%"PRId64", payload=%d, type=%d, sid=%d",
fmt, mh_size, chunk->extended_timestamp, chunk->header.timestamp, chunk->header.payload_length,
chunk->header.message_type, chunk->header.stream_id);
} else {
srs_verbose("header read completed. fmt=%d, mh_size=%d, ext_time=%d, time=%d, payload=%d, type=%d",
srs_verbose("header read completed. fmt=%d, mh_size=%d, ext_time=%d, time=%"PRId64", payload=%d, type=%d",
fmt, mh_size, chunk->extended_timestamp, chunk->header.timestamp, chunk->header.payload_length,
chunk->header.message_type);
}
} else {
srs_verbose("header read completed. fmt=%d, mh_size=%d, ext_time=%d, time=%d",
srs_verbose("header read completed. fmt=%d, mh_size=%d, ext_time=%d, time=%"PRId64"",
fmt, mh_size, chunk->extended_timestamp, chunk->header.timestamp);
}
} else {
@ -981,7 +988,7 @@ int SrsProtocol::read_message_header(SrsChunkStream* chunk, char fmt, int bh_siz
// ffmpeg/librtmp may donot send this filed, need to detect the value.
// @see also: http://blog.csdn.net/win_lin/article/details/13363699
int32_t timestamp = 0x00;
u_int32_t timestamp = 0x00;
char* pp = (char*)&timestamp;
pp[3] = *p++;
pp[2] = *p++;
@ -990,14 +997,14 @@ int SrsProtocol::read_message_header(SrsChunkStream* chunk, char fmt, int bh_siz
// compare to the chunk timestamp, which is set by chunk message header
// type 0,1 or 2.
int32_t chunk_timestamp = chunk->header.timestamp;
u_int32_t chunk_timestamp = chunk->header.timestamp;
if (chunk_timestamp > RTMP_EXTENDED_TIMESTAMP && chunk_timestamp != timestamp) {
mh_size -= 4;
srs_verbose("ignore the 4bytes extended timestamp. mh_size=%d", mh_size);
} else {
chunk->header.timestamp = timestamp;
}
srs_verbose("header read ext_time completed. time=%d", chunk->header.timestamp);
srs_verbose("header read ext_time completed. time=%"PRId64"", chunk->header.timestamp);
}
// valid message
@ -1027,7 +1034,7 @@ int SrsProtocol::read_message_payload(SrsChunkStream* chunk, int bh_size, int mh
buffer->erase(bh_size + mh_size);
srs_trace("get an empty RTMP "
"message(type=%d, size=%d, time=%d, sid=%d)", chunk->header.message_type,
"message(type=%d, size=%d, time=%"PRId64", sid=%d)", chunk->header.message_type,
chunk->header.payload_length, chunk->header.timestamp, chunk->header.stream_id);
*pmsg = chunk->msg;
@ -1068,13 +1075,13 @@ int SrsProtocol::read_message_payload(SrsChunkStream* chunk, int bh_size, int mh
if (chunk->header.payload_length == chunk->msg->size) {
*pmsg = chunk->msg;
chunk->msg = NULL;
srs_verbose("get entire RTMP message(type=%d, size=%d, time=%d, sid=%d)",
srs_verbose("get entire RTMP message(type=%d, size=%d, time=%"PRId64", sid=%d)",
chunk->header.message_type, chunk->header.payload_length,
chunk->header.timestamp, chunk->header.stream_id);
return ret;
}
srs_verbose("get partial RTMP message(type=%d, size=%d, time=%d, sid=%d), partial size=%d",
srs_verbose("get partial RTMP message(type=%d, size=%d, time=%"PRId64", sid=%d), partial size=%d",
chunk->header.message_type, chunk->header.payload_length,
chunk->header.timestamp, chunk->header.stream_id,
chunk->msg->size);

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

@ -33,8 +33,6 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <map>
#include <string>
#include <st.h>
#include <srs_core_log.hpp>
#include <srs_core_error.hpp>
@ -117,6 +115,7 @@ public:
virtual void set_recv_timeout(int64_t timeout_us);
virtual int64_t get_recv_timeout();
virtual void set_send_timeout(int64_t timeout_us);
virtual int64_t get_send_timeout();
virtual int64_t get_recv_bytes();
virtual int64_t get_send_bytes();
virtual int get_recv_kbps();
@ -205,8 +204,9 @@ struct SrsMessageHeader
* Four-byte field that contains a timestamp of the message.
* The 4 bytes are packed in the big-endian order.
* @remark, used as calc timestamp when decode and encode time.
* @remark, we use 64bits for large time for jitter detect and hls.
*/
u_int32_t timestamp;
int64_t timestamp;
SrsMessageHeader();
virtual ~SrsMessageHeader();
@ -1156,7 +1156,7 @@ int srs_rtmp_expect_message(SrsProtocol* protocol, SrsCommonMessage** pmsg, T**
T* pkt = dynamic_cast<T*>(msg->get_packet());
if (!pkt) {
delete msg;
srs_trace("drop message(type=%d, size=%d, time=%d, sid=%d).",
srs_trace("drop message(type=%d, size=%d, time=%"PRId64", sid=%d).",
msg->header.message_type, msg->header.payload_length,
msg->header.timestamp, msg->header.stream_id);
continue;

40
trunk/src/core/srs_core_reload.cpp
View file

@ -23,22 +23,54 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <srs_core_reload.hpp>
using namespace std;
#include <srs_core_error.hpp>
SrsReloadHandler::SrsReloadHandler()
ISrsReloadHandler::ISrsReloadHandler()
{
}
SrsReloadHandler::~SrsReloadHandler()
ISrsReloadHandler::~ISrsReloadHandler()
{
}
int SrsReloadHandler::on_reload_listen()
int ISrsReloadHandler::on_reload_listen()
{
return ERROR_SUCCESS;
}
int SrsReloadHandler::on_reload_pithy_print()
int ISrsReloadHandler::on_reload_pithy_print()
{
return ERROR_SUCCESS;
}
int ISrsReloadHandler::on_reload_vhost_removed(string /*vhost*/)
{
return ERROR_SUCCESS;
}
int ISrsReloadHandler::on_reload_gop_cache(string /*vhost*/)
{
return ERROR_SUCCESS;
}
int ISrsReloadHandler::on_reload_queue_length(string /*vhost*/)
{
return ERROR_SUCCESS;
}
int ISrsReloadHandler::on_reload_forward(string /*vhost*/)
{
return ERROR_SUCCESS;
}
int ISrsReloadHandler::on_reload_hls(string /*vhost*/)
{
return ERROR_SUCCESS;
}
int ISrsReloadHandler::on_reload_transcode(string /*vhost*/)
{
return ERROR_SUCCESS;
}

14
trunk/src/core/srs_core_reload.hpp
View file

@ -29,17 +29,25 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include <srs_core.hpp>
#include <string>
/**
* the handler for config reload.
*/
class SrsReloadHandler
class ISrsReloadHandler
{
public:
SrsReloadHandler();
virtual ~SrsReloadHandler();
ISrsReloadHandler();
virtual ~ISrsReloadHandler();
public:
virtual int on_reload_listen();
virtual int on_reload_pithy_print();
virtual int on_reload_vhost_removed(std::string vhost);
virtual int on_reload_gop_cache(std::string vhost);
virtual int on_reload_queue_length(std::string vhost);
virtual int on_reload_forward(std::string vhost);
virtual int on_reload_hls(std::string vhost);
virtual int on_reload_transcode(std::string vhost);
};
#endif

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

@ -33,6 +33,8 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <srs_core_handshake.hpp>
#include <srs_core_config.hpp>
using namespace std;
/**
* the signature for packets to client.
*/
@ -79,6 +81,23 @@ SrsRequest::~SrsRequest()
{
}
SrsRequest* SrsRequest::copy()
{
SrsRequest* cp = new SrsRequest();
cp->app = app;
cp->objectEncoding = objectEncoding;
cp->pageUrl = pageUrl;
cp->port = port;
cp->schema = schema;
cp->stream = stream;
cp->swfUrl = swfUrl;
cp->tcUrl = tcUrl;
cp->vhost = vhost;
return cp;
}
int SrsRequest::discovery_app()
{
int ret = ERROR_SUCCESS;
@ -125,6 +144,8 @@ int SrsRequest::discovery_app()
if (parsed_vhost) {
vhost = parsed_vhost->arg0();
}
// TODO: discovery the params of vhost.
srs_info("discovery app success. schema=%s, vhost=%s, port=%s, app=%s",
schema.c_str(), vhost.c_str(), port.c_str(), app.c_str());
@ -142,7 +163,7 @@ int SrsRequest::discovery_app()
return ret;
}
std::string SrsRequest::get_stream_url()
string SrsRequest::get_stream_url()
{
std::string url = "";
@ -162,7 +183,7 @@ void SrsRequest::strip()
trim(stream, "/ \n\r\t");
}
std::string& SrsRequest::trim(std::string& str, std::string chs)
std::string& SrsRequest::trim(string& str, string chs)
{
for (int i = 0; i < (int)chs.length(); i++) {
char ch = chs.at(i);
@ -245,6 +266,9 @@ int SrsRtmpClient::handshake()
SrsSocket skt(stfd);
skt.set_recv_timeout(protocol->get_recv_timeout());
skt.set_send_timeout(protocol->get_send_timeout());
SrsComplexHandshake complex_hs;
SrsSimpleHandshake simple_hs;
if ((ret = simple_hs.handshake_with_server(skt, complex_hs)) != ERROR_SUCCESS) {
@ -436,6 +460,11 @@ void SrsRtmp::set_send_timeout(int64_t timeout_us)
protocol->set_send_timeout(timeout_us);
}
int64_t SrsRtmp::get_send_timeout()
{
return protocol->get_send_timeout();
}
int64_t SrsRtmp::get_recv_bytes()
{
return protocol->get_recv_bytes();
@ -472,6 +501,9 @@ int SrsRtmp::handshake()
SrsSocket skt(stfd);
skt.set_recv_timeout(protocol->get_recv_timeout());
skt.set_send_timeout(protocol->get_send_timeout());
SrsComplexHandshake complex_hs;
SrsSimpleHandshake simple_hs;
if ((ret = simple_hs.handshake_with_client(skt, complex_hs)) != ERROR_SUCCESS) {
@ -1127,7 +1159,7 @@ int SrsRtmp::start_bandwidth_check(int max_play_kbps, int max_pub_kbps)
return ret;
}
int SrsRtmp::identify_create_stream_client(SrsCreateStreamPacket* req, int stream_id, SrsClientType& type, std::string& stream_name)
int SrsRtmp::identify_create_stream_client(SrsCreateStreamPacket* req, int stream_id, SrsClientType& type, string& stream_name)
{
int ret = ERROR_SUCCESS;
@ -1184,7 +1216,7 @@ int SrsRtmp::identify_create_stream_client(SrsCreateStreamPacket* req, int strea
return ret;
}
int SrsRtmp::identify_fmle_publish_client(SrsFMLEStartPacket* req, SrsClientType& type, std::string& stream_name)
int SrsRtmp::identify_fmle_publish_client(SrsFMLEStartPacket* req, SrsClientType& type, string& stream_name)
{
int ret = ERROR_SUCCESS;
@ -1208,7 +1240,7 @@ int SrsRtmp::identify_fmle_publish_client(SrsFMLEStartPacket* req, SrsClientType
return ret;
}
int SrsRtmp::identify_flash_publish_client(SrsPublishPacket* req, SrsClientType& type, std::string& stream_name)
int SrsRtmp::identify_flash_publish_client(SrsPublishPacket* req, SrsClientType& type, string& stream_name)
{
int ret = ERROR_SUCCESS;

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

@ -33,8 +33,6 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <string>
#include <st.h>
class SrsProtocol;
class ISrsMessage;
class SrsCommonMessage;
@ -69,6 +67,13 @@ struct SrsRequest
SrsRequest();
virtual ~SrsRequest();
/**
* 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();
/**
* disconvery vhost/app from tcUrl.
@ -148,6 +153,7 @@ public:
virtual void set_recv_timeout(int64_t timeout_us);
virtual int64_t get_recv_timeout();
virtual void set_send_timeout(int64_t timeout_us);
virtual int64_t get_send_timeout();
virtual int64_t get_recv_bytes();
virtual int64_t get_send_bytes();
virtual int get_recv_kbps();

73
trunk/src/core/srs_core_server.cpp
View file

@ -30,8 +30,6 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <algorithm>
#include <st.h>
#include <srs_core_log.hpp>
#include <srs_core_error.hpp>
#include <srs_core_client.hpp>
@ -48,24 +46,16 @@ SrsListener::SrsListener(SrsServer* _server, SrsListenerType _type)
port = 0;
server = _server;
type = _type;
tid = NULL;
loop = false;
pthread = new SrsThread(this, 0);
}
SrsListener::~SrsListener()
{
if (stfd) {
st_netfd_close(stfd);
stfd = NULL;
}
srs_close_stfd(stfd);
if (tid) {
loop = false;
st_thread_interrupt(tid);
st_thread_join(tid, NULL);
tid = NULL;
}
pthread->stop();
srs_freep(pthread);
// st does not close it sometimes,
// close it manually.
@ -118,8 +108,7 @@ int SrsListener::listen(int _port)
}
srs_verbose("st open socket success. fd=%d", fd);
if ((tid = st_thread_create(listen_thread, this, 1, 0)) == NULL) {
ret = ERROR_ST_CREATE_LISTEN_THREAD;
if ((ret = pthread->start()) != ERROR_SUCCESS) {
srs_error("st_thread_create listen thread error. ret=%d", ret);
return ret;
}
@ -130,41 +119,32 @@ int SrsListener::listen(int _port)
return ret;
}
void SrsListener::listen_cycle()
void SrsListener::on_enter_loop()
{
srs_trace("listen cycle start, port=%d, type=%d, fd=%d", port, type, fd);
}
int SrsListener::cycle()
{
int ret = ERROR_SUCCESS;
log_context->generate_id();
srs_trace("listen cycle start, port=%d, type=%d, fd=%d", port, type, fd);
st_netfd_t client_stfd = st_accept(stfd, NULL, NULL, ST_UTIME_NO_TIMEOUT);
if(client_stfd == NULL){
// ignore error.
srs_warn("ignore accept thread stoppped for accept client error");
return ret;
}
srs_verbose("get a client. fd=%d", st_netfd_fileno(client_stfd));
while (loop) {
st_netfd_t client_stfd = st_accept(stfd, NULL, NULL, ST_UTIME_NO_TIMEOUT);
if(client_stfd == NULL){
// ignore error.
srs_warn("ignore accept thread stoppped for accept client error");
continue;
}
srs_verbose("get a client. fd=%d", st_netfd_fileno(client_stfd));
if ((ret = server->accept_client(type, client_stfd)) != ERROR_SUCCESS) {
srs_warn("accept client error. ret=%d", ret);
continue;
}
srs_verbose("accept client finished. conns=%d, ret=%d", (int)conns.size(), ret);
if ((ret = server->accept_client(type, client_stfd)) != ERROR_SUCCESS) {
srs_warn("accept client error. ret=%d", ret);
return ret;
}
}
void* SrsListener::listen_thread(void* arg)
{
SrsListener* obj = (SrsListener*)arg;
srs_assert(obj != NULL);
obj->loop = true;
obj->listen_cycle();
srs_verbose("accept client finished. conns=%d, ret=%d", (int)conns.size(), ret);
return NULL;
return ret;
}
SrsServer::SrsServer()
@ -312,8 +292,7 @@ int SrsServer::accept_client(SrsListenerType type, st_netfd_t client_stfd)
srs_error("exceed the max connections, drop client: "
"clients=%d, max=%d, fd=%d", (int)conns.size(), max_connections, fd);
st_netfd_close(client_stfd);
::close(fd);
srs_close_stfd(client_stfd);
return ret;
}

17
trunk/src/core/srs_core_server.hpp
View file

@ -32,9 +32,8 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <vector>
#include <st.h>
#include <srs_core_reload.hpp>
#include <srs_core_thread.hpp>
class SrsServer;
class SrsConnection;
@ -45,7 +44,7 @@ enum SrsListenerType
SrsListenerApi
};
class SrsListener
class SrsListener : public ISrsThreadHandler
{
public:
SrsListenerType type;
@ -54,19 +53,19 @@ private:
st_netfd_t stfd;
int port;
SrsServer* server;
st_thread_t tid;
bool loop;
SrsThread* pthread;
public:
SrsListener(SrsServer* _server, SrsListenerType _type);
virtual ~SrsListener();
public:
virtual int listen(int port);
private:
virtual void listen_cycle();
static void* listen_thread(void* arg);
// interface ISrsThreadHandler.
public:
virtual void on_enter_loop();
virtual int cycle();
};
class SrsServer : public SrsReloadHandler
class SrsServer : public ISrsReloadHandler
{
friend class SrsListener;
private:

5
trunk/src/core/srs_core_socket.cpp
View file

@ -52,6 +52,11 @@ void SrsSocket::set_send_timeout(int64_t timeout_us)
send_timeout = timeout_us;
}
int64_t SrsSocket::get_send_timeout()
{
return send_timeout;
}
int64_t SrsSocket::get_recv_bytes()
{
return recv_bytes;

3
trunk/src/core/srs_core_socket.hpp
View file

@ -30,8 +30,6 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <srs_core.hpp>
#include <st.h>
/**
* the socket provides TCP socket over st,
* that is, the sync socket mechanism.
@ -52,6 +50,7 @@ public:
virtual void set_recv_timeout(int64_t timeout_us);
virtual int64_t get_recv_timeout();
virtual void set_send_timeout(int64_t timeout_us);
virtual int64_t get_send_timeout();
virtual int64_t get_recv_bytes();
virtual int64_t get_send_bytes();
virtual int get_recv_kbps();

533
trunk/src/core/srs_core_source.cpp
View file

@ -24,6 +24,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <srs_core_source.hpp>
#include <algorithm>
using namespace std;
#include <srs_core_log.hpp>
#include <srs_core_protocol.hpp>
@ -37,8 +38,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <srs_core_rtmp.hpp>
#define CONST_MAX_JITTER_MS 500
#define DEFAULT_FRAME_TIME_MS 10
#define PAUSED_SHRINK_SIZE 250
#define DEFAULT_FRAME_TIME_MS 40
SrsRtmpJitter::SrsRtmpJitter()
{
@ -49,9 +49,15 @@ SrsRtmpJitter::~SrsRtmpJitter()
{
}
int SrsRtmpJitter::correct(SrsSharedPtrMessage* msg, int tba, int tbv, int64_t* corrected_time)
int SrsRtmpJitter::correct(SrsSharedPtrMessage* msg, int tba, int tbv)
{
int ret = ERROR_SUCCESS;
// set to 0 for metadata.
if (!msg->header.is_video() && !msg->header.is_audio()) {
msg->header.timestamp = 0;
return ret;
}
int sample_rate = tba;
int frame_rate = tbv;
@ -66,16 +72,16 @@ int SrsRtmpJitter::correct(SrsSharedPtrMessage* msg, int tba, int tbv, int64_t*
* 3. last_pkt_correct_time: simply add the positive delta,
* and enforce the time monotonically.
*/
u_int32_t time = msg->header.timestamp;
int32_t delta = time - last_pkt_time;
int64_t time = msg->header.timestamp;
int64_t delta = time - last_pkt_time;
// if jitter detected, reset the delta.
if (delta < 0 || delta > CONST_MAX_JITTER_MS) {
// calc the right diff by audio sample rate
if (msg->header.is_audio() && sample_rate > 0) {
delta = (int32_t)(delta * 1000.0 / sample_rate);
delta = (int64_t)(delta * 1000.0 / sample_rate);
} else if (msg->header.is_video() && frame_rate > 0) {
delta = (int32_t)(delta * 1.0 / frame_rate);
delta = (int64_t)(delta * 1.0 / frame_rate);
} else {
delta = DEFAULT_FRAME_TIME_MS;
}
@ -85,20 +91,16 @@ int SrsRtmpJitter::correct(SrsSharedPtrMessage* msg, int tba, int tbv, int64_t*
delta = DEFAULT_FRAME_TIME_MS;
}
srs_info("jitter detected, last_pts=%d, pts=%d, diff=%d, last_time=%d, time=%d, diff=%d",
srs_info("jitter detected, last_pts=%"PRId64", pts=%"PRId64", diff=%"PRId64", last_time=%"PRId64", time=%"PRId64", diff=%"PRId64"",
last_pkt_time, time, time - last_pkt_time, last_pkt_correct_time, last_pkt_correct_time + delta, delta);
} else {
srs_verbose("timestamp no jitter. time=%d, last_pkt=%d, correct_to=%d",
srs_verbose("timestamp no jitter. time=%"PRId64", last_pkt=%"PRId64", correct_to=%"PRId64"",
time, last_pkt_time, last_pkt_correct_time + delta);
}
last_pkt_correct_time = srs_max(0, last_pkt_correct_time + delta);
if (corrected_time) {
*corrected_time = last_pkt_correct_time;
}
msg->header.timestamp = last_pkt_correct_time;
last_pkt_time = time;
return ret;
@ -109,19 +111,152 @@ int SrsRtmpJitter::get_time()
return (int)last_pkt_correct_time;
}
SrsMessageQueue::SrsMessageQueue()
{
queue_size_ms = 0;
av_start_time = av_end_time = -1;
}
SrsMessageQueue::~SrsMessageQueue()
{
clear();
}
void SrsMessageQueue::set_queue_size(double queue_size)
{
queue_size_ms = (int)(queue_size * 1000);
}
int SrsMessageQueue::enqueue(SrsSharedPtrMessage* msg)
{
int ret = ERROR_SUCCESS;
if (msg->header.is_video() || msg->header.is_audio()) {
if (av_start_time == -1) {
av_start_time = msg->header.timestamp;
}
av_end_time = msg->header.timestamp;
}
msgs.push_back(msg);
while (av_end_time - av_start_time > queue_size_ms) {
shrink();
}
return ret;
}
int SrsMessageQueue::get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& count)
{
int ret = ERROR_SUCCESS;
if (msgs.empty()) {
return ret;
}
if (max_count == 0) {
count = (int)msgs.size();
} else {
count = srs_min(max_count, (int)msgs.size());
}
if (count <= 0) {
return ret;
}
pmsgs = new SrsSharedPtrMessage*[count];
for (int i = 0; i < count; i++) {
pmsgs[i] = msgs[i];
}
SrsSharedPtrMessage* last = msgs[count - 1];
av_start_time = last->header.timestamp;
if (count == (int)msgs.size()) {
msgs.clear();
} else {
msgs.erase(msgs.begin(), msgs.begin() + count);
}
return ret;
}
void SrsMessageQueue::shrink()
{
int iframe_index = -1;
// issue the first iframe.
// skip the first frame, whatever the type of it,
// for when we shrinked, the first is the iframe,
// we will directly remove the gop next time.
for (int i = 1; i < (int)msgs.size(); i++) {
SrsSharedPtrMessage* msg = msgs[i];
if (msg->header.is_video()) {
if (SrsCodec::video_is_keyframe(msg->payload, msg->size)) {
// the max frame index to remove.
iframe_index = i;
// set the start time, we will remove until this frame.
av_start_time = msg->header.timestamp;
break;
}
}
}
// no iframe, clear the queue.
if (iframe_index < 0) {
clear();
return;
}
srs_trace("shrink the cache queue, size=%d, removed=%d, max=%.2f",
(int)msgs.size(), iframe_index, queue_size_ms / 1000.0);
// remove the first gop from the front
for (int i = 0; i < iframe_index; i++) {
SrsSharedPtrMessage* msg = msgs[i];
srs_freep(msg);
}
msgs.erase(msgs.begin(), msgs.begin() + iframe_index);
}
void SrsMessageQueue::clear()
{
std::vector<SrsSharedPtrMessage*>::iterator it;
for (it = msgs.begin(); it != msgs.end(); ++it) {
SrsSharedPtrMessage* msg = *it;
srs_freep(msg);
}
msgs.clear();
av_start_time = av_end_time = -1;
}
SrsConsumer::SrsConsumer(SrsSource* _source)
{
source = _source;
paused = false;
jitter = new SrsRtmpJitter();
queue = new SrsMessageQueue();
}
SrsConsumer::~SrsConsumer()
{
clear();
source->on_consumer_destroy(this);
srs_freep(jitter);
srs_freep(queue);
}
void SrsConsumer::set_queue_size(double queue_size)
{
queue->set_queue_size(queue_size);
}
int SrsConsumer::get_time()
@ -138,46 +273,21 @@ int SrsConsumer::enqueue(SrsSharedPtrMessage* msg, int tba, int tbv)
return ret;
}
// TODO: check the queue size and drop packets if overflow.
msgs.push_back(msg);
if ((ret = queue->enqueue(msg)) != ERROR_SUCCESS) {
return ret;
}
return ret;
}
int SrsConsumer::get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& count)
{
int ret = ERROR_SUCCESS;
if (msgs.empty()) {
return ret;
}
// paused, return nothing.
if (paused) {
if ((int)msgs.size() >= PAUSED_SHRINK_SIZE) {
shrink();
}
return ret;
return ERROR_SUCCESS;
}
if (max_count == 0) {
count = (int)msgs.size();
} else {
count = srs_min(max_count, (int)msgs.size());
}
pmsgs = new SrsSharedPtrMessage*[count];
for (int i = 0; i < count; i++) {
pmsgs[i] = msgs[i];
}
if (count == (int)msgs.size()) {
msgs.clear();
} else {
msgs.erase(msgs.begin(), msgs.begin() + count);
}
return ret;
return queue->get_packets(max_count, pmsgs, count);
}
int SrsConsumer::on_play_client_pause(bool is_pause)
@ -190,68 +300,6 @@ int SrsConsumer::on_play_client_pause(bool is_pause)
return ret;
}
void SrsConsumer::shrink()
{
int i = 0;
std::vector<SrsSharedPtrMessage*>::iterator it;
// issue the last video iframe.
bool has_video = false;
int frame_to_remove = 0;
std::vector<SrsSharedPtrMessage*>::iterator iframe = msgs.end();
for (i = 0, it = msgs.begin(); it != msgs.end(); ++it, i++) {
SrsSharedPtrMessage* msg = *it;
if (msg->header.is_video()) {
has_video = true;
if (SrsCodec::video_is_keyframe(msg->payload, msg->size)) {
iframe = it;
frame_to_remove = i + 1;
}
}
}
// last iframe is the first elem, ignore it.
if (iframe == msgs.begin()) {
return;
}
// recalc the frame to remove
if (iframe == msgs.end()) {
frame_to_remove = 0;
}
if (!has_video) {
frame_to_remove = (int)msgs.size();
}
srs_trace("shrink the cache queue, has_video=%d, has_iframe=%d, size=%d, removed=%d",
has_video, iframe != msgs.end(), (int)msgs.size(), frame_to_remove);
// if no video, remove all audio.
if (!has_video) {
clear();
return;
}
// if exists video Iframe, remove the frames before it.
if (iframe != msgs.end()) {
for (it = msgs.begin(); it != iframe; ++it) {
SrsSharedPtrMessage* msg = *it;
srs_freep(msg);
}
msgs.erase(msgs.begin(), iframe);
}
}
void SrsConsumer::clear()
{
std::vector<SrsSharedPtrMessage*>::iterator it;
for (it = msgs.begin(); it != msgs.end(); ++it) {
SrsSharedPtrMessage* msg = *it;
srs_freep(msg);
}
msgs.clear();
}
SrsGopCache::SrsGopCache()
{
cached_video_count = 0;
@ -344,22 +392,25 @@ int SrsGopCache::dump(SrsConsumer* consumer, int tba, int tbv)
std::map<std::string, SrsSource*> SrsSource::pool;
SrsSource* SrsSource::find(const std::string &stream_url)
SrsSource* SrsSource::find(SrsRequest* req)
{
string stream_url = req->get_stream_url();
string vhost = req->vhost;
if (pool.find(stream_url) == pool.end()) {
pool[stream_url] = new SrsSource(stream_url);
srs_verbose("create new source for url=%s", stream_url.c_str());
pool[stream_url] = new SrsSource(req);
srs_verbose("create new source for url=%s, vhost=%s", stream_url.c_str(), vhost.c_str());
}
return pool[stream_url];
}
SrsSource::SrsSource(std::string _stream_url)
SrsSource::SrsSource(SrsRequest* _req)
{
stream_url = _stream_url;
req = _req->copy();
#ifdef SRS_HLS
hls = new SrsHls();
hls = new SrsHls(this);
#endif
#ifdef SRS_FFMPEG
encoder = new SrsEncoder();
@ -371,10 +422,14 @@ SrsSource::SrsSource(std::string _stream_url)
_can_publish = true;
gop_cache = new SrsGopCache();
config->subscribe(this);
}
SrsSource::~SrsSource()
{
config->unsubscribe(this);
if (true) {
std::vector<SrsConsumer*>::iterator it;
for (it = consumers.begin(); it != consumers.end(); ++it) {
@ -405,6 +460,167 @@ SrsSource::~SrsSource()
#ifdef SRS_FFMPEG
srs_freep(encoder);
#endif
srs_freep(req);
}
int SrsSource::on_reload_gop_cache(string vhost)
{
int ret = ERROR_SUCCESS;
if (req->vhost != vhost) {
return ret;
}
// gop cache changed.
bool enabled_cache = config->get_gop_cache(vhost);
srs_trace("vhost %s gop_cache changed to %d, source url=%s",
vhost.c_str(), enabled_cache, req->get_stream_url().c_str());
set_cache(enabled_cache);
return ret;
}
int SrsSource::on_reload_queue_length(string vhost)
{
int ret = ERROR_SUCCESS;
if (req->vhost != vhost) {
return ret;
}
double queue_size = config->get_queue_length(req->vhost);
if (true) {
std::vector<SrsConsumer*>::iterator it;
for (it = consumers.begin(); it != consumers.end(); ++it) {
SrsConsumer* consumer = *it;
consumer->set_queue_size(queue_size);
}
srs_trace("consumers reload queue size success.");
}
if (true) {
std::vector<SrsForwarder*>::iterator it;
for (it = forwarders.begin(); it != forwarders.end(); ++it) {
SrsForwarder* forwarder = *it;
forwarder->set_queue_size(queue_size);
}
srs_trace("forwarders reload queue size success.");
}
return ret;
}
int SrsSource::on_reload_forward(string vhost)
{
int ret = ERROR_SUCCESS;
if (req->vhost != vhost) {
return ret;
}
// forwarders
destroy_forwarders();
if ((ret = create_forwarders()) != ERROR_SUCCESS) {
srs_error("create forwarders failed. ret=%d", ret);
return ret;
}
srs_trace("vhost %s forwarders reload success", vhost.c_str());
return ret;
}
int SrsSource::on_reload_hls(string vhost)
{
int ret = ERROR_SUCCESS;
if (req->vhost != vhost) {
return ret;
}
#ifdef SRS_HLS
hls->on_unpublish();
if ((ret = hls->on_publish(req)) != ERROR_SUCCESS) {
srs_error("hls publish failed. ret=%d", ret);
return ret;
}
srs_trace("vhost %s hls reload success", vhost.c_str());
#endif
return ret;
}
int SrsSource::on_reload_transcode(string vhost)
{
int ret = ERROR_SUCCESS;
if (req->vhost != vhost) {
return ret;
}
#ifdef SRS_FFMPEG
encoder->on_unpublish();
if ((ret = encoder->on_publish(req)) != ERROR_SUCCESS) {
srs_error("start encoder failed. ret=%d", ret);
return ret;
}
srs_trace("vhost %s transcode reload success", vhost.c_str());
#endif
return ret;
}
int SrsSource::on_forwarder_start(SrsForwarder* forwarder)
{
int ret = ERROR_SUCCESS;
// feed the forwarder the metadata/sequence header,
// when reload to enable the forwarder.
if (cache_metadata && (ret = forwarder->on_meta_data(cache_metadata->copy())) != ERROR_SUCCESS) {
srs_error("forwarder process onMetaData message failed. ret=%d", ret);
return ret;
}
if (cache_sh_video && (ret = forwarder->on_video(cache_sh_video->copy())) != ERROR_SUCCESS) {
srs_error("forwarder process video sequence header message failed. ret=%d", ret);
return ret;
}
if (cache_sh_audio && (ret = forwarder->on_audio(cache_sh_audio->copy())) != ERROR_SUCCESS) {
srs_error("forwarder process audio sequence header message failed. ret=%d", ret);
return ret;
}
return ret;
}
int SrsSource::on_hls_start()
{
int ret = ERROR_SUCCESS;
#ifdef SRS_HLS
// feed the hls the metadata/sequence header,
// when reload to enable the hls.
// TODO: maybe need to decode the metadata?
if (cache_sh_video && (ret = hls->on_video(cache_sh_video->copy())) != ERROR_SUCCESS) {
srs_error("hls process video sequence header message failed. ret=%d", ret);
return ret;
}
if (cache_sh_audio && (ret = hls->on_audio(cache_sh_audio->copy())) != ERROR_SUCCESS) {
srs_error("hls process audio sequence header message failed. ret=%d", ret);
return ret;
}
#endif
return ret;
}
bool SrsSource::can_publish()
@ -417,7 +633,7 @@ int SrsSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata
int ret = ERROR_SUCCESS;
#ifdef SRS_HLS
if ((ret = hls->on_meta_data(metadata)) != ERROR_SUCCESS) {
if (metadata && (ret = hls->on_meta_data(metadata->metadata)) != ERROR_SUCCESS) {
srs_error("hls process onMetaData message failed. ret=%d", ret);
return ret;
}
@ -425,6 +641,8 @@ int SrsSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata
metadata->metadata->set("server", new SrsAmf0String(
RTMP_SIG_SRS_KEY" "RTMP_SIG_SRS_VERSION" ("RTMP_SIG_SRS_URL_SHORT")"));
metadata->metadata->set("contributor",
new SrsAmf0String(RTMP_SIG_SRS_CONTRIBUTOR));
SrsAmf0Any* prop = NULL;
if ((prop = metadata->metadata->get_property("audiosamplerate")) != NULL) {
@ -620,7 +838,7 @@ int SrsSource::on_video(SrsCommonMessage* video)
// cache the last gop packets
if ((ret = gop_cache->cache(msg)) != ERROR_SUCCESS) {
srs_error("shrink gop cache failed. ret=%d", ret);
srs_error("gop cache msg failed. ret=%d", ret);
return ret;
}
srs_verbose("cache gop success.");
@ -628,39 +846,33 @@ int SrsSource::on_video(SrsCommonMessage* video)
return ret;
}
int SrsSource::on_publish(SrsRequest* req)
int SrsSource::on_publish(SrsRequest* _req)
{
int ret = ERROR_SUCCESS;
// update the request object.
srs_freep(req);
req = _req->copy();
srs_assert(req);
_can_publish = false;
// TODO: support reload.
// create forwarders
SrsConfDirective* conf = config->get_forward(req->vhost);
for (int i = 0; conf && i < (int)conf->args.size(); i++) {
std::string forward_server = conf->args.at(i);
SrsForwarder* forwarder = new SrsForwarder();
forwarders.push_back(forwarder);
if ((ret = forwarder->on_publish(req, forward_server)) != ERROR_SUCCESS) {
srs_error("start forwarder failed. "
"vhost=%s, app=%s, stream=%s, forward-to=%s",
req->vhost.c_str(), req->app.c_str(), req->stream.c_str(),
forward_server.c_str());
return ret;
}
if ((ret = create_forwarders()) != ERROR_SUCCESS) {
srs_error("create forwarders failed. ret=%d", ret);
return ret;
}
#ifdef SRS_FFMPEG
if ((ret = encoder->on_publish(req)) != ERROR_SUCCESS) {
srs_error("start encoder failed. ret=%d", ret);
return ret;
}
#endif
#ifdef SRS_HLS
if ((ret = hls->on_publish(req)) != ERROR_SUCCESS) {
srs_error("start hls failed. ret=%d", ret);
return ret;
}
#endif
@ -670,19 +882,14 @@ int SrsSource::on_publish(SrsRequest* req)
void SrsSource::on_unpublish()
{
// close all forwarders
std::vector<SrsForwarder*>::iterator it;
for (it = forwarders.begin(); it != forwarders.end(); ++it) {
SrsForwarder* forwarder = *it;
forwarder->on_unpublish();
srs_freep(forwarder);
}
forwarders.clear();
// destroy all forwarders
destroy_forwarders();
#ifdef SRS_FFMPEG
encoder->on_unpublish();
#endif
// TODO: HLS should continue previous sequence and stream.
#ifdef SRS_HLS
hls->on_unpublish();
#endif
@ -706,6 +913,9 @@ void SrsSource::on_unpublish()
consumer = new SrsConsumer(this);
consumers.push_back(consumer);
double queue_size = config->get_queue_length(req->vhost);
consumer->set_queue_size(queue_size);
if (cache_metadata && (ret = consumer->enqueue(cache_metadata->copy(), sample_rate, frame_rate)) != ERROR_SUCCESS) {
srs_error("dispatch metadata failed. ret=%d", ret);
@ -729,6 +939,8 @@ void SrsSource::on_unpublish()
return ret;
}
srs_trace("create consumer, queue_size=%.2f, tba=%d, tbv=%d", queue_size, sample_rate, frame_rate);
return ret;
}
@ -747,3 +959,40 @@ void SrsSource::set_cache(bool enabled)
gop_cache->set(enabled);
}
int SrsSource::create_forwarders()
{
int ret = ERROR_SUCCESS;
SrsConfDirective* conf = config->get_forward(req->vhost);
for (int i = 0; conf && i < (int)conf->args.size(); i++) {
std::string forward_server = conf->args.at(i);
SrsForwarder* forwarder = new SrsForwarder(this);
forwarders.push_back(forwarder);
double queue_size = config->get_queue_length(req->vhost);
forwarder->set_queue_size(queue_size);
if ((ret = forwarder->on_publish(req, forward_server)) != ERROR_SUCCESS) {
srs_error("start forwarder failed. "
"vhost=%s, app=%s, stream=%s, forward-to=%s",
req->vhost.c_str(), req->app.c_str(), req->stream.c_str(),
forward_server.c_str());
return ret;
}
}
return ret;
}
void SrsSource::destroy_forwarders()
{
std::vector<SrsForwarder*>::iterator it;
for (it = forwarders.begin(); it != forwarders.end(); ++it) {
SrsForwarder* forwarder = *it;
forwarder->on_unpublish();
srs_freep(forwarder);
}
forwarders.clear();
}

100
trunk/src/core/srs_core_source.hpp Normal file → Executable file
View file

@ -34,6 +34,8 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#include <vector>
#include <string>
#include <srs_core_reload.hpp>
class SrsSource;
class SrsCommonMessage;
class SrsOnMetaDataPacket;
@ -54,24 +56,64 @@ class SrsEncoder;
class SrsRtmpJitter
{
private:
u_int32_t last_pkt_time;
u_int32_t last_pkt_correct_time;
int64_t last_pkt_time;
int64_t last_pkt_correct_time;
public:
SrsRtmpJitter();
virtual ~SrsRtmpJitter();
public:
/**
* detect the time jitter and correct it.
* @param corrected_time output the 64bits time.
* ignore if NULL.
*/
virtual int correct(SrsSharedPtrMessage* msg, int tba, int tbv, int64_t* corrected_time = NULL);
virtual int correct(SrsSharedPtrMessage* msg, int tba, int tbv);
/**
* get current client time, the last packet time.
*/
virtual int get_time();
};
/**
* the message queue for the consumer(client), forwarder.
* we limit the size in seconds, drop old messages(the whole gop) if full.
*/
class SrsMessageQueue
{
private:
int64_t av_start_time;
int64_t av_end_time;
int queue_size_ms;
std::vector<SrsSharedPtrMessage*> msgs;
public:
SrsMessageQueue();
virtual ~SrsMessageQueue();
public:
/**
* set the queue size
* @param queue_size the queue size in seconds.
*/
virtual void set_queue_size(double queue_size);
public:
/**
* enqueue the message, the timestamp always monotonically.
* @param msg, the msg to enqueue, user never free it whatever the return code.
*/
virtual int enqueue(SrsSharedPtrMessage* msg);
/**
* get packets in consumer queue.
* @pmsgs SrsMessages*[], output the prt array.
* @count the count in array.
* @max_count the max count to dequeue, 0 to dequeue all.
*/
virtual int get_packets(int max_count, SrsSharedPtrMessage**& pmsgs, int& count);
private:
/**
* remove a gop from the front.
* if no iframe found, clear it.
*/
virtual void shrink();
virtual void clear();
};
/**
* the consumer for SrsSource, that is a play client.
*/
@ -80,11 +122,13 @@ class SrsConsumer
private:
SrsRtmpJitter* jitter;
SrsSource* source;
std::vector<SrsSharedPtrMessage*> msgs;
SrsMessageQueue* queue;
bool paused;
public:
SrsConsumer(SrsSource* _source);
virtual ~SrsConsumer();
public:
virtual void set_queue_size(double queue_size);
public:
/**
* get current client time, the last packet time.
@ -109,13 +153,6 @@ public:
* when client send the pause message.
*/
virtual int on_play_client_pause(bool is_pause);
private:
/**
* when paused, shrink the cache queue,
* remove to cache only one gop.
*/
virtual void shrink();
virtual void clear();
};
/**
@ -158,20 +195,21 @@ public:
/**
* live streaming source.
*/
class SrsSource
class SrsSource : public ISrsReloadHandler
{
private:
static std::map<std::string, SrsSource*> pool;
public:
/**
* find stream by vhost/app/stream.
* @stream_url the stream url, for example, myserver.xxx.com/app/stream
* @param req the client request.
* @return the matched source, never be NULL.
* @remark stream_url should without port and schema.
*/
static SrsSource* find(const std::string& stream_url);
static SrsSource* find(SrsRequest* req);
private:
std::string stream_url;
// deep copy of client request.
SrsRequest* req;
// to delivery stream to clients.
std::vector<SrsConsumer*> consumers;
// hls handler.
@ -206,19 +244,43 @@ private:
// the cached audio sequence header.
SrsSharedPtrMessage* cache_sh_audio;
public:
SrsSource(std::string _stream_url);
/**
* @param _req the client request object,
* this object will deep copy it for reload.
*/
SrsSource(SrsRequest* _req);
virtual ~SrsSource();
// interface ISrsReloadHandler
public:
virtual int on_reload_gop_cache(std::string vhost);
virtual int on_reload_queue_length(std::string vhost);
virtual int on_reload_forward(std::string vhost);
virtual int on_reload_hls(std::string vhost);
virtual int on_reload_transcode(std::string vhost);
public:
// for the SrsForwarder to callback to request the sequence headers.
virtual int on_forwarder_start(SrsForwarder* forwarder);
// for the SrsHls to callback to request the sequence headers.
virtual int on_hls_start();
public:
virtual bool can_publish();
virtual int on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata);
virtual int on_audio(SrsCommonMessage* audio);
virtual int on_video(SrsCommonMessage* video);
virtual int on_publish(SrsRequest* req);
/**
* publish stream event notify.
* @param _req the request from client, the source will deep copy it,
* for when reload the request of client maybe invalid.
*/
virtual int on_publish(SrsRequest* _req);
virtual void on_unpublish();
public:
virtual int create_consumer(SrsConsumer*& consumer);
virtual void on_consumer_destroy(SrsConsumer* consumer);
virtual void set_cache(bool enabled);
private:
virtual int create_forwarders();
virtual void destroy_forwarders();
};
#endif

156
trunk/src/core/srs_core_thread.cpp Normal file
View file

@ -0,0 +1,156 @@
/*
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_thread.hpp>
#include <srs_core_error.hpp>
#include <srs_core_log.hpp>
ISrsThreadHandler::ISrsThreadHandler()
{
}
ISrsThreadHandler::~ISrsThreadHandler()
{
}
void ISrsThreadHandler::on_enter_loop()
{
}
int ISrsThreadHandler::on_before_cycle()
{
int ret = ERROR_SUCCESS;
return ret;
}
int ISrsThreadHandler::on_end_cycle()
{
int ret = ERROR_SUCCESS;
return ret;
}
void ISrsThreadHandler::on_leave_loop()
{
}
SrsThread::SrsThread(ISrsThreadHandler* thread_handler, int64_t interval_ms)
{
handler = thread_handler;
cycle_interval_milliseconds = interval_ms;
tid = NULL;
loop = false;
}
SrsThread::~SrsThread()
{
stop();
}
int SrsThread::start()
{
int ret = ERROR_SUCCESS;
if(tid) {
srs_info("thread already running.");
return ret;
}
if((tid = st_thread_create(thread_fun, this, 1, 0)) == NULL){
ret = ERROR_ST_CREATE_CYCLE_THREAD;
srs_error("st_thread_create failed. ret=%d", ret);
return ret;
}
return ret;
}
void SrsThread::stop()
{
if (tid) {
loop = false;
// the interrupt will cause the socket to read/write error,
// which will terminate the cycle thread.
st_thread_interrupt(tid);
// wait the thread to exit.
st_thread_join(tid, NULL);
tid = NULL;
}
}
void SrsThread::thread_cycle()
{
int ret = ERROR_SUCCESS;
srs_assert(handler);
log_context->generate_id();
srs_trace("thread cycle start");
handler->on_end_cycle();
loop = true;
while (loop) {
if ((ret = handler->on_before_cycle()) != ERROR_SUCCESS) {
srs_warn("thread on before cycle failed, ignored and retry, ret=%d", ret);
goto failed;
}
srs_info("thread on before cycle success");
if ((ret = handler->cycle()) != ERROR_SUCCESS) {
srs_warn("thread cycle failed, ignored and retry, ret=%d", ret);
goto failed;
}
srs_info("thread cycle success");
if ((ret = handler->on_end_cycle()) != ERROR_SUCCESS) {
srs_warn("thread on end cycle failed, ignored and retry, ret=%d", ret);
goto failed;
}
srs_info("thread on end cycle success");
failed:
if (!loop) {
break;
}
st_usleep(cycle_interval_milliseconds * 1000);
}
handler->on_leave_loop();
srs_trace("thread cycle finished");
}
void* SrsThread::thread_fun(void* arg)
{
SrsThread* obj = (SrsThread*)arg;
srs_assert(obj);
obj->thread_cycle();
return NULL;
}

98
trunk/src/core/srs_core_thread.hpp Normal file
View file

@ -0,0 +1,98 @@
/*
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_THREAD_HPP
#define SRS_CORE_THREAD_HPP
/*
#include <srs_core_thread.hpp>
*/
#include <srs_core.hpp>
/**
* the handler for the thread, callback interface.
* the thread model defines as:
* handler->on_enter_loop()
* while loop:
* handler->on_before_cycle()
* handler->cycle()
* handler->on_end_cycle()
* if !loop then break for user stop thread.
* sleep(CycleIntervalMilliseconds)
* handler->on_leave_loop()
* when stop, the thread will interrupt the st_thread,
* which will cause the socket to return error and
* terminate the cycle thread.
*/
class ISrsThreadHandler
{
public:
ISrsThreadHandler();
virtual ~ISrsThreadHandler();
public:
virtual void on_enter_loop();
virtual int on_before_cycle();
virtual int cycle() = 0;
virtual int on_end_cycle();
virtual void on_leave_loop();
};
/**
* provides servies from st_thread_t,
* for common thread usage.
*/
class SrsThread
{
private:
st_thread_t tid;
bool loop;
private:
ISrsThreadHandler* handler;
int64_t cycle_interval_milliseconds;
public:
/**
* initialize the thread.
* @param thread_handler, the cycle handler for the thread.
* @param interval_ms, the sleep interval when cycle finished.
*/
SrsThread(ISrsThreadHandler* thread_handler, int64_t interval_ms);
virtual ~SrsThread();
public:
/**
* start the thread, invoke the cycle of handler util
* user stop the thread.
* @remark ignore any error of cycle of handler.
* @remark user can start multiple times, ignore if already started.
*/
virtual int start();
/**
* stop the thread, wait for the thread to terminate.
* @remark user can stop multiple times, ignore if already stopped.
*/
virtual void stop();
private:
virtual void thread_cycle();
static void* thread_fun(void* arg);
};
#endif

114
trunk/src/srs/srs.upp
View file

@ -1,60 +1,62 @@
file
main readonly separator,
..\main\srs_main_server.cpp,
auto readonly separator,
..\..\objs\srs_auto_headers.hpp,
core readonly separator,
..\core\srs_core.hpp,
..\core\srs_core.cpp,
..\core\srs_core_error.hpp,
..\core\srs_core_error.cpp,
..\core\srs_core_autofree.hpp,
..\core\srs_core_autofree.cpp,
..\core\srs_core_server.hpp,
..\core\srs_core_server.cpp,
..\core\srs_core_reload.hpp,
..\core\srs_core_reload.cpp,
..\core\srs_core_config.hpp,
..\core\srs_core_config.cpp,
..\core\srs_core_refer.hpp,
..\core\srs_core_refer.cpp,
..\core\srs_core_conn.hpp,
..\core\srs_core_conn.cpp,
..\core\srs_core_client.hpp,
..\core\srs_core_client.cpp,
..\core\srs_core_http.hpp,
..\core\srs_core_http.cpp,
..\core\srs_core_source.hpp,
..\core\srs_core_source.cpp,
..\core\srs_core_forward.hpp,
..\core\srs_core_forward.cpp,
..\core\srs_core_encoder.hpp,
..\core\srs_core_encoder.cpp,
..\core\srs_core_hls.hpp,
..\core\srs_core_hls.cpp,
..\core\srs_core_codec.hpp,
..\core\srs_core_codec.cpp,
..\core\srs_core_rtmp.hpp,
..\core\srs_core_rtmp.cpp,
..\core\srs_core_handshake.hpp,
..\core\srs_core_handshake.cpp,
..\core\srs_core_protocol.hpp,
..\core\srs_core_protocol.cpp,
..\core\srs_core_amf0.hpp,
..\core\srs_core_amf0.cpp,
..\core\srs_core_stream.hpp,
..\core\srs_core_stream.cpp,
..\core\srs_core_socket.hpp,
..\core\srs_core_socket.cpp,
..\core\srs_core_buffer.hpp,
..\core\srs_core_buffer.cpp,
..\core\srs_core_pithy_print.hpp,
..\core\srs_core_pithy_print.cpp,
..\core\srs_core_log.hpp,
..\core\srs_core_log.cpp,
research readonly separator,
..\..\research\ts_info.cc;
main readonly separator,
..\main\srs_main_server.cpp,
auto readonly separator,
..\..\objs\srs_auto_headers.hpp,
core readonly separator,
..\core\srs_core.hpp,
..\core\srs_core.cpp,
..\core\srs_core_amf0.hpp,
..\core\srs_core_amf0.cpp,
..\core\srs_core_autofree.hpp,
..\core\srs_core_autofree.cpp,
..\core\srs_core_buffer.hpp,
..\core\srs_core_buffer.cpp,
..\core\srs_core_client.hpp,
..\core\srs_core_client.cpp,
..\core\srs_core_codec.hpp,
..\core\srs_core_codec.cpp,
..\core\srs_core_config.hpp,
..\core\srs_core_config.cpp,
..\core\srs_core_conn.hpp,
..\core\srs_core_conn.cpp,
..\core\srs_core_encoder.hpp,
..\core\srs_core_encoder.cpp,
..\core\srs_core_error.hpp,
..\core\srs_core_error.cpp,
..\core\srs_core_forward.hpp,
..\core\srs_core_forward.cpp,
..\core\srs_core_handshake.hpp,
..\core\srs_core_handshake.cpp,
..\core\srs_core_hls.hpp,
..\core\srs_core_hls.cpp,
..\core\srs_core_http.hpp,
..\core\srs_core_http.cpp,
..\core\srs_core_log.hpp,
..\core\srs_core_log.cpp,
..\core\srs_core_pithy_print.hpp,
..\core\srs_core_pithy_print.cpp,
..\core\srs_core_protocol.hpp,
..\core\srs_core_protocol.cpp,
..\core\srs_core_refer.hpp,
..\core\srs_core_refer.cpp,
..\core\srs_core_reload.hpp,
..\core\srs_core_reload.cpp,
..\core\srs_core_rtmp.hpp,
..\core\srs_core_rtmp.cpp,
..\core\srs_core_thread.hpp,
..\core\srs_core_thread.cpp,
..\core\srs_core_server.hpp,
..\core\srs_core_server.cpp,
..\core\srs_core_stream.hpp,
..\core\srs_core_stream.cpp,
..\core\srs_core_socket.hpp,
..\core\srs_core_socket.cpp,
..\core\srs_core_source.hpp,
..\core\srs_core_source.cpp,
research readonly separator,
..\..\research\ts_info.cc;
mainconfig
"" = "MAIN";
"" = "MAIN";