winlin

Merge branch '2.0release' into develop

@@ -13,6 +13,7 @@ RMB 500-999 @@ -13,6 +13,7 @@ RMB 500-999
13 * [2015-xx-xx xx:xx] xxx 13 * [2015-xx-xx xx:xx] xxx
14 14
15 RMB 100-499 15 RMB 100-499
  16 +* [2015-04-10 19:52] 阳成飞
16 * [2015-03-30 13:34] 扶凯 17 * [2015-03-30 13:34] 扶凯
17 * [2015-03-29 11-07] 姚伟斌 18 * [2015-03-29 11-07] 姚伟斌
18 * [2015-03-14 20:21] 万伟 19 * [2015-03-14 20:21] 万伟
@@ -613,6 +613,11 @@ vhost with-hls.srs.com { @@ -613,6 +613,11 @@ vhost with-hls.srs.com {
613 # whether cleanup the old ts files. 613 # whether cleanup the old ts files.
614 # default: on 614 # default: on
615 hls_cleanup on; 615 hls_cleanup on;
  616 + # the max size to notify hls,
  617 + # to read max bytes from ts of specified cdn network,
  618 + # @remark only used when on_hls_notify is config.
  619 + # default: 64
  620 + hls_nb_notify 64;
616 621
617 # on_hls, never config in here, should config in http_hooks. 622 # on_hls, never config in here, should config in http_hooks.
618 # for the hls http callback, @see http_hooks.on_hls of vhost hooks.callback.srs.com 623 # for the hls http callback, @see http_hooks.on_hls of vhost hooks.callback.srs.com
@@ -36,7 +36,7 @@ reload(sys) @@ -36,7 +36,7 @@ reload(sys)
36 exec("sys.setdefaultencoding('utf-8')") 36 exec("sys.setdefaultencoding('utf-8')")
37 assert sys.getdefaultencoding().lower() == "utf-8" 37 assert sys.getdefaultencoding().lower() == "utf-8"
38 38
39 -import os, json, time, datetime, cherrypy, threading 39 +import os, json, time, datetime, cherrypy, threading, urllib2
40 40
41 # simple log functions. 41 # simple log functions.
42 def trace(msg): 42 def trace(msg):
@@ -308,6 +308,41 @@ class RESTDvrs(object): @@ -308,6 +308,41 @@ class RESTDvrs(object):
308 308
309 return code 309 return code
310 310
  311 +
  312 +'''
  313 +handle the hls proxy requests: hls stream.
  314 +'''
  315 +class RESTProxy(object):
  316 + exposed = True
  317 +
  318 + '''
  319 + for SRS hook: on_hls_notify
  320 + on_hls_notify:
  321 + when srs reap a ts file of hls, call this hook,
  322 + used to push file to cdn network, by get the ts file from cdn network.
  323 + so we use HTTP GET and use the variable following:
  324 + [app], replace with the app.
  325 + [stream], replace with the stream.
  326 + [ts_url], replace with the ts url.
  327 + ignore any return data of server.
  328 + '''
  329 + def GET(self, *args, **kwargs):
  330 + enable_crossdomain()
  331 +
  332 + url = "http://" + "/".join(args);
  333 + print "start to proxy url: %s"%url
  334 +
  335 + f = None
  336 + try:
  337 + f = urllib2.urlopen(url)
  338 + f.read()
  339 + except:
  340 + print "error proxy url: %s"%url
  341 + finally:
  342 + if f: f.close()
  343 + print "completed proxy url: %s"%url
  344 + return url
  345 +
311 ''' 346 '''
312 handle the hls requests: hls stream. 347 handle the hls requests: hls stream.
313 ''' 348 '''
@@ -1195,6 +1230,7 @@ class V1(object): @@ -1195,6 +1230,7 @@ class V1(object):
1195 self.sessions = RESTSessions() 1230 self.sessions = RESTSessions()
1196 self.dvrs = RESTDvrs() 1231 self.dvrs = RESTDvrs()
1197 self.hls = RESTHls() 1232 self.hls = RESTHls()
  1233 + self.proxy = RESTProxy()
1198 self.chats = RESTChats() 1234 self.chats = RESTChats()
1199 self.servers = RESTServers() 1235 self.servers = RESTServers()
1200 self.nodes = RESTNodes() 1236 self.nodes = RESTNodes()
@@ -87,7 +87,7 @@ int SrsDvrAsyncCallThread::cycle() @@ -87,7 +87,7 @@ int SrsDvrAsyncCallThread::cycle()
87 for (it = copies.begin(); it != copies.end(); ++it) { 87 for (it = copies.begin(); it != copies.end(); ++it) {
88 ISrsDvrAsyncCall* call = *it; 88 ISrsDvrAsyncCall* call = *it;
89 if ((ret = call->call()) != ERROR_SUCCESS) { 89 if ((ret = call->call()) != ERROR_SUCCESS) {
90 - srs_warn("dvr: ignore callback %s, ret=%d", call->to_string().c_str(), ret); 90 + srs_warn("ignore async callback %s, ret=%d", call->to_string().c_str(), ret);
91 } 91 }
92 srs_freep(call); 92 srs_freep(call);
93 } 93 }
@@ -1487,7 +1487,7 @@ int SrsConfig::check_config() @@ -1487,7 +1487,7 @@ int SrsConfig::check_config()
1487 string m = conf->at(j)->name.c_str(); 1487 string m = conf->at(j)->name.c_str();
1488 if (m != "enabled" && m != "hls_entry_prefix" && m != "hls_path" && m != "hls_fragment" && m != "hls_window" && m != "hls_on_error" 1488 if (m != "enabled" && m != "hls_entry_prefix" && m != "hls_path" && m != "hls_fragment" && m != "hls_window" && m != "hls_on_error"
1489 && m != "hls_storage" && m != "hls_mount" && m != "hls_td_ratio" && m != "hls_aof_ratio" && m != "hls_acodec" && m != "hls_vcodec" 1489 && m != "hls_storage" && m != "hls_mount" && m != "hls_td_ratio" && m != "hls_aof_ratio" && m != "hls_acodec" && m != "hls_vcodec"
1490 - && m != "hls_m3u8_file" && m != "hls_ts_file" && m != "hls_ts_floor" && m != "hls_cleanup" 1490 + && m != "hls_m3u8_file" && m != "hls_ts_file" && m != "hls_ts_floor" && m != "hls_cleanup" && m != "hls_nb_notify"
1491 ) { 1491 ) {
1492 ret = ERROR_SYSTEM_CONFIG_INVALID; 1492 ret = ERROR_SYSTEM_CONFIG_INVALID;
1493 srs_error("unsupported vhost hls directive %s, ret=%d", m.c_str(), ret); 1493 srs_error("unsupported vhost hls directive %s, ret=%d", m.c_str(), ret);
@@ -3418,6 +3418,22 @@ string SrsConfig::get_hls_vcodec(string vhost) @@ -3418,6 +3418,22 @@ string SrsConfig::get_hls_vcodec(string vhost)
3418 return conf->arg0(); 3418 return conf->arg0();
3419 } 3419 }
3420 3420
  3421 +int SrsConfig::get_vhost_hls_nb_notify(string vhost)
  3422 +{
  3423 + SrsConfDirective* conf = get_hls(vhost);
  3424 +
  3425 + if (!conf) {
  3426 + return SRS_CONF_DEFAULT_HLS_NB_NOTIFY;
  3427 + }
  3428 +
  3429 + conf = conf->get("hls_nb_notify");
  3430 + if (!conf || conf->arg0().empty()) {
  3431 + return SRS_CONF_DEFAULT_HLS_NB_NOTIFY;
  3432 + }
  3433 +
  3434 + return ::atoi(conf->arg0().c_str());
  3435 +}
  3436 +
3421 bool SrsConfig::get_hls_cleanup(string vhost) 3437 bool SrsConfig::get_hls_cleanup(string vhost)
3422 { 3438 {
3423 SrsConfDirective* hls = get_hls(vhost); 3439 SrsConfDirective* hls = get_hls(vhost);
@@ -63,6 +63,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. @@ -63,6 +63,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
63 #define SRS_CONF_DEFAULT_HLS_ACODEC "aac" 63 #define SRS_CONF_DEFAULT_HLS_ACODEC "aac"
64 #define SRS_CONF_DEFAULT_HLS_VCODEC "h264" 64 #define SRS_CONF_DEFAULT_HLS_VCODEC "h264"
65 #define SRS_CONF_DEFAULT_HLS_CLEANUP true 65 #define SRS_CONF_DEFAULT_HLS_CLEANUP true
  66 +#define SRS_CONF_DEFAULT_HLS_NB_NOTIFY 64
66 #define SRS_CONF_DEFAULT_DVR_PATH "./objs/nginx/html/[app]/[stream].[timestamp].flv" 67 #define SRS_CONF_DEFAULT_DVR_PATH "./objs/nginx/html/[app]/[stream].[timestamp].flv"
67 #define SRS_CONF_DEFAULT_DVR_PLAN_SESSION "session" 68 #define SRS_CONF_DEFAULT_DVR_PLAN_SESSION "session"
68 #define SRS_CONF_DEFAULT_DVR_PLAN_SEGMENT "segment" 69 #define SRS_CONF_DEFAULT_DVR_PLAN_SEGMENT "segment"
@@ -959,8 +960,12 @@ public: @@ -959,8 +960,12 @@ public:
959 * whether cleanup the old ts files. 960 * whether cleanup the old ts files.
960 */ 961 */
961 virtual bool get_hls_cleanup(std::string vhost); 962 virtual bool get_hls_cleanup(std::string vhost);
962 -  
963 - // hds section 963 + /**
  964 + * get the size of bytes to read from cdn network, for the on_hls_notify callback,
  965 + * that is, to read max bytes of the bytes from the callback, or timeout or error.
  966 + */
  967 + virtual int get_vhost_hls_nb_notify(std::string vhost);
  968 +// hds section
964 private: 969 private:
965 /** 970 /**
966 * get the hds directive of vhost. 971 * get the hds directive of vhost.
@@ -240,10 +240,11 @@ int SrsDvrAsyncCallOnHlsNotify::call() @@ -240,10 +240,11 @@ int SrsDvrAsyncCallOnHlsNotify::call()
240 return ret; 240 return ret;
241 } 241 }
242 242
  243 + int nb_notify = _srs_config->get_vhost_hls_nb_notify(req->vhost);
243 for (int i = 0; i < (int)on_hls->args.size(); i++) { 244 for (int i = 0; i < (int)on_hls->args.size(); i++) {
244 std::string url = on_hls->args.at(i); 245 std::string url = on_hls->args.at(i);
245 - if ((ret = SrsHttpHooks::on_hls_notify(url, req, ts_url)) != ERROR_SUCCESS) {  
246 - srs_error("hook client on_hls_notify failed. url=%s, ret=%d", url.c_str(), ret); 246 + if ((ret = SrsHttpHooks::on_hls_notify(url, req, ts_url, nb_notify)) != ERROR_SUCCESS) {
  247 + srs_error("hook client on_hls_notify failed. url=%s, ts=%s, ret=%d", url.c_str(), ts_url.c_str(), ret);
247 return ret; 248 return ret;
248 } 249 }
249 } 250 }
@@ -53,6 +53,9 @@ using namespace std; @@ -53,6 +53,9 @@ using namespace std;
53 #define SRS_CONSTS_HTTP_PUT HTTP_PUT 53 #define SRS_CONSTS_HTTP_PUT HTTP_PUT
54 #define SRS_CONSTS_HTTP_DELETE HTTP_DELETE 54 #define SRS_CONSTS_HTTP_DELETE HTTP_DELETE
55 55
  56 +// for ead all of http body, read each time.
  57 +#define SRS_HTTP_READ_CACHE_BYTES 4096
  58 +
56 #define SRS_HTTP_DEFAULT_PAGE "index.html" 59 #define SRS_HTTP_DEFAULT_PAGE "index.html"
57 60
58 int srs_go_http_response_json(ISrsHttpResponseWriter* w, string data) 61 int srs_go_http_response_json(ISrsHttpResponseWriter* w, string data)
@@ -889,7 +892,8 @@ SrsHttpResponseReader::SrsHttpResponseReader(SrsHttpMessage* msg, SrsStSocket* i @@ -889,7 +892,8 @@ SrsHttpResponseReader::SrsHttpResponseReader(SrsHttpMessage* msg, SrsStSocket* i
889 skt = io; 892 skt = io;
890 owner = msg; 893 owner = msg;
891 is_eof = false; 894 is_eof = false;
892 - nb_read = 0; 895 + nb_total_read = 0;
  896 + nb_left_chunk = 0;
893 buffer = NULL; 897 buffer = NULL;
894 } 898 }
895 899
@@ -901,6 +905,8 @@ int SrsHttpResponseReader::initialize(SrsFastBuffer* body) @@ -901,6 +905,8 @@ int SrsHttpResponseReader::initialize(SrsFastBuffer* body)
901 { 905 {
902 int ret = ERROR_SUCCESS; 906 int ret = ERROR_SUCCESS;
903 907
  908 + nb_left_chunk = 0;
  909 + nb_total_read = 0;
904 buffer = body; 910 buffer = body;
905 911
906 return ret; 912 return ret;
@@ -911,7 +917,7 @@ bool SrsHttpResponseReader::eof() @@ -911,7 +917,7 @@ bool SrsHttpResponseReader::eof()
911 return is_eof; 917 return is_eof;
912 } 918 }
913 919
914 -int SrsHttpResponseReader::read(std::string& data) 920 +int SrsHttpResponseReader::read(char* data, int nb_data, int* nb_read)
915 { 921 {
916 int ret = ERROR_SUCCESS; 922 int ret = ERROR_SUCCESS;
917 923
@@ -923,95 +929,115 @@ int SrsHttpResponseReader::read(std::string& data) @@ -923,95 +929,115 @@ int SrsHttpResponseReader::read(std::string& data)
923 929
924 // chunked encoding. 930 // chunked encoding.
925 if (owner->is_chunked()) { 931 if (owner->is_chunked()) {
926 - return read_chunked(data); 932 + return read_chunked(data, nb_data, nb_read);
927 } 933 }
928 934
929 // read by specified content-length 935 // read by specified content-length
930 - int max = (int)owner->content_length() - (int)nb_read; 936 + int max = (int)owner->content_length() - (int)nb_total_read;
931 if (max <= 0) { 937 if (max <= 0) {
932 is_eof = true; 938 is_eof = true;
933 return ret; 939 return ret;
934 } 940 }
935 - return read_specified(max, data); 941 +
  942 + // change the max to read.
  943 + nb_data = srs_min(nb_data, max);
  944 + return read_specified(data, nb_data, nb_read);
936 } 945 }
937 946
938 -int SrsHttpResponseReader::read_chunked(std::string& data) 947 +int SrsHttpResponseReader::read_chunked(char* data, int nb_data, int* nb_read)
939 { 948 {
940 int ret = ERROR_SUCCESS; 949 int ret = ERROR_SUCCESS;
941 950
  951 + // when no bytes left in chunk,
942 // parse the chunk length first. 952 // parse the chunk length first.
943 - char* at = NULL;  
944 - int length = 0;  
945 - while (!at) {  
946 - // find the CRLF of chunk header end.  
947 - char* start = buffer->bytes();  
948 - char* end = start + buffer->size();  
949 - for (char* p = start; p < end - 1; p++) {  
950 - if (p[0] == SRS_HTTP_CR && p[1] == SRS_HTTP_LF) {  
951 - // invalid chunk, ignore.  
952 - if (p == start) {  
953 - ret = ERROR_HTTP_INVALID_CHUNK_HEADER;  
954 - srs_error("chunk header start with CRLF. ret=%d", ret);  
955 - return ret; 953 + if (nb_left_chunk <= 0) {
  954 + char* at = NULL;
  955 + int length = 0;
  956 + while (!at) {
  957 + // find the CRLF of chunk header end.
  958 + char* start = buffer->bytes();
  959 + char* end = start + buffer->size();
  960 + for (char* p = start; p < end - 1; p++) {
  961 + if (p[0] == SRS_HTTP_CR && p[1] == SRS_HTTP_LF) {
  962 + // invalid chunk, ignore.
  963 + if (p == start) {
  964 + ret = ERROR_HTTP_INVALID_CHUNK_HEADER;
  965 + srs_error("chunk header start with CRLF. ret=%d", ret);
  966 + return ret;
  967 + }
  968 + length = (int)(p - start + 2);
  969 + at = buffer->read_slice(length);
  970 + break;
956 } 971 }
957 - length = (int)(p - start + 2);  
958 - at = buffer->read_slice(length); 972 + }
  973 +
  974 + // got at, ok.
  975 + if (at) {
959 break; 976 break;
960 } 977 }
  978 +
  979 + // when empty, only grow 1bytes, but the buffer will cache more.
  980 + if ((ret = buffer->grow(skt, buffer->size() + 1)) != ERROR_SUCCESS) {
  981 + if (!srs_is_client_gracefully_close(ret)) {
  982 + srs_error("read body from server failed. ret=%d", ret);
  983 + }
  984 + return ret;
  985 + }
961 } 986 }
  987 + srs_assert(length >= 3);
962 988
963 - // got at, ok.  
964 - if (at) {  
965 - break;  
966 - } 989 + // it's ok to set the pos and pos+1 to NULL.
  990 + at[length - 1] = 0;
  991 + at[length - 2] = 0;
967 992
968 - // when empty, only grow 1bytes, but the buffer will cache more.  
969 - if ((ret = buffer->grow(skt, buffer->size() + 1)) != ERROR_SUCCESS) {  
970 - if (!srs_is_client_gracefully_close(ret)) {  
971 - srs_error("read body from server failed. ret=%d", ret);  
972 - } 993 + // size is the bytes size, excludes the chunk header and end CRLF.
  994 + int ilength = (int)::strtol(at, NULL, 16);
  995 + if (ilength < 0) {
  996 + ret = ERROR_HTTP_INVALID_CHUNK_HEADER;
  997 + srs_error("chunk header negative, length=%d. ret=%d", ilength, ret);
973 return ret; 998 return ret;
974 } 999 }
  1000 +
  1001 + // all bytes in chunk is left now.
  1002 + nb_left_chunk = ilength;
975 } 1003 }
976 - srs_assert(length >= 3);  
977 1004
978 - // it's ok to set the pos and pos+1 to NULL.  
979 - at[length - 1] = 0;  
980 - at[length - 2] = 0; 1005 + // left bytes in chunk, read some.
  1006 + srs_assert(nb_left_chunk);
981 1007
982 - // size is the bytes size, excludes the chunk header and end CRLF.  
983 - int ilength = (int)::strtol(at, NULL, 16);  
984 - if (ilength < 0) {  
985 - ret = ERROR_HTTP_INVALID_CHUNK_HEADER;  
986 - srs_error("chunk header negative, length=%d. ret=%d", ilength, ret);  
987 - return ret; 1008 + int nb_bytes = srs_min(nb_left_chunk, nb_data);
  1009 + ret = read_specified(data, nb_bytes, &nb_bytes);
  1010 +
  1011 + // the nb_bytes used for output already read size of bytes.
  1012 + if (nb_read) {
  1013 + *nb_read = nb_bytes;
988 } 1014 }
  1015 + nb_left_chunk -= nb_bytes;
989 1016
990 - // when empty, only grow 1bytes, but the buffer will cache more.  
991 - if ((ret = buffer->grow(skt, ilength + 2)) != ERROR_SUCCESS) {  
992 - if (!srs_is_client_gracefully_close(ret)) {  
993 - srs_error("read body from server failed. ret=%d", ret);  
994 - } 1017 + // error or still left bytes in chunk, ignore and read in future.
  1018 + if (nb_left_chunk > 0 || (ret != ERROR_SUCCESS)) {
995 return ret; 1019 return ret;
996 } 1020 }
997 - srs_info("http: read %d chunk", ilength); 1021 + srs_info("http: read %d bytes of chunk", nb_bytes);
998 1022
999 // read payload when length specifies some payload. 1023 // read payload when length specifies some payload.
1000 - if (ilength <= 0) { 1024 + if (nb_left_chunk <= 0) {
1001 is_eof = true; 1025 is_eof = true;
1002 - } else {  
1003 - srs_assert(ilength);  
1004 - data.append(buffer->read_slice(ilength), ilength);  
1005 - nb_read += ilength;  
1006 } 1026 }
1007 1027
1008 // the CRLF of chunk payload end. 1028 // the CRLF of chunk payload end.
  1029 + if ((ret = buffer->grow(skt, 2)) != ERROR_SUCCESS) {
  1030 + if (!srs_is_client_gracefully_close(ret)) {
  1031 + srs_error("read EOF of chunk from server failed. ret=%d", ret);
  1032 + }
  1033 + return ret;
  1034 + }
1009 buffer->read_slice(2); 1035 buffer->read_slice(2);
1010 1036
1011 return ret; 1037 return ret;
1012 } 1038 }
1013 1039
1014 -int SrsHttpResponseReader::read_specified(int max, std::string& data) 1040 +int SrsHttpResponseReader::read_specified(char* data, int nb_data, int* nb_read)
1015 { 1041 {
1016 int ret = ERROR_SUCCESS; 1042 int ret = ERROR_SUCCESS;
1017 1043
@@ -1025,14 +1051,21 @@ int SrsHttpResponseReader::read_specified(int max, std::string& data) @@ -1025,14 +1051,21 @@ int SrsHttpResponseReader::read_specified(int max, std::string& data)
1025 } 1051 }
1026 } 1052 }
1027 1053
1028 - int nb_bytes = srs_min(max, buffer->size()); 1054 + int nb_bytes = srs_min(nb_data, buffer->size());
1029 1055
  1056 + // read data to buffer.
1030 srs_assert(nb_bytes); 1057 srs_assert(nb_bytes);
1031 - data.append(buffer->read_slice(nb_bytes), nb_bytes);  
1032 - nb_read += nb_bytes; 1058 + char* p = buffer->read_slice(nb_bytes);
  1059 + memcpy(data, p, nb_bytes);
  1060 + if (nb_read) {
  1061 + *nb_read = nb_bytes;
  1062 + }
  1063 +
  1064 + // increase the total read to determine whether EOF.
  1065 + nb_total_read += nb_bytes;
1033 1066
1034 // when read completed, eof. 1067 // when read completed, eof.
1035 - if (nb_read >= (int)owner->content_length()) { 1068 + if (nb_total_read >= (int)owner->content_length()) {
1036 is_eof = true; 1069 is_eof = true;
1037 } 1070 }
1038 1071
@@ -1223,11 +1256,20 @@ int SrsHttpMessage::body_read_all(string& body) @@ -1223,11 +1256,20 @@ int SrsHttpMessage::body_read_all(string& body)
1223 { 1256 {
1224 int ret = ERROR_SUCCESS; 1257 int ret = ERROR_SUCCESS;
1225 1258
  1259 + // cache to read.
  1260 + char* buf = new char[SRS_HTTP_READ_CACHE_BYTES];
  1261 + SrsAutoFree(char, buf);
  1262 +
1226 // whatever, read util EOF. 1263 // whatever, read util EOF.
1227 while (!_body->eof()) { 1264 while (!_body->eof()) {
1228 - if ((ret = _body->read(body)) != ERROR_SUCCESS) { 1265 + int nb_read = 0;
  1266 + if ((ret = _body->read(buf, SRS_HTTP_READ_CACHE_BYTES, &nb_read)) != ERROR_SUCCESS) {
1229 return ret; 1267 return ret;
1230 } 1268 }
  1269 +
  1270 + if (nb_read > 0) {
  1271 + body.append(buf, nb_read);
  1272 + }
1231 } 1273 }
1232 1274
1233 return ret; 1275 return ret;
@@ -196,10 +196,13 @@ public: @@ -196,10 +196,13 @@ public:
196 */ 196 */
197 virtual bool eof() = 0; 197 virtual bool eof() = 0;
198 /** 198 /**
199 - * read from the response body.  
200 - * @remark when eof(), return error.  
201 - */  
202 - virtual int read(std::string& data) = 0; 199 + * read from the response body.
  200 + * @param data, the buffer to read data buffer to.
  201 + * @param nb_data, the max size of data buffer.
  202 + * @param nb_read, the actual read size of bytes. NULL to ignore.
  203 + * @remark when eof(), return error.
  204 + */
  205 + virtual int read(char* data, int nb_data, int* nb_read) = 0;
203 }; 206 };
204 207
205 // Objects implementing the Handler interface can be 208 // Objects implementing the Handler interface can be
@@ -431,7 +434,10 @@ private: @@ -431,7 +434,10 @@ private:
431 SrsHttpMessage* owner; 434 SrsHttpMessage* owner;
432 SrsFastBuffer* buffer; 435 SrsFastBuffer* buffer;
433 bool is_eof; 436 bool is_eof;
434 - int64_t nb_read; 437 + // the left bytes in chunk.
  438 + int nb_left_chunk;
  439 + // already read total bytes.
  440 + int64_t nb_total_read;
435 public: 441 public:
436 SrsHttpResponseReader(SrsHttpMessage* msg, SrsStSocket* io); 442 SrsHttpResponseReader(SrsHttpMessage* msg, SrsStSocket* io);
437 virtual ~SrsHttpResponseReader(); 443 virtual ~SrsHttpResponseReader();
@@ -443,10 +449,10 @@ public: @@ -443,10 +449,10 @@ public:
443 // interface ISrsHttpResponseReader 449 // interface ISrsHttpResponseReader
444 public: 450 public:
445 virtual bool eof(); 451 virtual bool eof();
446 - virtual int read(std::string& data); 452 + virtual int read(char* data, int nb_data, int* nb_read);
447 private: 453 private:
448 - virtual int read_chunked(std::string& data);  
449 - virtual int read_specified(int max, std::string& data); 454 + virtual int read_chunked(char* data, int nb_data, int* nb_read);
  455 + virtual int read_specified(char* data, int nb_data, int* nb_read);
450 }; 456 };
451 457
452 // for http header. 458 // for http header.
@@ -37,15 +37,13 @@ using namespace std; @@ -37,15 +37,13 @@ using namespace std;
37 #include <srs_app_utility.hpp> 37 #include <srs_app_utility.hpp>
38 #include <srs_core_autofree.hpp> 38 #include <srs_core_autofree.hpp>
39 39
40 -// when error, http client sleep for a while and retry.  
41 -#define SRS_HTTP_CLIENT_SLEEP_US (int64_t)(3*1000*1000LL)  
42 -  
43 SrsHttpClient::SrsHttpClient() 40 SrsHttpClient::SrsHttpClient()
44 { 41 {
45 connected = false; 42 connected = false;
46 stfd = NULL; 43 stfd = NULL;
47 skt = NULL; 44 skt = NULL;
48 parser = NULL; 45 parser = NULL;
  46 + timeout_us = 0;
49 } 47 }
50 48
51 SrsHttpClient::~SrsHttpClient() 49 SrsHttpClient::~SrsHttpClient()
@@ -54,7 +52,7 @@ SrsHttpClient::~SrsHttpClient() @@ -54,7 +52,7 @@ SrsHttpClient::~SrsHttpClient()
54 srs_freep(parser); 52 srs_freep(parser);
55 } 53 }
56 54
57 -int SrsHttpClient::initialize(string h, int p) 55 +int SrsHttpClient::initialize(string h, int p, int64_t t_us)
58 { 56 {
59 int ret = ERROR_SUCCESS; 57 int ret = ERROR_SUCCESS;
60 58
@@ -68,6 +66,7 @@ int SrsHttpClient::initialize(string h, int p) @@ -68,6 +66,7 @@ int SrsHttpClient::initialize(string h, int p)
68 66
69 host = h; 67 host = h;
70 port = p; 68 port = p;
  69 + timeout_us = t_us;
71 70
72 return ret; 71 return ret;
73 } 72 }
@@ -183,10 +182,9 @@ int SrsHttpClient::connect() @@ -183,10 +182,9 @@ int SrsHttpClient::connect()
183 disconnect(); 182 disconnect();
184 183
185 // open socket. 184 // open socket.
186 - int64_t timeout = SRS_HTTP_CLIENT_SLEEP_US;  
187 - if ((ret = srs_socket_connect(host, port, timeout, &stfd)) != ERROR_SUCCESS) { 185 + if ((ret = srs_socket_connect(host, port, timeout_us, &stfd)) != ERROR_SUCCESS) {
188 srs_warn("http client failed, server=%s, port=%d, timeout=%"PRId64", ret=%d", 186 srs_warn("http client failed, server=%s, port=%d, timeout=%"PRId64", ret=%d",
189 - host.c_str(), port, timeout, ret); 187 + host.c_str(), port, timeout_us, ret);
190 return ret; 188 return ret;
191 } 189 }
192 srs_info("connect to server success. server=%s, port=%d", host, port); 190 srs_info("connect to server success. server=%s, port=%d", host, port);
@@ -195,6 +193,10 @@ int SrsHttpClient::connect() @@ -195,6 +193,10 @@ int SrsHttpClient::connect()
195 skt = new SrsStSocket(stfd); 193 skt = new SrsStSocket(stfd);
196 connected = true; 194 connected = true;
197 195
  196 + // set the recv/send timeout in us.
  197 + skt->set_recv_timeout(timeout_us);
  198 + skt->set_send_timeout(timeout_us);
  199 +
198 return ret; 200 return ret;
199 } 201 }
200 202
@@ -40,6 +40,9 @@ class SrsHttpParser; @@ -40,6 +40,9 @@ class SrsHttpParser;
40 class SrsHttpMessage; 40 class SrsHttpMessage;
41 class SrsStSocket; 41 class SrsStSocket;
42 42
  43 +// the default timeout for http client.
  44 +#define SRS_HTTP_CLIENT_TIMEOUT_US (int64_t)(30*1000*1000LL)
  45 +
43 /** 46 /**
44 * http client to GET/POST/PUT/DELETE uri 47 * http client to GET/POST/PUT/DELETE uri
45 */ 48 */
@@ -51,6 +54,7 @@ private: @@ -51,6 +54,7 @@ private:
51 SrsStSocket* skt; 54 SrsStSocket* skt;
52 SrsHttpParser* parser; 55 SrsHttpParser* parser;
53 private: 56 private:
  57 + int64_t timeout_us;
54 // host name or ip. 58 // host name or ip.
55 std::string host; 59 std::string host;
56 int port; 60 int port;
@@ -61,7 +65,7 @@ public: @@ -61,7 +65,7 @@ public:
61 /** 65 /**
62 * initialize the client, connect to host and port. 66 * initialize the client, connect to host and port.
63 */ 67 */
64 - virtual int initialize(std::string h, int p); 68 + virtual int initialize(std::string h, int p, int64_t t_us = SRS_HTTP_CLIENT_TIMEOUT_US);
65 public: 69 public:
66 /** 70 /**
67 * to post data to the uri. 71 * to post data to the uri.
@@ -42,8 +42,12 @@ using namespace std; @@ -42,8 +42,12 @@ using namespace std;
42 #define SRS_HTTP_RESPONSE_OK SRS_XSTR(ERROR_SUCCESS) 42 #define SRS_HTTP_RESPONSE_OK SRS_XSTR(ERROR_SUCCESS)
43 43
44 #define SRS_HTTP_HEADER_BUFFER 1024 44 #define SRS_HTTP_HEADER_BUFFER 1024
  45 +#define SRS_HTTP_READ_BUFFER 4096
45 #define SRS_HTTP_BODY_BUFFER 32 * 1024 46 #define SRS_HTTP_BODY_BUFFER 32 * 1024
46 47
  48 +// the timeout for hls notify, in us.
  49 +#define SRS_HLS_NOTIFY_TIMEOUT_US (int64_t)(10*1000*1000LL)
  50 +
47 SrsHttpHooks::SrsHttpHooks() 51 SrsHttpHooks::SrsHttpHooks()
48 { 52 {
49 } 53 }
@@ -326,7 +330,7 @@ int SrsHttpHooks::on_hls(string url, SrsRequest* req, string file, int sn, doubl @@ -326,7 +330,7 @@ int SrsHttpHooks::on_hls(string url, SrsRequest* req, string file, int sn, doubl
326 return ret; 330 return ret;
327 } 331 }
328 332
329 -int SrsHttpHooks::on_hls_notify(std::string url, SrsRequest* req, std::string ts_url) 333 +int SrsHttpHooks::on_hls_notify(std::string url, SrsRequest* req, std::string ts_url, int nb_notify)
330 { 334 {
331 int ret = ERROR_SUCCESS; 335 int ret = ERROR_SUCCESS;
332 336
@@ -341,6 +345,8 @@ int SrsHttpHooks::on_hls_notify(std::string url, SrsRequest* req, std::string ts @@ -341,6 +345,8 @@ int SrsHttpHooks::on_hls_notify(std::string url, SrsRequest* req, std::string ts
341 url = srs_string_replace(url, "[stream]", req->stream); 345 url = srs_string_replace(url, "[stream]", req->stream);
342 url = srs_string_replace(url, "[ts_url]", ts_url); 346 url = srs_string_replace(url, "[ts_url]", ts_url);
343 347
  348 + int64_t starttime = srs_update_system_time_ms();
  349 +
344 SrsHttpUri uri; 350 SrsHttpUri uri;
345 if ((ret = uri.initialize(url)) != ERROR_SUCCESS) { 351 if ((ret = uri.initialize(url)) != ERROR_SUCCESS) {
346 srs_error("http: post failed. url=%s, ret=%d", url.c_str(), ret); 352 srs_error("http: post failed. url=%s, ret=%d", url.c_str(), ret);
@@ -348,7 +354,7 @@ int SrsHttpHooks::on_hls_notify(std::string url, SrsRequest* req, std::string ts @@ -348,7 +354,7 @@ int SrsHttpHooks::on_hls_notify(std::string url, SrsRequest* req, std::string ts
348 } 354 }
349 355
350 SrsHttpClient http; 356 SrsHttpClient http;
351 - if ((ret = http.initialize(uri.get_host(), uri.get_port())) != ERROR_SUCCESS) { 357 + if ((ret = http.initialize(uri.get_host(), uri.get_port(), SRS_HLS_NOTIFY_TIMEOUT_US)) != ERROR_SUCCESS) {
352 return ret; 358 return ret;
353 } 359 }
354 360
@@ -358,16 +364,23 @@ int SrsHttpHooks::on_hls_notify(std::string url, SrsRequest* req, std::string ts @@ -358,16 +364,23 @@ int SrsHttpHooks::on_hls_notify(std::string url, SrsRequest* req, std::string ts
358 } 364 }
359 SrsAutoFree(SrsHttpMessage, msg); 365 SrsAutoFree(SrsHttpMessage, msg);
360 366
  367 + int nb_buf = srs_min(nb_notify, SRS_HTTP_READ_BUFFER);
  368 + char* buf = new char[nb_buf];
  369 + SrsAutoFree(char, buf);
  370 +
  371 + int nb_read = 0;
361 ISrsHttpResponseReader* br = msg->body_reader(); 372 ISrsHttpResponseReader* br = msg->body_reader();
362 - while (!br->eof()) {  
363 - std::string data;  
364 - if ((ret = br->read(data)) != ERROR_SUCCESS) { 373 + while (nb_read < nb_notify && !br->eof()) {
  374 + int nb_bytes = 0;
  375 + if ((ret = br->read(buf, nb_buf, &nb_bytes)) != ERROR_SUCCESS) {
365 break; 376 break;
366 } 377 }
  378 + nb_read += nb_bytes;
367 } 379 }
368 380
369 - srs_trace("http hook on_hls_notify success. client_id=%d, url=%s, code=%d, ret=%d",  
370 - client_id, url.c_str(), msg->status_code(), ret); 381 + int spenttime = (int)(srs_update_system_time_ms() - starttime);
  382 + srs_trace("http hook on_hls_notify success. client_id=%d, url=%s, code=%d, spent=%dms, read=%dB, ret=%d",
  383 + client_id, url.c_str(), msg->status_code(), spenttime, nb_read, ret);
371 384
372 // ignore any error for on_hls_notify. 385 // ignore any error for on_hls_notify.
373 ret = ERROR_SUCCESS; 386 ret = ERROR_SUCCESS;
@@ -110,8 +110,9 @@ public: @@ -110,8 +110,9 @@ public:
110 * @param url the api server url, to process the event. 110 * @param url the api server url, to process the event.
111 * ignore if empty. 111 * ignore if empty.
112 * @param ts_url the ts uri, used to replace the variable [ts_url] in url. 112 * @param ts_url the ts uri, used to replace the variable [ts_url] in url.
  113 + * @param nb_notify the max bytes to read from notify server.
113 */ 114 */
114 - static int on_hls_notify(std::string url, SrsRequest* req, std::string ts_url); 115 + static int on_hls_notify(std::string url, SrsRequest* req, std::string ts_url, int nb_notify);
115 private: 116 private:
116 static int do_post(std::string url, std::string req, int& code, std::string& res); 117 static int do_post(std::string url, std::string req, int& code, std::string& res);
117 }; 118 };
@@ -134,7 +134,7 @@ int64_t srs_update_system_time_ms() @@ -134,7 +134,7 @@ int64_t srs_update_system_time_ms()
134 if (_srs_system_time_us_cache <= 0) { 134 if (_srs_system_time_us_cache <= 0) {
135 _srs_system_time_us_cache = now_us; 135 _srs_system_time_us_cache = now_us;
136 _srs_system_time_startup_time = now_us; 136 _srs_system_time_startup_time = now_us;
137 - return _srs_system_time_us_cache; 137 + return _srs_system_time_us_cache / 1000;
138 } 138 }
139 139
140 // use relative time. 140 // use relative time.
@@ -151,7 +151,7 @@ int64_t srs_update_system_time_ms() @@ -151,7 +151,7 @@ int64_t srs_update_system_time_ms()
151 srs_info("system time updated, startup=%"PRId64"us, now=%"PRId64"us", 151 srs_info("system time updated, startup=%"PRId64"us, now=%"PRId64"us",
152 _srs_system_time_startup_time, _srs_system_time_us_cache); 152 _srs_system_time_startup_time, _srs_system_time_us_cache);
153 153
154 - return _srs_system_time_us_cache; 154 + return _srs_system_time_us_cache / 1000;
155 } 155 }
156 156
157 string srs_dns_resolve(string host) 157 string srs_dns_resolve(string host)
@@ -390,7 +390,7 @@ bool srs_aac_startswith_adts(SrsStream* stream) @@ -390,7 +390,7 @@ bool srs_aac_startswith_adts(SrsStream* stream)
390 char* bytes = stream->data() + stream->pos(); 390 char* bytes = stream->data() + stream->pos();
391 char* p = bytes; 391 char* p = bytes;
392 392
393 - if (!stream->require(p - bytes + 2)) { 393 + if (!stream->require((int)(p - bytes) + 2)) {
394 return false; 394 return false;
395 } 395 }
396 396