srs/trunk/src/app/srs_app_dvr.cpp

/**
 * The MIT License (MIT)
 *
 * Copyright (c) 2013-2020 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_dvr.hpp>

#include <fcntl.h>
#include <sstream>
#include <algorithm>
using namespace std;

#include <srs_app_config.hpp>
#include <srs_rtmp_stack.hpp>
#include <srs_core_autofree.hpp>
#include <srs_kernel_utility.hpp>
#include <srs_app_http_hooks.hpp>
#include <srs_kernel_codec.hpp>
#include <srs_kernel_flv.hpp>
#include <srs_kernel_file.hpp>
#include <srs_protocol_amf0.hpp>
#include <srs_kernel_buffer.hpp>
#include <srs_protocol_json.hpp>
#include <srs_app_utility.hpp>
#include <srs_kernel_mp4.hpp>
#include <srs_app_fragment.hpp>

SrsDvrSegmenter::SrsDvrSegmenter()
{
    req = NULL;
    jitter = NULL;
    plan = NULL;
    wait_keyframe = true;
    
    fragment = new SrsFragment();
    fs = new SrsFileWriter();
    jitter_algorithm = SrsRtmpJitterAlgorithmOFF;
    
    _srs_config->subscribe(this);
}

SrsDvrSegmenter::~SrsDvrSegmenter()
{
    _srs_config->unsubscribe(this);
    
    srs_freep(fragment);
    srs_freep(jitter);
    srs_freep(fs);
}

srs_error_t SrsDvrSegmenter::initialize(SrsDvrPlan* p, SrsRequest* r)
{
    req = r;
    plan = p;
    
    jitter_algorithm = (SrsRtmpJitterAlgorithm)_srs_config->get_dvr_time_jitter(req->vhost);
    wait_keyframe = _srs_config->get_dvr_wait_keyframe(req->vhost);
    
    return srs_success;
}

SrsFragment* SrsDvrSegmenter::current()
{
    return fragment;
}

srs_error_t SrsDvrSegmenter::open()
{
    srs_error_t err = srs_success;
    
    // ignore when already open.
    if (fs->is_open()) {
        return err;
    }
    
    string path = generate_path();
    if (srs_path_exists(path)) {
        return srs_error_new(ERROR_DVR_CANNOT_APPEND, "DVR can't append to exists path=%s", path.c_str());
    }
    fragment->set_path(path);
    
    // create dir first.
    if ((err = fragment->create_dir()) != srs_success) {
        return srs_error_wrap(err, "create dir");
    }
    
    // create jitter.
    srs_freep(jitter);
    jitter = new SrsRtmpJitter();
    
    // open file writer, in append or create mode.
    string tmp_dvr_file = fragment->tmppath();
    if ((err = fs->open(tmp_dvr_file)) != srs_success) {
        return srs_error_wrap(err, "open file %s", path.c_str());
    }
    
    // initialize the encoder.
    if ((err = open_encoder()) != srs_success) {
        return srs_error_wrap(err, "open encoder");
    }
    
    srs_trace("dvr stream %s to file %s", req->stream.c_str(), path.c_str());
    return err;
}

srs_error_t SrsDvrSegmenter::write_metadata(SrsSharedPtrMessage* metadata)
{
    return encode_metadata(metadata);
}

srs_error_t SrsDvrSegmenter::write_audio(SrsSharedPtrMessage* shared_audio, SrsFormat* format)
{
    srs_error_t err = srs_success;
    
    SrsSharedPtrMessage* audio = shared_audio->copy();
    SrsAutoFree(SrsSharedPtrMessage, audio);
    
    if ((err = jitter->correct(audio, jitter_algorithm)) != srs_success) {
        return srs_error_wrap(err, "jitter");
    }
    
    if ((err = on_update_duration(audio)) != srs_success) {
        return srs_error_wrap(err, "update duration");
    }
    
    if ((err = encode_audio(audio, format)) != srs_success) {
        return srs_error_wrap(err, "encode audio");
    }
    
    return err;
}

srs_error_t SrsDvrSegmenter::write_video(SrsSharedPtrMessage* shared_video, SrsFormat* format)
{
    srs_error_t err = srs_success;
    
    SrsSharedPtrMessage* video = shared_video->copy();
    SrsAutoFree(SrsSharedPtrMessage, video);
    
    if ((err = jitter->correct(video, jitter_algorithm)) != srs_success) {
        return srs_error_wrap(err, "jitter");
    }
    
    if ((err = encode_video(video, format)) != srs_success) {
        return srs_error_wrap(err, "encode video");
    }
    
    if ((err = on_update_duration(video)) != srs_success) {
        return srs_error_wrap(err, "update duration");
    }
    
    return err;
}

srs_error_t SrsDvrSegmenter::close()
{
    srs_error_t err = srs_success;
    
    // ignore when already closed.
    if (!fs->is_open()) {
        return err;
    }
    
    // Close the encoder, then close the fs object.
    err = close_encoder();
    fs->close(); // Always close the file.
    if (err != srs_success) {
        return srs_error_wrap(err, "close encoder");
    }
    
    // when tmp flv file exists, reap it.
    if ((err = fragment->rename()) != srs_success) {
        return srs_error_wrap(err, "rename fragment");
    }
    
    // TODO: FIXME: the http callback is async, which will trigger thread switch,
    //          so the on_video maybe invoked during the http callback, and error.
    if ((err = plan->on_reap_segment()) != srs_success) {
        return srs_error_wrap(err, "reap segment");
    }
    
    return err;
}

string SrsDvrSegmenter::generate_path()
{
    // the path in config, for example,
    //      /data/[vhost]/[app]/[stream]/[2006]/[01]/[02]/[15].[04].[05].[999].flv
    std::string path_config = _srs_config->get_dvr_path(req->vhost);
    
    // add [stream].[timestamp].flv as filename for dir
    if (!srs_string_ends_with(path_config, ".flv", ".mp4")) {
        path_config += "/[stream].[timestamp].flv";
    }
    
    // the flv file path
    std::string flv_path = path_config;
    flv_path = srs_path_build_stream(flv_path, req->vhost, req->app, req->stream);
    flv_path = srs_path_build_timestamp(flv_path);
    
    return flv_path;
}

srs_error_t SrsDvrSegmenter::on_update_duration(SrsSharedPtrMessage* msg)
{
    fragment->append(msg->timestamp);
    return srs_success;
}

srs_error_t SrsDvrSegmenter::on_reload_vhost_dvr(std::string vhost)
{
    srs_error_t err = srs_success;
    
    if (req->vhost != vhost) {
        return err;
    }
    
    jitter_algorithm = (SrsRtmpJitterAlgorithm)_srs_config->get_dvr_time_jitter(req->vhost);
    wait_keyframe = _srs_config->get_dvr_wait_keyframe(req->vhost);
    
    return err;
}

SrsDvrFlvSegmenter::SrsDvrFlvSegmenter()
{
    enc = new SrsFlvTransmuxer();
    
    duration_offset = 0;
    filesize_offset = 0;
    
    has_keyframe = false;
}

SrsDvrFlvSegmenter::~SrsDvrFlvSegmenter()
{
    srs_freep(enc);
}

srs_error_t SrsDvrFlvSegmenter::refresh_metadata()
{
    srs_error_t err = srs_success;
    
    // no duration or filesize specified.
    if (!duration_offset || !filesize_offset) {
        return err;
    }
    
    int64_t cur = fs->tellg();
    
    // buffer to write the size.
    char* buf = new char[SrsAmf0Size::number()];
    SrsAutoFreeA(char, buf);
    
    SrsBuffer stream(buf, SrsAmf0Size::number());
    
    // filesize to buf.
    SrsAmf0Any* size = SrsAmf0Any::number((double)cur);
    SrsAutoFree(SrsAmf0Any, size);
    
    stream.skip(-1 * stream.pos());
    if ((err = size->write(&stream)) != srs_success) {
        return srs_error_wrap(err, "write filesize");
    }
    
    // update the flesize.
    fs->seek2(filesize_offset);
    if ((err = fs->write(buf, SrsAmf0Size::number(), NULL)) != srs_success) {
        return srs_error_wrap(err, "update filesize");
    }
    
    // duration to buf
    SrsAmf0Any* dur = SrsAmf0Any::number((double)srsu2ms(fragment->duration()) / 1000.0);
    SrsAutoFree(SrsAmf0Any, dur);
    
    stream.skip(-1 * stream.pos());
    if ((err = dur->write(&stream)) != srs_success) {
        return srs_error_wrap(err, "write duration");
    }
    
    // update the duration
    fs->seek2(duration_offset);
    if ((err = fs->write(buf, SrsAmf0Size::number(), NULL)) != srs_success) {
        return srs_error_wrap(err, "update duration");
    }
    
    // reset the offset.
    fs->seek2(cur);
    
    return err;
}

srs_error_t SrsDvrFlvSegmenter::open_encoder()
{
    srs_error_t err = srs_success;
    
    has_keyframe = false;
    
    // update the duration and filesize offset.
    duration_offset = 0;
    filesize_offset = 0;
    
    srs_freep(enc);
    enc = new SrsFlvTransmuxer();
    
    if ((err = enc->initialize(fs)) != srs_success) {
        return srs_error_wrap(err, "init encoder");
    }
    
    // write the flv header to writer.
    if ((err = enc->write_header()) != srs_success) {
        return srs_error_wrap(err, "write flv header");
    }
    
    return err;
}

srs_error_t SrsDvrFlvSegmenter::encode_metadata(SrsSharedPtrMessage* metadata)
{
    srs_error_t err = srs_success;
    
    // Ignore when metadata already written.
    if (duration_offset || filesize_offset) {
        return err;
    }

    SrsBuffer stream(metadata->payload, metadata->size);
    
    SrsAmf0Any* name = SrsAmf0Any::str();
    SrsAutoFree(SrsAmf0Any, name);
    if ((err = name->read(&stream)) != srs_success) {
        return srs_error_wrap(err, "read name");
    }
    
    SrsAmf0Object* obj = SrsAmf0Any::object();
    SrsAutoFree(SrsAmf0Object, obj);
    if ((err = obj->read(&stream)) != srs_success) {
        return srs_error_wrap(err, "read object");
    }
    
    // remove duration and filesize.
    obj->set("filesize", NULL);
    obj->set("duration", NULL);
    
    // add properties.
    obj->set("service", SrsAmf0Any::str(RTMP_SIG_SRS_SERVER));
    obj->set("filesize", SrsAmf0Any::number(0));
    obj->set("duration", SrsAmf0Any::number(0));
    
    int size = name->total_size() + obj->total_size();
    char* payload = new char[size];
    SrsAutoFreeA(char, payload);
    
    // 11B flv header, 3B object EOF, 8B number value, 1B number flag.
    duration_offset = fs->tellg() + size + 11 - SrsAmf0Size::object_eof() - SrsAmf0Size::number();
    // 2B string flag, 8B number value, 8B string 'duration', 1B number flag
    filesize_offset = duration_offset - SrsAmf0Size::utf8("duration") - SrsAmf0Size::number();
    
    // convert metadata to bytes.
    SrsBuffer buf(payload, size);
    
    if ((err = name->write(&buf)) != srs_success) {
        return srs_error_wrap(err, "write name");
    }
    if ((err = obj->write(&buf)) != srs_success) {
        return srs_error_wrap(err, "write object");
    }
    
    // to flv file.
    if ((err = enc->write_metadata(18, payload, size)) != srs_success) {
        return srs_error_wrap(err, "write metadata");
    }
    
    return err;
}

srs_error_t SrsDvrFlvSegmenter::encode_audio(SrsSharedPtrMessage* audio, SrsFormat* format)
{
    srs_error_t err = srs_success;
    
    char* payload = audio->payload;
    int size = audio->size;
    if ((err = enc->write_audio(audio->timestamp, payload, size)) != srs_success) {
        return srs_error_wrap(err, "write audio");
    }
    
    return err;
}

srs_error_t SrsDvrFlvSegmenter::encode_video(SrsSharedPtrMessage* video, SrsFormat* format)
{
    srs_error_t err = srs_success;
    
    char* payload = video->payload;
    int size = video->size;
    bool sh = (format->video->avc_packet_type == SrsVideoAvcFrameTraitSequenceHeader);
    bool keyframe = (!sh && format->video->frame_type == SrsVideoAvcFrameTypeKeyFrame);
    
    if (keyframe) {
        has_keyframe = true;
    }
    
    // accept the sequence header here.
    // when got no keyframe, ignore when should wait keyframe.
    if (!has_keyframe && !sh && wait_keyframe) {
        return err;
    }
    
    if ((err = enc->write_video(video->timestamp, payload, size)) != srs_success) {
        return srs_error_wrap(err, "write video");
    }
    
    return err;
}

srs_error_t SrsDvrFlvSegmenter::close_encoder()
{
    return refresh_metadata();
}

SrsDvrMp4Segmenter::SrsDvrMp4Segmenter()
{
    enc = new SrsMp4Encoder();
}

SrsDvrMp4Segmenter::~SrsDvrMp4Segmenter()
{
    srs_freep(enc);
}

srs_error_t SrsDvrMp4Segmenter::refresh_metadata()
{
    return srs_success;
}

srs_error_t SrsDvrMp4Segmenter::open_encoder()
{
    srs_error_t err = srs_success;
    
    srs_freep(enc);
    enc = new SrsMp4Encoder();
    
    if ((err = enc->initialize(fs)) != srs_success) {
        return srs_error_wrap(err, "init encoder");
    }
    
    return err;
}

srs_error_t SrsDvrMp4Segmenter::encode_metadata(SrsSharedPtrMessage* /*metadata*/)
{
    return srs_success;
}

srs_error_t SrsDvrMp4Segmenter::encode_audio(SrsSharedPtrMessage* audio, SrsFormat* format)
{
    srs_error_t err = srs_success;
    
    SrsAudioCodecId sound_format = format->acodec->id;
    SrsAudioSampleRate sound_rate = format->acodec->sound_rate;
    SrsAudioSampleBits sound_size = format->acodec->sound_size;
    SrsAudioChannels channels = format->acodec->sound_type;
    
    SrsAudioAacFrameTrait ct = format->audio->aac_packet_type;
    if (ct == SrsAudioAacFrameTraitSequenceHeader) {
        enc->acodec = sound_format;
        enc->sample_rate = sound_rate;
        enc->sound_bits = sound_size;
        enc->channels = channels;
    }
    
    uint8_t* sample = (uint8_t*)format->raw;
    uint32_t nb_sample = (uint32_t)format->nb_raw;
    
    uint32_t dts = (uint32_t)audio->timestamp;
    if ((err = enc->write_sample(format, SrsMp4HandlerTypeSOUN, 0x00, ct, dts, dts, sample, nb_sample)) != srs_success) {
        return srs_error_wrap(err, "write sample");
    }
    
    return err;
}

srs_error_t SrsDvrMp4Segmenter::encode_video(SrsSharedPtrMessage* video, SrsFormat* format)
{
    srs_error_t err = srs_success;
    
    SrsVideoAvcFrameType frame_type = format->video->frame_type;
    SrsVideoCodecId codec_id = format->vcodec->id;
    
    SrsVideoAvcFrameTrait ct = format->video->avc_packet_type;
    uint32_t cts = (uint32_t)format->video->cts;
    
    if (ct == SrsVideoAvcFrameTraitSequenceHeader) {
        enc->vcodec = codec_id;
    }
    
    uint32_t dts = (uint32_t)video->timestamp;
    uint32_t pts = dts + cts;
    
    uint8_t* sample = (uint8_t*)format->raw;
    uint32_t nb_sample = (uint32_t)format->nb_raw;
    if ((err = enc->write_sample(format, SrsMp4HandlerTypeVIDE, frame_type, ct, dts, pts, sample, nb_sample)) != srs_success) {
        return srs_error_wrap(err, "write sample");
    }
    
    return err;
}

srs_error_t SrsDvrMp4Segmenter::close_encoder()
{
    srs_error_t err = srs_success;
    
    if ((err = enc->flush()) != srs_success) {
        return srs_error_wrap(err, "flush encoder");
    }
    
    return err;
}

SrsDvrAsyncCallOnDvr::SrsDvrAsyncCallOnDvr(SrsContextId c, SrsRequest* r, string p)
{
    cid = c;
    req = r->copy();
    path = p;
}

SrsDvrAsyncCallOnDvr::~SrsDvrAsyncCallOnDvr()
{
    srs_freep(req);
}

srs_error_t SrsDvrAsyncCallOnDvr::call()
{
    srs_error_t err = srs_success;
    
    if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
        return err;
    }
    
    // the http hooks will cause context switch,
    // so we must copy all hooks for the on_connect may freed.
    // @see https://github.com/ossrs/srs/issues/475
    vector<string> hooks;
    
    if (true) {
        SrsConfDirective* conf = _srs_config->get_vhost_on_dvr(req->vhost);
        if (conf) {
            hooks = conf->args;
        }
    }
    
    for (int i = 0; i < (int)hooks.size(); i++) {
        std::string url = hooks.at(i);
        if ((err = SrsHttpHooks::on_dvr(cid, url, req, path)) != srs_success) {
            return srs_error_wrap(err, "callback on_dvr %s", url.c_str());
        }
    }
    
    return err;
}

string SrsDvrAsyncCallOnDvr::to_string()
{
    std::stringstream ss;
    ss << "vhost=" << req->vhost << ", file=" << path;
    return ss.str();
}

SrsDvrPlan::SrsDvrPlan()
{
    req = NULL;
    hub = NULL;
    
    dvr_enabled = false;
    segment = NULL;
    async = new SrsAsyncCallWorker();
}

SrsDvrPlan::~SrsDvrPlan()
{
    srs_freep(segment);
    srs_freep(async);
}

srs_error_t SrsDvrPlan::initialize(SrsOriginHub* h, SrsDvrSegmenter* s, SrsRequest* r)
{
    srs_error_t err = srs_success;
    
    hub = h;
    req = r;
    segment = s;
    
    if ((err = segment->initialize(this, r)) != srs_success) {
        return srs_error_wrap(err, "segmenter");
    }
    
    return err;
}

srs_error_t SrsDvrPlan::on_publish()
{
    srs_error_t err = srs_success;

    if ((err = async->start()) != srs_success) {
        return srs_error_wrap(err, "async");
    }

    return err;
}

void SrsDvrPlan::on_unpublish()
{
    async->stop();
}

srs_error_t SrsDvrPlan::on_meta_data(SrsSharedPtrMessage* shared_metadata)
{
    srs_error_t err = srs_success;
    
    if (!dvr_enabled) {
        return err;
    }
    
    return segment->write_metadata(shared_metadata);
}

srs_error_t SrsDvrPlan::on_audio(SrsSharedPtrMessage* shared_audio, SrsFormat* format)
{
    srs_error_t err = srs_success;
    
    if (!dvr_enabled) {
        return err;
    }
    
    if ((err = segment->write_audio(shared_audio, format)) != srs_success) {
        return srs_error_wrap(err, "write audio");
    }
    
    return err;
}

srs_error_t SrsDvrPlan::on_video(SrsSharedPtrMessage* shared_video, SrsFormat* format)
{
    srs_error_t err = srs_success;
    
    if (!dvr_enabled) {
        return err;
    }
    
    if ((err = segment->write_video(shared_video, format)) != srs_success) {
        return srs_error_wrap(err, "write video");
    }
    
    return err;
}

srs_error_t SrsDvrPlan::on_reap_segment()
{
    srs_error_t err = srs_success;
    
    SrsContextId cid = _srs_context->get_id();
    
    SrsFragment* fragment = segment->current();
    string fullpath = fragment->fullpath();
    
    if ((err = async->execute(new SrsDvrAsyncCallOnDvr(cid, req, fullpath))) != srs_success) {
        return srs_error_wrap(err, "reap segment");
    }
    
    return err;
}

srs_error_t SrsDvrPlan::create_plan(string vhost, SrsDvrPlan** pplan)
{
    std::string plan = _srs_config->get_dvr_plan(vhost);
    if (srs_config_dvr_is_plan_segment(plan)) {
        *pplan = new SrsDvrSegmentPlan();
    } else if (srs_config_dvr_is_plan_session(plan)) {
        *pplan = new SrsDvrSessionPlan();
    } else {
        return srs_error_new(ERROR_DVR_ILLEGAL_PLAN, "illegal plan=%s, vhost=%s",
            plan.c_str(), vhost.c_str());
    }
    
    return srs_success;
}

SrsDvrSessionPlan::SrsDvrSessionPlan()
{
}

SrsDvrSessionPlan::~SrsDvrSessionPlan()
{
}

srs_error_t SrsDvrSessionPlan::on_publish()
{
    srs_error_t err = srs_success;

    if ((err = SrsDvrPlan::on_publish()) != srs_success) {
        return err;
    }
    
    // support multiple publish.
    if (dvr_enabled) {
        return err;
    }
    
    if (!_srs_config->get_dvr_enabled(req->vhost)) {
        return err;
    }
    
    if ((err = segment->close()) != srs_success) {
        return srs_error_wrap(err, "close segment");
    }
    
    if ((err = segment->open()) != srs_success) {
        return srs_error_wrap(err, "open segment");
    }
    
    dvr_enabled = true;
    
    return err;
}

void SrsDvrSessionPlan::on_unpublish()
{
    // support multiple publish.
    if (!dvr_enabled) {
        return;
    }
    
    // ignore error.
    srs_error_t err = segment->close();
    if (err != srs_success) {
        srs_warn("ignore flv close error %s", srs_error_desc(err).c_str());
    }
    
    dvr_enabled = false;

    // We should notify the on_dvr, then stop the async.
    // @see https://github.com/ossrs/srs/issues/1601
    SrsDvrPlan::on_unpublish();
}

SrsDvrSegmentPlan::SrsDvrSegmentPlan()
{
    cduration = 0;
    wait_keyframe = false;
}

SrsDvrSegmentPlan::~SrsDvrSegmentPlan()
{
}

srs_error_t SrsDvrSegmentPlan::initialize(SrsOriginHub* h, SrsDvrSegmenter* s, SrsRequest* r)
{
    srs_error_t err = srs_success;
    
    if ((err = SrsDvrPlan::initialize(h, s, r)) != srs_success) {
        return srs_error_wrap(err, "segment plan");
    }
    
    wait_keyframe = _srs_config->get_dvr_wait_keyframe(req->vhost);
    
    cduration = _srs_config->get_dvr_duration(req->vhost);
    
    return srs_success;
}

srs_error_t SrsDvrSegmentPlan::on_publish()
{
    srs_error_t err = srs_success;

    if ((err = SrsDvrPlan::on_publish()) != srs_success) {
        return err;
    }
    
    // support multiple publish.
    if (dvr_enabled) {
        return err;
    }
    
    if (!_srs_config->get_dvr_enabled(req->vhost)) {
        return err;
    }
    
    if ((err = segment->close()) != srs_success) {
        return srs_error_wrap(err, "segment close");
    }
    
    if ((err = segment->open()) != srs_success) {
        return srs_error_wrap(err, "segment open");
    }
    
    dvr_enabled = true;
    
    return err;
}

void SrsDvrSegmentPlan::on_unpublish()
{
    srs_error_t err = srs_success;

    if ((err = segment->close()) != srs_success) {
        srs_warn("ignore err %s", srs_error_desc(err).c_str());
        srs_freep(err);
    }

    dvr_enabled = false;

    // We should notify the on_dvr, then stop the async.
    // @see https://github.com/ossrs/srs/issues/1601
    SrsDvrPlan::on_unpublish();
}

srs_error_t SrsDvrSegmentPlan::on_audio(SrsSharedPtrMessage* shared_audio, SrsFormat* format)
{
    srs_error_t err = srs_success;
    
    if ((err = update_duration(shared_audio)) != srs_success) {
        return srs_error_wrap(err, "update duration");
    }
    
    if ((err = SrsDvrPlan::on_audio(shared_audio, format)) != srs_success) {
        return srs_error_wrap(err, "consume audio");
    }
    
    return err;
}

srs_error_t SrsDvrSegmentPlan::on_video(SrsSharedPtrMessage* shared_video, SrsFormat* format)
{
    srs_error_t err = srs_success;
    
    if ((err = update_duration(shared_video)) != srs_success) {
        return srs_error_wrap(err, "update duration");
    }
    
    if ((err = SrsDvrPlan::on_video(shared_video, format)) != srs_success) {
        return srs_error_wrap(err, "consume video");
    }
    
    return err;
}

srs_error_t SrsDvrSegmentPlan::update_duration(SrsSharedPtrMessage* msg)
{
    srs_error_t err = srs_success;
    
    srs_assert(segment);
    
    // ignore if duration ok.
    SrsFragment* fragment = segment->current();
    if (cduration <= 0 || fragment->duration() < cduration) {
        return err;
    }
    
    // when wait keyframe, ignore if no frame arrived.
    // @see https://github.com/ossrs/srs/issues/177
    if (wait_keyframe) {
        if (!msg->is_video()) {
            return err;
        }
        
        char* payload = msg->payload;
        int size = msg->size;
        bool is_key_frame = SrsFlvVideo::h264(payload, size) && SrsFlvVideo::keyframe(payload, size) && !SrsFlvVideo::sh(payload, size);
        if (!is_key_frame) {
            return err;
        }
    }
    
    // reap segment
    if ((err = segment->close()) != srs_success) {
        return srs_error_wrap(err, "segment close");
    }
    
    // open new flv file
    if ((err = segment->open()) != srs_success) {
        return srs_error_wrap(err, "segment open");
    }
    
    // update sequence header
    if ((err = hub->on_dvr_request_sh()) != srs_success) {
        return srs_error_wrap(err, "request sh");
    }
    
    return err;
}

srs_error_t SrsDvrSegmentPlan::on_reload_vhost_dvr(string vhost)
{
    srs_error_t err = srs_success;
    
    if (req->vhost != vhost) {
        return err;
    }
    
    wait_keyframe = _srs_config->get_dvr_wait_keyframe(req->vhost);
    
    cduration = _srs_config->get_dvr_duration(req->vhost);
    
    return err;
}

SrsDvr::SrsDvr()
{
    hub = NULL;
    plan = NULL;
    req = NULL;
    actived = false;
    
    _srs_config->subscribe(this);
}

SrsDvr::~SrsDvr()
{
    _srs_config->unsubscribe(this);
    
    srs_freep(plan);
}

srs_error_t SrsDvr::initialize(SrsOriginHub* h, SrsRequest* r)
{
    srs_error_t err = srs_success;
    
    req = r;
    hub = h;
    
    SrsConfDirective* conf = _srs_config->get_dvr_apply(r->vhost);
    actived = srs_config_apply_filter(conf, r);
    
    srs_freep(plan);
    if ((err = SrsDvrPlan::create_plan(r->vhost, &plan)) != srs_success) {
        return srs_error_wrap(err, "create plan");
    }
    
    std::string path = _srs_config->get_dvr_path(r->vhost);
    SrsDvrSegmenter* segmenter = NULL;
    if (srs_string_ends_with(path, ".mp4")) {
        segmenter = new SrsDvrMp4Segmenter();
    } else {
        segmenter = new SrsDvrFlvSegmenter();
    }
    
    if ((err = plan->initialize(hub, segmenter, r)) != srs_success) {
        return srs_error_wrap(err, "plan initialize");
    }
    
    return err;
}

srs_error_t SrsDvr::on_publish()
{
    srs_error_t err = srs_success;
    
    // the dvr for this stream is not actived.
    if (!actived) {
        return err;
    }
    
    if ((err = plan->on_publish()) != srs_success) {
        return srs_error_wrap(err, "publish");
    }
    
    return err;
}

void SrsDvr::on_unpublish()
{
    plan->on_unpublish();
}

srs_error_t SrsDvr::on_meta_data(SrsSharedPtrMessage* metadata)
{
    srs_error_t err = srs_success;
    
    // the dvr for this stream is not actived.
    if (!actived) {
        return err;
    }
    
    if ((err = plan->on_meta_data(metadata)) != srs_success) {
        return srs_error_wrap(err, "metadata");
    }
    
    return err;
}

srs_error_t SrsDvr::on_audio(SrsSharedPtrMessage* shared_audio, SrsFormat* format)
{
    // the dvr for this stream is not actived.
    if (!actived) {
        return srs_success;
    }
    
    return plan->on_audio(shared_audio, format);
}

srs_error_t SrsDvr::on_video(SrsSharedPtrMessage* shared_video, SrsFormat* format)
{
    // the dvr for this stream is not actived.
    if (!actived) {
        return srs_success;
    }
    
    return plan->on_video(shared_video, format);
}

srs_error_t SrsDvr::on_reload_vhost_dvr_apply(string vhost)
{
    srs_error_t err = srs_success;
    
    SrsConfDirective* conf = _srs_config->get_dvr_apply(req->vhost);
    bool v = srs_config_apply_filter(conf, req);
    
    // the apply changed, republish the dvr.
    if (v == actived) {
        return err;
    }
    actived = v;
    
    on_unpublish();
    if (!actived) {
        return err;
    }
    
    if ((err = on_publish()) != srs_success) {
        return srs_error_wrap(err, "on publish");
    }
    if ((err = hub->on_dvr_request_sh()) != srs_success) {
        return srs_error_wrap(err, "request sh");
    }
    
    return err;
}