Blame view

trunk/src/app/srs_app_rtmp_conn.cpp 39.7 KB
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26
/*
The MIT License (MIT)

Copyright (c) 2013-2014 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_app_rtmp_conn.hpp>

#include <stdlib.h>
27 28 29
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
30 31 32 33 34 35 36 37 38 39 40 41 42 43 44

using namespace std;

#include <srs_kernel_error.hpp>
#include <srs_kernel_log.hpp>
#include <srs_protocol_rtmp.hpp>
#include <srs_core_autofree.hpp>
#include <srs_app_source.hpp>
#include <srs_app_server.hpp>
#include <srs_app_pithy_print.hpp>
#include <srs_app_config.hpp>
#include <srs_app_refer.hpp>
#include <srs_app_hls.hpp>
#include <srs_app_http.hpp>
#include <srs_app_bandwidth.hpp>
45
#include <srs_app_st_socket.hpp>
46
#include <srs_app_http_hooks.hpp>
47
#include <srs_app_edge.hpp>
48
#include <srs_app_utility.hpp>
49
#include <srs_protocol_msg_array.hpp>
50
#include <srs_protocol_amf0.hpp>
51
#include <srs_app_recv_thread.hpp>
52
winlin authored
53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69
// when stream is busy, for example, streaming is already
// publishing, when a new client to request to publish,
// sleep a while and close the connection.
#define SRS_STREAM_BUSY_SLEEP_US (int64_t)(3*1000*1000LL)

// the timeout to wait encoder to republish
// if timeout, close the connection.
#define SRS_REPUBLISH_SEND_TIMEOUT_US (int64_t)(3*60*1000*1000LL)
// if timeout, close the connection.
#define SRS_REPUBLISH_RECV_TIMEOUT_US (int64_t)(3*60*1000*1000LL)

// the timeout to wait client data, when client paused
// if timeout, close the connection.
#define SRS_PAUSED_SEND_TIMEOUT_US (int64_t)(30*60*1000*1000LL)
// if timeout, close the connection.
#define SRS_PAUSED_RECV_TIMEOUT_US (int64_t)(30*60*1000*1000LL)
70 71 72
// when edge timeout, retry next.
#define SRS_EDGE_TOKEN_TRAVERSE_TIMEOUT_US (int64_t)(3*1000*1000LL)
73
SrsRtmpConn::SrsRtmpConn(SrsServer* srs_server, st_netfd_t client_stfd)
74 75 76 77
    : SrsConnection(srs_server, client_stfd)
{
    req = new SrsRequest();
    res = new SrsResponse();
78
    skt = new SrsStSocket(client_stfd);
79 80 81
    rtmp = new SrsRtmpServer(skt);
    refer = new SrsRefer();
    bandwidth = new SrsBandwidth();
82
    duration = 0;
83 84
    kbps = new SrsKbps();
    kbps->set_io(skt, skt);
85 86 87 88
    
    _srs_config->subscribe(this);
}
89
SrsRtmpConn::~SrsRtmpConn()
90 91 92 93 94 95 96 97 98
{
    _srs_config->unsubscribe(this);
    
    srs_freep(req);
    srs_freep(res);
    srs_freep(rtmp);
    srs_freep(skt);
    srs_freep(refer);
    srs_freep(bandwidth);
99
    srs_freep(kbps);
100 101
}
102 103 104 105 106
void SrsRtmpConn::kbps_resample()
{
    kbps->sample();
}
107
// TODO: return detail message when error for client.
108
int SrsRtmpConn::do_cycle()
109 110 111
{
    int ret = ERROR_SUCCESS;
    
winlin authored
112
    srs_trace("RTMP client ip=%s", ip.c_str());
113
winlin authored
114 115
    rtmp->set_recv_timeout(SRS_CONSTS_RTMP_RECV_TIMEOUT_US);
    rtmp->set_send_timeout(SRS_CONSTS_RTMP_SEND_TIMEOUT_US);
116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152
    
    if ((ret = rtmp->handshake()) != ERROR_SUCCESS) {
        srs_error("rtmp handshake failed. ret=%d", ret);
        return ret;
    }
    srs_verbose("rtmp handshake success");
    
    if ((ret = rtmp->connect_app(req)) != ERROR_SUCCESS) {
        srs_error("rtmp connect vhost/app failed. ret=%d", ret);
        return ret;
    }
    srs_verbose("rtmp connect app success");
    
    // discovery vhost, resolve the vhost from config
    SrsConfDirective* parsed_vhost = _srs_config->get_vhost(req->vhost);
    if (parsed_vhost) {
        req->vhost = parsed_vhost->arg0();
    }
    
    srs_info("discovery app success. schema=%s, vhost=%s, port=%s, app=%s",
        req->schema.c_str(), req->vhost.c_str(), req->port.c_str(), req->app.c_str());
    
    if (req->schema.empty() || req->vhost.empty() || req->port.empty() || req->app.empty()) {
        ret = ERROR_RTMP_REQ_TCURL;
        srs_error("discovery tcUrl failed. "
            "tcUrl=%s, schema=%s, vhost=%s, port=%s, app=%s, ret=%d",
            req->tcUrl.c_str(), req->schema.c_str(), req->vhost.c_str(), req->port.c_str(), req->app.c_str(), ret);
        return ret;
    }
    
    // check vhost
    if ((ret = check_vhost()) != ERROR_SUCCESS) {
        srs_error("check vhost failed. ret=%d", ret);
        return ret;
    }
    srs_verbose("check vhost success.");
    
winlin authored
153
    srs_trace("connect app, "
154
        "tcUrl=%s, pageUrl=%s, swfUrl=%s, schema=%s, vhost=%s, port=%s, app=%s, args=%s", 
155 156
        req->tcUrl.c_str(), req->pageUrl.c_str(), req->swfUrl.c_str(), 
        req->schema.c_str(), req->vhost.c_str(), req->port.c_str(),
157
        req->app.c_str(), (req->args? "(obj)":"null"));
158
    
159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187
    // show client identity
    if(req->args) {
        std::string srs_version;
        std::string srs_server_ip;
        int srs_pid = 0;
        int srs_id = 0;
        
        SrsAmf0Any* prop = NULL;
        if ((prop = req->args->ensure_property_string("srs_version")) != NULL) {
            srs_version = prop->to_str();
        }
        if ((prop = req->args->ensure_property_string("srs_server_ip")) != NULL) {
            srs_server_ip = prop->to_str();
        }
        if ((prop = req->args->ensure_property_number("srs_pid")) != NULL) {
            srs_pid = (int)prop->to_number();
        }
        if ((prop = req->args->ensure_property_number("srs_id")) != NULL) {
            srs_id = (int)prop->to_number();
        }
        
        srs_info("edge-srs ip=%s, version=%s, pid=%d, id=%d", 
            srs_server_ip.c_str(), srs_version.c_str(), srs_pid, srs_id);
        if (srs_pid > 0) {
            srs_trace("edge-srs ip=%s, version=%s, pid=%d, id=%d", 
                srs_server_ip.c_str(), srs_version.c_str(), srs_pid, srs_id);
        }
    }
    
188
    ret = service_cycle();
189
    http_hooks_on_close();
190 191 192 193
    
    return ret;
}
194
int SrsRtmpConn::on_reload_vhost_removed(string vhost)
195 196 197 198 199 200 201 202 203 204 205 206 207 208 209
{
    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;
}
210 211 212 213 214 215 216 217 218 219

int64_t SrsRtmpConn::get_send_bytes_delta()
{
    return kbps->get_send_bytes_delta();
}

int64_t SrsRtmpConn::get_recv_bytes_delta()
{
    return kbps->get_recv_bytes_delta();
}
220
    
221
int SrsRtmpConn::service_cycle()
222 223 224
{    
    int ret = ERROR_SUCCESS;
    
225
    if ((ret = rtmp->set_window_ack_size((int)(2.5 * 1000 * 1000))) != ERROR_SUCCESS) {
226 227 228 229 230
        srs_error("set window acknowledgement size failed. ret=%d", ret);
        return ret;
    }
    srs_verbose("set window acknowledgement size success");
        
231
    if ((ret = rtmp->set_peer_bandwidth((int)(2.5 * 1000 * 1000), 2)) != ERROR_SUCCESS) {
232 233 234 235 236
        srs_error("set peer bandwidth failed. ret=%d", ret);
        return ret;
    }
    srs_verbose("set peer bandwidth success");
237 238 239
    // get the ip which client connected.
    std::string local_ip = srs_get_local_ip(st_netfd_fileno(stfd));
    
240 241
    // do bandwidth test if connect to the vhost which is for bandwidth check.
    if (_srs_config->get_bw_check_enabled(req->vhost)) {
242
        return bandwidth->bandwidth_check(rtmp, skt, req, local_ip);
243 244
    }
    
tufang14 authored
245
    // do token traverse before serve it.
winlin authored
246
    // @see https://github.com/winlinvip/simple-rtmp-server/pull/239
tufang14 authored
247 248 249 250 251 252 253 254 255
    bool vhost_is_edge = _srs_config->get_vhost_is_edge(req->vhost);
    bool edge_traverse = _srs_config->get_vhost_edge_token_traverse(req->vhost);
    if (vhost_is_edge && edge_traverse) {
        if ((ret = check_edge_token_traverse_auth()) != ERROR_SUCCESS) {
            srs_warn("token auth failed, ret=%d", ret);
            return ret;
        }
    }
    
winlin authored
256
    // response the client connect ok.
257
    if ((ret = rtmp->response_connect_app(req, local_ip.c_str())) != ERROR_SUCCESS) {
258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295
        srs_error("response connect app failed. ret=%d", ret);
        return ret;
    }
    srs_verbose("response connect app success");
        
    if ((ret = rtmp->on_bw_done()) != ERROR_SUCCESS) {
        srs_error("on_bw_done failed. ret=%d", ret);
        return ret;
    }
    srs_verbose("on_bw_done success");
    
    while (true) {
        ret = stream_service_cycle();
        
        // stream service must terminated with error, never success.
        srs_assert(ret != ERROR_SUCCESS);
        
        // when not system control error, fatal error, return.
        if (!srs_is_system_control_error(ret)) {
            if (ret != ERROR_SOCKET_TIMEOUT && !srs_is_client_gracefully_close(ret)) {
                srs_error("stream service cycle failed. ret=%d", ret);
            }
            return ret;
        }
        
        // for republish, continue service
        if (ret == ERROR_CONTROL_REPUBLISH) {
            // set timeout to a larger value, wait for encoder to republish.
            rtmp->set_send_timeout(SRS_REPUBLISH_RECV_TIMEOUT_US);
            rtmp->set_recv_timeout(SRS_REPUBLISH_SEND_TIMEOUT_US);
            
            srs_trace("control message(unpublish) accept, retry stream service.");
            continue;
        }
        
        // for "some" system control error, 
        // logical accept and retry stream service.
        if (ret == ERROR_CONTROL_RTMP_CLOSE) {
winlin authored
296 297
            // TODO: FIXME: use ping message to anti-death of socket.
            // @see: https://github.com/winlinvip/simple-rtmp-server/issues/39
298 299 300 301 302 303 304 305 306 307 308 309 310 311
            // set timeout to a larger value, for user paused.
            rtmp->set_recv_timeout(SRS_PAUSED_RECV_TIMEOUT_US);
            rtmp->set_send_timeout(SRS_PAUSED_SEND_TIMEOUT_US);
            
            srs_trace("control message(close) accept, retry stream service.");
            continue;
        }
        
        // for other system control message, fatal error.
        srs_error("control message(%d) reject as error. ret=%d", ret, ret);
        return ret;
    }
}
312
int SrsRtmpConn::stream_service_cycle()
313 314 315
{
    int ret = ERROR_SUCCESS;
        
316
    SrsRtmpConnType type;
317
    if ((ret = rtmp->identify_client(res->stream_id, type, req->stream, req->duration)) != ERROR_SUCCESS) {
winlin authored
318 319 320
        if (!srs_is_client_gracefully_close(ret)) {
            srs_error("identify client failed. ret=%d", ret);
        }
321 322
        return ret;
    }
323
    req->strip();
winlin authored
324
    srs_trace("client identified, type=%s, stream_name=%s, duration=%.2f", 
325
        srs_client_type_string(type).c_str(), req->stream.c_str(), req->duration);
326 327

    // client is identified, set the timeout to service timeout.
winlin authored
328 329
    rtmp->set_recv_timeout(SRS_CONSTS_RTMP_RECV_TIMEOUT_US);
    rtmp->set_send_timeout(SRS_CONSTS_RTMP_SEND_TIMEOUT_US);
330 331 332 333 334 335 336
    
    // set chunk size to larger.
    int chunk_size = _srs_config->get_chunk_size(req->vhost);
    if ((ret = rtmp->set_chunk_size(chunk_size)) != ERROR_SUCCESS) {
        srs_error("set chunk_size=%d failed. ret=%d", chunk_size, ret);
        return ret;
    }
337
    srs_info("set chunk_size=%d success", chunk_size);
338
    
339 340
    bool vhost_is_edge = _srs_config->get_vhost_is_edge(req->vhost);
    
341
    // find a source to serve.
342 343 344 345
    SrsSource* source = NULL;
    if ((ret = SrsSource::find(req, &source)) != ERROR_SUCCESS) {
        return ret;
    }
346 347
    srs_assert(source != NULL);
    
348 349 350 351 352
    // check ASAP, to fail it faster if invalid.
    if (type != SrsRtmpConnPlay && !vhost_is_edge) {
        // check publish available
        // for edge, never check it, for edge use proxy mode.
        if (!source->can_publish()) {
353 354 355 356 357 358 359
            ret = ERROR_SYSTEM_STREAM_BUSY;
            srs_warn("stream %s is already publishing. ret=%d", 
                req->get_stream_url().c_str(), ret);
            // to delay request
            st_usleep(SRS_STREAM_BUSY_SLEEP_US);
            return ret;
        }
360 361 362
    }
    
    bool enabled_cache = _srs_config->get_gop_cache(req->vhost);
363 364 365
    srs_trace("source url=%s, ip=%s, cache=%d, is_edge=%d, source_id=%d[%d]", 
        req->get_stream_url().c_str(), ip.c_str(), enabled_cache, vhost_is_edge, 
        source->source_id(), source->source_id());
366 367 368
    source->set_cache(enabled_cache);
    
    switch (type) {
369
        case SrsRtmpConnPlay: {
370 371
            srs_verbose("start to play stream %s.", req->stream.c_str());
            
winlin authored
372
            if (vhost_is_edge) {
373
                // notice edge to start for the first client.
374 375
                if ((ret = source->on_edge_start_play()) != ERROR_SUCCESS) {
                    srs_error("notice edge start play stream failed. ret=%d", ret);
winlin authored
376 377 378 379
                    return ret;
                }
            }
            
380
            // response connection start play
381 382 383 384
            if ((ret = rtmp->start_play(res->stream_id)) != ERROR_SUCCESS) {
                srs_error("start to play stream failed. ret=%d", ret);
                return ret;
            }
385
            if ((ret = http_hooks_on_play()) != ERROR_SUCCESS) {
386 387 388
                srs_error("http hook on_play failed. ret=%d", ret);
                return ret;
            }
389
            
390 391
            srs_info("start to play stream %s success", req->stream.c_str());
            ret = playing(source);
392
            http_hooks_on_stop();
393
            
394 395
            return ret;
        }
396
        case SrsRtmpConnFMLEPublish: {
397 398
            srs_verbose("FMLE start to publish stream %s.", req->stream.c_str());
            
winlin authored
399 400 401 402 403 404 405
            if (vhost_is_edge) {
                if ((ret = source->on_edge_start_publish()) != ERROR_SUCCESS) {
                    srs_error("notice edge start publish stream failed. ret=%d", ret);
                    return ret;
                }
            }
            
406 407 408 409
            if ((ret = rtmp->start_fmle_publish(res->stream_id)) != ERROR_SUCCESS) {
                srs_error("start to publish stream failed. ret=%d", ret);
                return ret;
            }
410
            
411 412 413 414
            if (!vhost_is_edge) {
                if ((ret = source->acquire_publish()) != ERROR_SUCCESS) {
                    return ret;
                }
415
            }
416
            
417
            ret = fmle_publishing(source);
418 419 420
            
            if (!vhost_is_edge) {
                source->release_publish();
421
            }
422
            
423 424
            return ret;
        }
425
        case SrsRtmpConnFlashPublish: {
426 427
            srs_verbose("flash start to publish stream %s.", req->stream.c_str());
            
428 429 430 431 432 433 434
            if (vhost_is_edge) {
                if ((ret = source->on_edge_start_publish()) != ERROR_SUCCESS) {
                    srs_error("notice edge start publish stream failed. ret=%d", ret);
                    return ret;
                }
            }
            
435 436 437 438
            if ((ret = rtmp->start_flash_publish(res->stream_id)) != ERROR_SUCCESS) {
                srs_error("flash start to publish stream failed. ret=%d", ret);
                return ret;
            }
439
            
440 441 442 443
            if (!vhost_is_edge) {
                if ((ret = source->acquire_publish()) != ERROR_SUCCESS) {
                    return ret;
                }
444
            }
445
            
446
            ret = flash_publishing(source);
447 448 449
            
            if (!vhost_is_edge) {
                source->release_publish();
450 451
            }
            
452 453 454 455 456 457 458 459
            return ret;
        }
        default: {
            ret = ERROR_SYSTEM_CLIENT_INVALID;
            srs_info("invalid client type=%d. ret=%d", type, ret);
            return ret;
        }
    }
460
461 462 463
    return ret;
}
464
int SrsRtmpConn::check_vhost()
465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493
{
    int ret = ERROR_SUCCESS;
    
    srs_assert(req != NULL);
    
    SrsConfDirective* vhost = _srs_config->get_vhost(req->vhost);
    if (vhost == NULL) {
        ret = ERROR_RTMP_VHOST_NOT_FOUND;
        srs_error("vhost %s not found. ret=%d", req->vhost.c_str(), ret);
        return ret;
    }
    
    if (!_srs_config->get_vhost_enabled(req->vhost)) {
        ret = ERROR_RTMP_VHOST_NOT_FOUND;
        srs_error("vhost %s disabled. ret=%d", req->vhost.c_str(), ret);
        return ret;
    }
    
    if (req->vhost != vhost->arg0()) {
        srs_trace("vhost change from %s to %s", req->vhost.c_str(), vhost->arg0().c_str());
        req->vhost = vhost->arg0();
    }
    
    if ((ret = refer->check(req->pageUrl, _srs_config->get_refer(req->vhost))) != ERROR_SUCCESS) {
        srs_error("check refer failed. ret=%d", ret);
        return ret;
    }
    srs_verbose("check refer success.");
    
494
    if ((ret = http_hooks_on_connect()) != ERROR_SUCCESS) {
495 496 497 498 499 500
        return ret;
    }
    
    return ret;
}
501
int SrsRtmpConn::playing(SrsSource* source)
502 503 504
{
    int ret = ERROR_SUCCESS;
    
505
    // use isolate thread to recv, 
506
    // @see: https://github.com/winlinvip/simple-rtmp-server/issues/217
507 508
    SrsQueueRecvThread trd(rtmp, 
        SRS_CONSTS_RTMP_PULSE_TIMEOUT_US / 1000);
509 510 511 512 513 514 515 516 517 518 519 520 521
    
    // start isolate recv thread.
    if ((ret = trd.start()) != ERROR_SUCCESS) {
        srs_error("start isolate recv thread failed. ret=%d", ret);
        return ret;
    }
    
    // delivery messages for clients playing stream.
    ret = do_playing(source, &trd);
    
    // stop isolate recv thread
    trd.stop();
    
522 523 524 525 526
    // warn for the message is dropped.
    if (!trd.empty()) {
        srs_warn("drop the received %d messages", trd.size());
    }
    
527 528 529
    return ret;
}
530
int SrsRtmpConn::do_playing(SrsSource* source, SrsQueueRecvThread* trd)
531 532 533
{
    int ret = ERROR_SUCCESS;
    
534 535 536 537 538 539 540 541 542 543 544 545 546
    if ((ret = refer->check(req->pageUrl, _srs_config->get_refer_play(req->vhost))) != ERROR_SUCCESS) {
        srs_error("check play_refer failed. ret=%d", ret);
        return ret;
    }
    srs_verbose("check play_refer success.");
    
    SrsConsumer* consumer = NULL;
    if ((ret = source->create_consumer(consumer)) != ERROR_SUCCESS) {
        srs_error("create consumer failed. ret=%d", ret);
        return ret;
    }
    
    srs_assert(consumer != NULL);
547
    SrsAutoFree(SrsConsumer, consumer);
548 549
    srs_verbose("consumer created success.");
    
550 551
    // initialize other components
    SrsPithyPrint pithy_print(SRS_CONSTS_STAGE_PLAY_USER);
552
    SrsMessageArray msgs(SYS_CONSTS_MAX_PLAY_SEND_MSGS);
553
    bool user_specified_duration_to_stop = (req->duration > 0);
554
    int64_t starttime = -1;
555
    
556
    while (true) {
557
        // to use isolate thread to recv, can improve about 33% performance.
winlin authored
558
        // @see: https://github.com/winlinvip/simple-rtmp-server/issues/196
559 560 561
        // @see: https://github.com/winlinvip/simple-rtmp-server/issues/217
        while (!trd->empty()) {
            SrsMessage* msg = trd->pump();
562
            srs_verbose("pump client message to process.");
563
            
564
            if ((ret = process_play_control_msg(consumer, msg)) != ERROR_SUCCESS) {
565
                if (!srs_is_system_control_error(ret) && !srs_is_client_gracefully_close(ret)) {
566
                    srs_error("process play control message failed. ret=%d", ret);
567 568 569 570 571
                }
                return ret;
            }
        }
        
572 573 574 575 576 577 578 579
        // quit when recv thread error.
        if ((ret = trd->error_code()) != ERROR_SUCCESS) {
            if (!srs_is_client_gracefully_close(ret)) {
                srs_error("recv thread failed. ret=%d", ret);
            }
            return ret;
        }
        
580 581 582
        // collect elapse for pithy print.
        pithy_print.elapse();
        
583
        // get messages from consumer.
584
        // each msg in msgs.msgs must be free, for the SrsMessageArray never free them.
585
        int count = 0;
586
        if ((ret = consumer->dump_packets(msgs.max, msgs.msgs, count)) != ERROR_SUCCESS) {
587 588 589
            srs_error("get messages from consumer failed. ret=%d", ret);
            return ret;
        }
590 591 592 593 594 595
        
        // no message to send, sleep a while.
        if (count <= 0) {
            srs_verbose("sleep for no messages to send");
            st_usleep(SRS_CONSTS_RTMP_PULSE_TIMEOUT_US);
        }
596 597 598 599 600 601 602 603

        // reportable
        if (pithy_print.can_print()) {
            kbps->sample();
            srs_trace("-> "SRS_CONSTS_LOG_PLAY
                " time=%"PRId64", msgs=%d, okbps=%d,%d,%d, ikbps=%d,%d,%d", 
                pithy_print.age(), count,
                kbps->get_send_kbps(), kbps->get_send_kbps_30s(), kbps->get_send_kbps_5m(),
604 605
                kbps->get_recv_kbps(), kbps->get_recv_kbps_30s(), kbps->get_recv_kbps_5m()
            );
606 607
        }
        
608 609 610 611
        // only when user specifies the duration, 
        // we start to collect the durations for each message.
        if (user_specified_duration_to_stop) {
            for (int i = 0; i < count; i++) {
612
                SrsMessage* msg = msgs.msgs[i];
613
                
614 615 616
                // foreach msg, collect the duration.
                // @remark: never use msg when sent it, for the protocol sdk will free it.
                if (starttime < 0 || starttime > msg->header.timestamp) {
617 618
                    starttime = msg->header.timestamp;
                }
619 620 621
                duration += msg->header.timestamp - starttime;
                starttime = msg->header.timestamp;
            }
622 623
        }
        
624
        // sendout messages, all messages are freed by send_and_free_messages().
625
        if (count > 0) {
626
            // no need to assert msg, for the rtmp will assert it.
627
            if ((ret = rtmp->send_and_free_messages(msgs.msgs, count, res->stream_id)) != ERROR_SUCCESS) {
628 629 630
                if (!srs_is_client_gracefully_close(ret)) {
                    srs_error("send messages to client failed. ret=%d", ret);
                }
631 632
                return ret;
            }
633 634 635 636
        }
        
        // if duration specified, and exceed it, stop play live.
        // @see: https://github.com/winlinvip/simple-rtmp-server/issues/45
637 638 639 640 641 642
        if (user_specified_duration_to_stop) {
            if (duration >= (int64_t)req->duration) {
                ret = ERROR_RTMP_DURATION_EXCEED;
                srs_trace("stop live for duration exceed. ret=%d", ret);
                return ret;
            }
643 644 645 646 647 648
        }
    }
    
    return ret;
}
649
int SrsRtmpConn::fmle_publishing(SrsSource* source)
650 651 652
{
    int ret = ERROR_SUCCESS;
    
653 654 655 656 657 658 659
    bool vhost_is_edge = _srs_config->get_vhost_is_edge(req->vhost);
            
    if ((ret = http_hooks_on_publish()) != ERROR_SUCCESS) {
        srs_error("http hook on_publish failed. ret=%d", ret);
        return ret;
    }
660 661
    // use isolate thread to recv,
    // @see: https://github.com/winlinvip/simple-rtmp-server/issues/237
662
    SrsPublishRecvThread trd(rtmp, st_netfd_fileno(stfd),
663 664
        SRS_CONSTS_RTMP_RECV_TIMEOUT_US / 1000, 
        this, source, true, vhost_is_edge);
665
666
    srs_info("start to publish stream %s success", req->stream.c_str());
667 668 669 670
    ret = do_publishing(source, &trd);

    // stop isolate recv thread
    trd.stop();
671 672 673 674 675 676 677 678 679 680 681 682 683 684

    // when edge, notice edge to change state.
    // when origin, notice all service to unpublish.
    if (vhost_is_edge) {
        source->on_edge_proxy_unpublish();
    } else {
        source->on_unpublish();
    }

    http_hooks_on_unpublish();
    
    return ret;
}
685
int SrsRtmpConn::flash_publishing(SrsSource* source)
686 687
{
    int ret = ERROR_SUCCESS;
688
689
    bool vhost_is_edge = _srs_config->get_vhost_is_edge(req->vhost);
690
691 692 693 694
    if ((ret = http_hooks_on_publish()) != ERROR_SUCCESS) {
        srs_error("http hook on_publish failed. ret=%d", ret);
        return ret;
    }
695 696 697

    // use isolate thread to recv,
    // @see: https://github.com/winlinvip/simple-rtmp-server/issues/237
698
    SrsPublishRecvThread trd(rtmp, st_netfd_fileno(stfd),
699 700
        SRS_CONSTS_RTMP_RECV_TIMEOUT_US / 1000, 
        this, source, false, vhost_is_edge);
701 702 703 704 705 706

    srs_info("start to publish stream %s success", req->stream.c_str());
    ret = do_publishing(source, &trd);

    // stop isolate recv thread
    trd.stop();
707 708 709 710 711 712 713 714

    // when edge, notice edge to change state.
    // when origin, notice all service to unpublish.
    if (vhost_is_edge) {
        source->on_edge_proxy_unpublish();
    } else {
        source->on_unpublish();
    }
715
716
    http_hooks_on_unpublish();
717
718 719 720
    return ret;
}
721
int SrsRtmpConn::do_publishing(SrsSource* source, SrsPublishRecvThread* trd)
722 723
{
    int ret = ERROR_SUCCESS;
724
725
    if ((ret = refer->check(req->pageUrl, _srs_config->get_refer_publish(req->vhost))) != ERROR_SUCCESS) {
726
        srs_error("check publish_refer failed. ret=%d", ret);
727 728
        return ret;
    }
729 730
    srs_verbose("check publish_refer success.");
winlin authored
731
    SrsPithyPrint pithy_print(SRS_CONSTS_STAGE_PUBLISH_USER);
732
733
    bool vhost_is_edge = _srs_config->get_vhost_is_edge(req->vhost);
734
735
    // when edge, ignore the publish event, directly proxy it.
736
    if (!vhost_is_edge) {
737 738
        // notify the hls to prepare when publish start.
        if ((ret = source->on_publish()) != ERROR_SUCCESS) {
739
            srs_error("hls on_publish failed. ret=%d", ret);
740 741
            return ret;
        }
742
        srs_verbose("hls on_publish success.");
743
    }
744 745 746 747 748 749 750 751

    // start isolate recv thread.
    if ((ret = trd->start()) != ERROR_SUCCESS) {
        srs_error("start isolate recv thread failed. ret=%d", ret);
        return ret;
    }

    int64_t nb_msgs = 0;
752
    while (true) {
753 754
        // cond wait for error.
        trd->wait(SRS_CONSTS_RTMP_RECV_TIMEOUT_US / 1000);
755
756 757 758 759
        // check the thread error code.
        if ((ret = trd->error_code()) != ERROR_SUCCESS) {
            if (!srs_is_client_gracefully_close(ret)) {
                srs_error("recv thread failed. ret=%d", ret);
760
            }
761
            return ret;
762 763
        }
764 765 766 767 768 769 770 771 772
        // when not got any messages, timeout.
        if (trd->nb_msgs() <= nb_msgs) {
            ret = ERROR_SOCKET_TIMEOUT;
            srs_warn("publish timeout %"PRId64"us, nb_msgs=%"PRId64", ret=%d",
                     SRS_CONSTS_RTMP_RECV_TIMEOUT_US, nb_msgs, ret);
            break;
        }
        nb_msgs = trd->nb_msgs();
773
        pithy_print.elapse();
774 775 776

        // reportable
        if (pithy_print.can_print()) {
777
            kbps->sample();
778 779
            srs_trace("<- "SRS_CONSTS_LOG_CLIENT_PUBLISH
                " time=%"PRId64", okbps=%d,%d,%d, ikbps=%d,%d,%d", pithy_print.age(),
780 781
                kbps->get_send_kbps(), kbps->get_send_kbps_30s(), kbps->get_send_kbps_5m(),
                kbps->get_recv_kbps(), kbps->get_recv_kbps_30s(), kbps->get_recv_kbps_5m());
782
        }
783 784 785 786 787 788 789 790
    }

    return ret;
}

int SrsRtmpConn::handle_publish_message(SrsSource* source, SrsMessage* msg, bool is_fmle, bool vhost_is_edge)
{
    int ret = ERROR_SUCCESS;
791
    
792 793 794 795 796 797 798 799 800 801 802 803
    // process publish event.
    if (msg->header.is_amf0_command() || msg->header.is_amf3_command()) {
        SrsPacket* pkt = NULL;
        if ((ret = rtmp->decode_message(msg, &pkt)) != ERROR_SUCCESS) {
            srs_error("fmle decode unpublish message failed. ret=%d", ret);
            return ret;
        }

        SrsAutoFree(SrsPacket, pkt);

        // for flash, any packet is republish.
        if (!is_fmle) {
804 805 806 807 808 809
            // flash unpublish.
            // TODO: maybe need to support republish.
            srs_trace("flash flash publish finished.");
            return ERROR_CONTROL_REPUBLISH;
        }
810 811 812 813 814 815 816
        // for fmle, drop others except the fmle start packet.
        if (dynamic_cast<SrsFMLEStartPacket*>(pkt)) {
            SrsFMLEStartPacket* unpublish = dynamic_cast<SrsFMLEStartPacket*>(pkt);
            if ((ret = rtmp->fmle_unpublish(res->stream_id, unpublish->transaction_id)) != ERROR_SUCCESS) {
                return ret;
            }
            return ERROR_CONTROL_REPUBLISH;
817
        }
818 819 820 821 822 823 824 825 826

        srs_trace("fmle ignore AMF0/AMF3 command message.");
        return ret;
    }

    // video, audio, data message
    if ((ret = process_publish_message(source, msg, vhost_is_edge)) != ERROR_SUCCESS) {
        srs_error("fmle process publish message failed. ret=%d", ret);
        return ret;
827 828 829 830 831
    }
    
    return ret;
}
832
int SrsRtmpConn::process_publish_message(SrsSource* source, SrsMessage* msg, bool vhost_is_edge)
833 834 835
{
    int ret = ERROR_SUCCESS;
    
836 837
    // for edge, directly proxy message to origin.
    if (vhost_is_edge) {
838 839 840 841 842
        if ((ret = source->on_edge_proxy_publish(msg)) != ERROR_SUCCESS) {
            srs_error("edge publish proxy msg failed. ret=%d", ret);
            return ret;
        }
        return ret;
843 844
    }
    
845 846 847 848 849 850
    // process audio packet
    if (msg->header.is_audio()) {
        if ((ret = source->on_audio(msg)) != ERROR_SUCCESS) {
            srs_error("source process audio message failed. ret=%d", ret);
            return ret;
        }
winlin authored
851
        return ret;
852 853 854 855 856 857 858
    }
    // process video packet
    if (msg->header.is_video()) {
        if ((ret = source->on_video(msg)) != ERROR_SUCCESS) {
            srs_error("source process video message failed. ret=%d", ret);
            return ret;
        }
winlin authored
859
        return ret;
860 861
    }
    
862 863 864 865 866 867 868 869 870
    // process aggregate packet
    if (msg->header.is_aggregate()) {
        if ((ret = source->on_aggregate(msg)) != ERROR_SUCCESS) {
            srs_error("source process aggregate message failed. ret=%d", ret);
            return ret;
        }
        return ret;
    }
    
871 872
    // process onMetaData
    if (msg->header.is_amf0_data() || msg->header.is_amf3_data()) {
873
        SrsPacket* pkt = NULL;
874
        if ((ret = rtmp->decode_message(msg, &pkt)) != ERROR_SUCCESS) {
875 876 877
            srs_error("decode onMetaData message failed. ret=%d", ret);
            return ret;
        }
878
        SrsAutoFree(SrsPacket, pkt);
879 880 881 882 883 884 885
    
        if (dynamic_cast<SrsOnMetaDataPacket*>(pkt)) {
            SrsOnMetaDataPacket* metadata = dynamic_cast<SrsOnMetaDataPacket*>(pkt);
            if ((ret = source->on_meta_data(msg, metadata)) != ERROR_SUCCESS) {
                srs_error("source process onMetaData message failed. ret=%d", ret);
                return ret;
            }
886
            srs_info("process onMetaData message success.");
887 888 889
            return ret;
        }
        
890
        srs_info("ignore AMF0/AMF3 data message.");
891 892 893 894 895 896
        return ret;
    }
    
    return ret;
}
897
int SrsRtmpConn::process_play_control_msg(SrsConsumer* consumer, SrsMessage* msg)
898 899 900 901 902 903 904
{
    int ret = ERROR_SUCCESS;
    
    if (!msg) {
        srs_verbose("ignore all empty message.");
        return ret;
    }
905
    SrsAutoFree(SrsMessage, msg);
906 907 908 909 910 911
    
    if (!msg->header.is_amf0_command() && !msg->header.is_amf3_command()) {
        srs_info("ignore all message except amf0/amf3 command.");
        return ret;
    }
    
912
    SrsPacket* pkt = NULL;
913
    if ((ret = rtmp->decode_message(msg, &pkt)) != ERROR_SUCCESS) {
914 915 916 917 918
        srs_error("decode the amf0/amf3 command packet failed. ret=%d", ret);
        return ret;
    }
    srs_info("decode the amf0/amf3 command packet success.");
    
919
    SrsAutoFree(SrsPacket, pkt);
920
    
921 922
    // for jwplayer/flowplayer, which send close as pause message.
    // @see https://github.com/winlinvip/simple-rtmp-server/issues/6
923
    SrsCloseStreamPacket* close = dynamic_cast<SrsCloseStreamPacket*>(pkt);
924 925 926 927 928 929
    if (close) {
        ret = ERROR_CONTROL_RTMP_CLOSE;
        srs_trace("system control message: rtmp close stream. ret=%d", ret);
        return ret;
    }
    
930 931 932
    // call msg,
    // support response null first,
    // @see https://github.com/winlinvip/simple-rtmp-server/issues/106
933
    // TODO: FIXME: response in right way, or forward in edge mode.
934
    SrsCallPacket* call = dynamic_cast<SrsCallPacket*>(pkt);
winlin authored
935 936 937 938 939 940 941 942 943 944 945
    if (call) {
        // only response it when transaction id not zero,
        // for the zero means donot need response.
        if (call->transaction_id > 0) {
            SrsCallResPacket* res = new SrsCallResPacket(call->transaction_id);
            res->command_object = SrsAmf0Any::null();
            res->response = SrsAmf0Any::null();
            if ((ret = rtmp->send_and_free_packet(res, 0)) != ERROR_SUCCESS) {
                srs_warn("response call failed. ret=%d", ret);
                return ret;
            }
946 947 948 949 950
        }
        return ret;
    }
    
    // pause or other msg.
951
    SrsPausePacket* pause = dynamic_cast<SrsPausePacket*>(pkt);
952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970
    if (!pause) {
        srs_info("ignore all amf0/amf3 command except pause.");
        return ret;
    }
    
    if ((ret = rtmp->on_play_client_pause(res->stream_id, pause->is_pause)) != ERROR_SUCCESS) {
        srs_error("rtmp process play client pause failed. ret=%d", ret);
        return ret;
    }
    
    if ((ret = consumer->on_play_client_pause(pause->is_pause)) != ERROR_SUCCESS) {
        srs_error("consumer process play client pause failed. ret=%d", ret);
        return ret;
    }
    srs_info("process pause success, is_pause=%d, time=%d.", pause->is_pause, pause->time_ms);
    
    return ret;
}
971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989
int SrsRtmpConn::check_edge_token_traverse_auth()
{
    int ret = ERROR_SUCCESS;
    
    srs_assert(req);
    
    st_netfd_t stsock = NULL;
    SrsConfDirective* conf = _srs_config->get_vhost_edge_origin(req->vhost);
    for (int i = 0; i < (int)conf->args.size(); i++) {
        if ((ret = connect_server(i, &stsock)) == ERROR_SUCCESS) {
            break;
        }
    }
    if (ret != ERROR_SUCCESS) {
        srs_warn("token traverse connect failed. ret=%d", ret);
        return ret;
    }
    
    srs_assert(stsock);
990
    SrsStSocket* io = new SrsStSocket(stsock);
991 992
    SrsRtmpClient* client = new SrsRtmpClient(io);
    
993
    ret = do_token_traverse_auth(client);
994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012

    srs_freep(client);
    srs_freep(io);
    srs_close_stfd(stsock);

    return ret;
}

int SrsRtmpConn::connect_server(int origin_index, st_netfd_t* pstsock)
{
    int ret = ERROR_SUCCESS;
    
    SrsConfDirective* conf = _srs_config->get_vhost_edge_origin(req->vhost);
    srs_assert(conf);
    
    // select the origin.
    std::string server = conf->args.at(origin_index % conf->args.size());
    origin_index = (origin_index + 1) % conf->args.size();
    
1013 1014
    std::string s_port = SRS_CONSTS_RTMP_DEFAULT_PORT;
    int port = ::atoi(SRS_CONSTS_RTMP_DEFAULT_PORT);
1015 1016 1017 1018 1019 1020 1021 1022
    size_t pos = server.find(":");
    if (pos != std::string::npos) {
        s_port = server.substr(pos + 1);
        server = server.substr(0, pos);
        port = ::atoi(s_port.c_str());
    }
    
    // open socket.
1023 1024 1025 1026 1027
    st_netfd_t stsock = NULL;
    int64_t timeout = SRS_EDGE_TOKEN_TRAVERSE_TIMEOUT_US;
    if ((ret = srs_socket_connect(server, port, timeout, &stsock)) != ERROR_SUCCESS) {
        srs_warn("edge token traverse failed, tcUrl=%s to server=%s, port=%d, timeout=%"PRId64", ret=%d",
            req->tcUrl.c_str(), server.c_str(), port, timeout, ret);
1028 1029 1030 1031 1032 1033 1034 1035
        return ret;
    }
    srs_info("edge token auth connected, url=%s/%s, server=%s:%d", req->tcUrl.c_str(), req->stream.c_str(), server.c_str(), port);
    
    *pstsock = stsock;
    return ret;
}
1036
int SrsRtmpConn::do_token_traverse_auth(SrsRtmpClient* client)
1037 1038 1039 1040 1041
{
    int ret = ERROR_SUCCESS;
    
    srs_assert(client);
winlin authored
1042 1043
    client->set_recv_timeout(SRS_CONSTS_RTMP_RECV_TIMEOUT_US);
    client->set_send_timeout(SRS_CONSTS_RTMP_SEND_TIMEOUT_US);
1044 1045 1046 1047 1048
    
    if ((ret = client->handshake()) != ERROR_SUCCESS) {
        srs_error("handshake with server failed. ret=%d", ret);
        return ret;
    }
1049 1050 1051
    
    // for token tranverse, always take the debug info(which carries token).
    if ((ret = client->connect_app(req->app, req->tcUrl, req, true)) != ERROR_SUCCESS) {
1052 1053 1054 1055 1056 1057 1058 1059 1060
        srs_error("connect with server failed, tcUrl=%s. ret=%d", req->tcUrl.c_str(), ret);
        return ret;
    }
    
    srs_trace("edge token auth ok, tcUrl=%s", req->tcUrl.c_str());
    
    return ret;
}
1061
int SrsRtmpConn::http_hooks_on_connect()
1062 1063 1064
{
    int ret = ERROR_SUCCESS;
    
1065
#ifdef SRS_AUTO_HTTP_CALLBACK
1066
    if (_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
1067 1068 1069 1070
        // HTTP: on_connect 
        SrsConfDirective* on_connect = _srs_config->get_vhost_on_connect(req->vhost);
        if (!on_connect) {
            srs_info("ignore the empty http callback: on_connect");
1071 1072
            return ret;
        }
1073 1074 1075 1076 1077 1078 1079 1080 1081
        
        int connection_id = _srs_context->get_id();
        for (int i = 0; i < (int)on_connect->args.size(); i++) {
            std::string url = on_connect->args.at(i);
            if ((ret = SrsHttpHooks::on_connect(url, connection_id, ip, req)) != ERROR_SUCCESS) {
                srs_error("hook client on_connect failed. url=%s, ret=%d", url.c_str(), ret);
                return ret;
            }
        }
1082 1083 1084 1085 1086 1087
    }
#endif

    return ret;
}
1088
void SrsRtmpConn::http_hooks_on_close()
1089
{
1090
#ifdef SRS_AUTO_HTTP_CALLBACK
1091
    if (_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104
        // whatever the ret code, notify the api hooks.
        // HTTP: on_close 
        SrsConfDirective* on_close = _srs_config->get_vhost_on_close(req->vhost);
        if (!on_close) {
            srs_info("ignore the empty http callback: on_close");
            return;
        }
        
        int connection_id = _srs_context->get_id();
        for (int i = 0; i < (int)on_close->args.size(); i++) {
            std::string url = on_close->args.at(i);
            SrsHttpHooks::on_close(url, connection_id, ip, req);
        }
1105 1106 1107 1108
    }
#endif
}
1109
int SrsRtmpConn::http_hooks_on_publish()
1110 1111 1112
{
    int ret = ERROR_SUCCESS;
    
1113
#ifdef SRS_AUTO_HTTP_CALLBACK
1114
    if (_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
1115 1116 1117 1118
        // HTTP: on_publish 
        SrsConfDirective* on_publish = _srs_config->get_vhost_on_publish(req->vhost);
        if (!on_publish) {
            srs_info("ignore the empty http callback: on_publish");
1119 1120
            return ret;
        }
1121 1122 1123 1124 1125 1126 1127 1128 1129
        
        int connection_id = _srs_context->get_id();
        for (int i = 0; i < (int)on_publish->args.size(); i++) {
            std::string url = on_publish->args.at(i);
            if ((ret = SrsHttpHooks::on_publish(url, connection_id, ip, req)) != ERROR_SUCCESS) {
                srs_error("hook client on_publish failed. url=%s, ret=%d", url.c_str(), ret);
                return ret;
            }
        }
1130 1131 1132 1133 1134 1135
    }
#endif

    return ret;
}
1136
void SrsRtmpConn::http_hooks_on_unpublish()
1137
{
1138
#ifdef SRS_AUTO_HTTP_CALLBACK
1139
    if (_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152
        // whatever the ret code, notify the api hooks.
        // HTTP: on_unpublish 
        SrsConfDirective* on_unpublish = _srs_config->get_vhost_on_unpublish(req->vhost);
        if (!on_unpublish) {
            srs_info("ignore the empty http callback: on_unpublish");
            return;
        }
        
        int connection_id = _srs_context->get_id();
        for (int i = 0; i < (int)on_unpublish->args.size(); i++) {
            std::string url = on_unpublish->args.at(i);
            SrsHttpHooks::on_unpublish(url, connection_id, ip, req);
        }
1153 1154 1155 1156
    }
#endif
}
1157
int SrsRtmpConn::http_hooks_on_play()
1158 1159 1160
{
    int ret = ERROR_SUCCESS;
    
1161
#ifdef SRS_AUTO_HTTP_CALLBACK
1162
    if (_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
1163 1164 1165 1166
        // HTTP: on_play 
        SrsConfDirective* on_play = _srs_config->get_vhost_on_play(req->vhost);
        if (!on_play) {
            srs_info("ignore the empty http callback: on_play");
1167 1168
            return ret;
        }
1169 1170 1171 1172 1173 1174 1175 1176 1177
        
        int connection_id = _srs_context->get_id();
        for (int i = 0; i < (int)on_play->args.size(); i++) {
            std::string url = on_play->args.at(i);
            if ((ret = SrsHttpHooks::on_play(url, connection_id, ip, req)) != ERROR_SUCCESS) {
                srs_error("hook client on_play failed. url=%s, ret=%d", url.c_str(), ret);
                return ret;
            }
        }
1178 1179 1180 1181 1182 1183
    }
#endif

    return ret;
}
1184
void SrsRtmpConn::http_hooks_on_stop()
1185
{
1186
#ifdef SRS_AUTO_HTTP_CALLBACK
1187
    if (_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200
        // whatever the ret code, notify the api hooks.
        // HTTP: on_stop 
        SrsConfDirective* on_stop = _srs_config->get_vhost_on_stop(req->vhost);
        if (!on_stop) {
            srs_info("ignore the empty http callback: on_stop");
            return;
        }
        
        int connection_id = _srs_context->get_id();
        for (int i = 0; i < (int)on_stop->args.size(); i++) {
            std::string url = on_stop->args.at(i);
            SrsHttpHooks::on_stop(url, connection_id, ip, req);
        }
1201 1202 1203 1204 1205
    }
#endif

    return;
}
1206