Streams.cpp 14 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
/*
 * Streams.cpp
 *****************************************************************************
 * Copyright (C) 2014 - VideoLAN authors
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as published
 * by the Free Software Foundation; either version 2.1 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin Street, Fifth Floor, Boston MA 02110-1301, USA.
 *****************************************************************************/
#include "Streams.hpp"
21
#include "StreamsType.hpp"
22
#include "http/HTTPConnection.hpp"
23 24 25
#include "http/HTTPConnectionManager.h"
#include "http/Chunk.h"
#include "logic/AbstractAdaptationLogic.h"
26
#include "SegmentTracker.hpp"
27 28
#include <vlc_stream.h>
#include <vlc_demux.h>
29

30
using namespace adaptative;
31 32 33
using namespace adaptative::http;
using namespace adaptative::logic;

34 35
Stream::Stream(const std::string &mime)
{
36
    init(mimeToType(mime), mimeToFormat(mime));
37 38
}

39
Stream::Stream(const StreamType type, const StreamFormat format)
40
{
41
    init(type, format);
42 43
}

44
void Stream::init(const StreamType type_, const StreamFormat format_)
45 46
{
    type = type_;
47
    format = format_;
48
    output = NULL;
49
    adaptationLogic = NULL;
50 51
    currentChunk = NULL;
    eof = false;
52
    segmentTracker = NULL;
53 54 55 56 57 58 59
}

Stream::~Stream()
{
    delete currentChunk;
    delete adaptationLogic;
    delete output;
60
    delete segmentTracker;
61 62
}

63
StreamType Stream::mimeToType(const std::string &mime)
64
{
65
    StreamType mimetype;
66
    if (!mime.compare(0, 6, "video/"))
67
        mimetype = StreamType::VIDEO;
68
    else if (!mime.compare(0, 6, "audio/"))
69
        mimetype = StreamType::AUDIO;
70
    else if (!mime.compare(0, 12, "application/"))
71
        mimetype = StreamType::APPLICATION;
72
    else /* unknown of unsupported */
73
        mimetype = StreamType::UNKNOWN;
74 75 76
    return mimetype;
}

77
StreamFormat Stream::mimeToFormat(const std::string &mime)
78
{
79
    StreamFormat format = StreamFormat::UNSUPPORTED;
80 81 82 83 84
    std::string::size_type pos = mime.find("/");
    if(pos != std::string::npos)
    {
        std::string tail = mime.substr(pos + 1);
        if(tail == "mp4")
85
            format = StreamFormat::MP4;
86
        else if (tail == "mp2t")
87
            format = StreamFormat::MPEG2TS;
88 89 90 91
    }
    return format;
}

92 93
void Stream::create(demux_t *demux, AbstractAdaptationLogic *logic,
                    SegmentTracker *tracker, AbstractStreamOutputFactory &factory)
94
{
95
    output = factory.create(demux, format);
96 97
    adaptationLogic = logic;
    segmentTracker = tracker;
98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119
}

bool Stream::isEOF() const
{
    return false;
}

mtime_t Stream::getPCR() const
{
    return output->getPCR();
}

int Stream::getGroup() const
{
    return output->getGroup();
}

int Stream::esCount() const
{
    return output->esCount();
}

120 121 122 123
bool Stream::operator ==(const Stream &stream) const
{
    return stream.type == type;
}
124 125 126 127 128

Chunk * Stream::getChunk()
{
    if (currentChunk == NULL)
    {
129
        currentChunk = segmentTracker->getNextChunk(type, output->switchAllowed());
130 131 132 133 134 135
        if (currentChunk == NULL)
            eof = true;
    }
    return currentChunk;
}

136 137 138 139 140
bool Stream::seekAble() const
{
    return (output && output->seekAble());
}

141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156
Stream::status Stream::demux(HTTPConnectionManager *connManager, mtime_t nz_deadline)
{
    if(nz_deadline + VLC_TS_0 > output->getPCR()) /* not already demuxed */
    {
        /* need to read, demuxer still buffering, ... */
        if(read(connManager) <= 0)
            return Stream::status_eof;

        if(nz_deadline + VLC_TS_0 > output->getPCR()) /* need to read more */
            return Stream::status_buffering;
    }

    output->sendToDecoder(nz_deadline);
    return Stream::status_demuxed;
}

157 158 159 160 161 162 163 164 165 166 167 168 169 170
size_t Stream::read(HTTPConnectionManager *connManager)
{
    Chunk *chunk = getChunk();
    if(!chunk)
        return 0;

    if(!chunk->getConnection())
    {
       if(!connManager->connectChunk(chunk))
        return 0;
    }

    size_t readsize = 0;

171
    /* New chunk, do query */
172 173
    if(chunk->getBytesRead() == 0)
    {
174 175 176 177 178 179 180
        if(chunk->getConnection()->query(chunk->getPath()) != VLC_SUCCESS)
        {
            chunk->getConnection()->releaseChunk();
            currentChunk = NULL;
            delete chunk;
            return 0;
        }
181 182
    }

183 184 185 186
    /* Because we don't know Chunk size at start, we need to get size
       from content length */
    readsize = chunk->getBytesToRead();
    if (readsize > 32768)
187 188 189 190 191 192 193 194 195 196
        readsize = 32768;

    block_t *block = block_Alloc(readsize);
    if(!block)
        return 0;

    mtime_t time = mdate();
    ssize_t ret = chunk->getConnection()->read(block->p_buffer, readsize);
    time = mdate() - time;

197
    if(ret < 0)
198 199 200 201 202 203 204 205 206 207 208 209
    {
        block_Release(block);
        chunk->getConnection()->releaseChunk();
        currentChunk = NULL;
        delete chunk;
        return 0;
    }
    else
    {
        block->i_buffer = (size_t)ret;

        adaptationLogic->updateDownloadRate(block->i_buffer, time);
210
        chunk->onDownload(&block);
211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226

        if (chunk->getBytesToRead() == 0)
        {
            chunk->getConnection()->releaseChunk();
            currentChunk = NULL;
            delete chunk;
        }
    }

    readsize = block->i_buffer;

    output->pushBlock(block);

    return readsize;
}

227 228
bool Stream::setPosition(mtime_t time, bool tryonly)
{
229
    bool ret = segmentTracker->setPosition(time, output->reinitsOnSeek(), tryonly);
230
    if(!tryonly && ret)
231
    {
232
        output->setPosition(time);
233 234 235 236 237 238 239 240 241 242
        if(output->reinitsOnSeek())
        {
            if(currentChunk)
            {
                currentChunk->getConnection()->releaseChunk();
                delete currentChunk;
            }
            currentChunk = NULL;
        }
    }
243 244 245
    return ret;
}

246 247 248 249 250
mtime_t Stream::getPosition() const
{
    return segmentTracker->getSegmentStart();
}

251 252 253 254 255
void Stream::prune()
{
    segmentTracker->pruneFromCurrent();
}

256 257 258 259
AbstractStreamOutput::AbstractStreamOutput(demux_t *demux)
{
    realdemux = demux;
    pcr = VLC_TS_0;
260
    group = 0;
261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280
}

AbstractStreamOutput::~AbstractStreamOutput()
{
}

mtime_t AbstractStreamOutput::getPCR() const
{
    return pcr;
}

int AbstractStreamOutput::getGroup() const
{
    return group;
}

BaseStreamOutput::BaseStreamOutput(demux_t *demux, const std::string &name) :
    AbstractStreamOutput(demux)
{
    this->name = name;
281
    seekable = true;
282
    restarting = false;
283
    demuxstream = NULL;
284 285 286 287 288

    fakeesout = new es_out_t;
    if (!fakeesout)
        throw VLC_ENOMEM;

289 290
    vlc_mutex_init(&lock);

291 292 293 294 295 296
    fakeesout->pf_add = esOutAdd;
    fakeesout->pf_control = esOutControl;
    fakeesout->pf_del = esOutDel;
    fakeesout->pf_destroy = esOutDestroy;
    fakeesout->pf_send = esOutSend;
    fakeesout->p_sys = (es_out_sys_t*) this;
297

298 299
    demuxstream = stream_DemuxNew(realdemux, name.c_str(), fakeesout);
    if(!demuxstream)
300
        throw VLC_EGENERIC;
301 302
}

303
BaseStreamOutput::~BaseStreamOutput()
304 305 306
{
    if (demuxstream)
        stream_Delete(demuxstream);
307 308 309 310 311

    /* shouldn't be any */
    std::list<Demuxed *>::const_iterator it;
    for(it=queues.begin(); it!=queues.end();++it)
        delete *it;
312 313 314

    delete fakeesout;
    vlc_mutex_destroy(&lock);
315 316
}

317
int BaseStreamOutput::esCount() const
318
{
319
    return queues.size();
320 321
}

322
void BaseStreamOutput::pushBlock(block_t *block)
323 324 325 326
{
    stream_DemuxSend(demuxstream, block);
}

327
bool BaseStreamOutput::seekAble() const
328
{
329 330
    bool b_canswitch = switchAllowed();
    return (demuxstream && seekable && b_canswitch);
331 332
}

333
void BaseStreamOutput::setPosition(mtime_t nztime)
334
{
335
    vlc_mutex_lock(&lock);
336 337 338 339 340 341 342
    std::list<Demuxed *>::const_iterator it;
    for(it=queues.begin(); it!=queues.end();++it)
    {
        Demuxed *pair = *it;
        if(pair->p_queue && pair->p_queue->i_dts > VLC_TS_0 + nztime)
            pair->drop();
    }
343
    pcr = VLC_TS_INVALID;
344
    vlc_mutex_unlock(&lock);
345 346
    es_out_Control(realdemux->out, ES_OUT_SET_NEXT_DISPLAY_TIME,
                   VLC_TS_0 + nztime);
347 348
    if(reinitsOnSeek())
        restart();
349 350
}

351 352 353 354 355 356 357
bool BaseStreamOutput::restart()
{
    stream_t *newdemuxstream = stream_DemuxNew(realdemux, name.c_str(), fakeesout);
    if(!newdemuxstream)
        return false;

    vlc_mutex_lock(&lock);
358
    restarting = true;
359 360 361 362 363 364 365 366 367 368 369 370
    stream_t *olddemuxstream = demuxstream;
    demuxstream = newdemuxstream;
    vlc_mutex_unlock(&lock);

    if(olddemuxstream)
        stream_Delete(olddemuxstream);

    return true;
}

bool BaseStreamOutput::reinitsOnSeek() const
{
371 372 373 374 375 376 377 378 379 380
    return true;
}

bool BaseStreamOutput::switchAllowed() const
{
    bool b_allowed;
    vlc_mutex_lock(const_cast<vlc_mutex_t *>(&lock));
    b_allowed = !restarting;
    vlc_mutex_unlock(const_cast<vlc_mutex_t *>(&lock));
    return b_allowed;
381 382 383
}

void BaseStreamOutput::sendToDecoder(mtime_t nzdeadline)
384
{
385
    vlc_mutex_lock(&lock);
386 387 388 389
    sendToDecoderUnlocked(nzdeadline);
    vlc_mutex_unlock(&lock);
}

390
void BaseStreamOutput::sendToDecoderUnlocked(mtime_t nzdeadline)
391
{
392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409
    std::list<Demuxed *>::const_iterator it;
    for(it=queues.begin(); it!=queues.end();++it)
    {
        Demuxed *pair = *it;
        while(pair->p_queue && pair->p_queue->i_dts <= VLC_TS_0 + nzdeadline)
        {
            block_t *p_block = pair->p_queue;
            pair->p_queue = pair->p_queue->p_next;
            p_block->p_next = NULL;

            if(pair->pp_queue_last == &p_block->p_next)
                pair->pp_queue_last = &pair->p_queue;

            realdemux->out->pf_send(realdemux->out, pair->es_id, p_block);
        }
    }
}

410
BaseStreamOutput::Demuxed::Demuxed(es_out_id_t *id, const es_format_t *fmt)
411 412 413
{
    p_queue = NULL;
    pp_queue_last = &p_queue;
414 415 416
    es_id = id;
    es_format_Init(&fmtcpy, UNKNOWN_ES, 0);
    es_format_Copy(&fmtcpy, fmt);
417 418
}

419
BaseStreamOutput::Demuxed::~Demuxed()
420
{
421
    es_format_Clean(&fmtcpy);
422 423 424
    drop();
}

425
void BaseStreamOutput::Demuxed::drop()
426 427 428 429 430 431
{
    block_ChainRelease(p_queue);
    p_queue = NULL;
    pp_queue_last = &p_queue;
}

432
/* Static callbacks */
433
es_out_id_t * BaseStreamOutput::esOutAdd(es_out_t *fakees, const es_format_t *p_fmt)
434
{
435
    BaseStreamOutput *me = (BaseStreamOutput *) fakees->p_sys;
436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470

    es_out_id_t *p_es = NULL;

    vlc_mutex_lock(&me->lock);

    std::list<Demuxed *>::iterator it;
    bool b_hasestorecyle = false;
    for(it=me->queues.begin(); it!=me->queues.end();++it)
    {
        Demuxed *pair = *it;
        b_hasestorecyle |= pair->recycle;

        if( p_es )
            continue;

        if( me->restarting )
        {
            /* If we're recycling from same format */
            if( es_format_IsSimilar(p_fmt, &pair->fmtcpy) &&
                    p_fmt->i_extra == pair->fmtcpy.i_extra &&
                    !memcmp(p_fmt->p_extra, pair->fmtcpy.p_extra, p_fmt->i_extra) )
            {
                msg_Err(me->realdemux, "using recycled");
                pair->recycle = false;
                p_es = pair->es_id;
            }
        }
    }

    if(!b_hasestorecyle)
    {
        me->restarting = false;
    }

    if(!p_es)
471
    {
472 473
        p_es = me->realdemux->out->pf_add(me->realdemux->out, p_fmt);
        if(p_es)
474
        {
475 476 477
            Demuxed *pair = new (std::nothrow) Demuxed(p_es, p_fmt);
            if(pair)
                me->queues.push_back(pair);
478 479
        }
    }
480 481
    vlc_mutex_unlock(&me->lock);

482
    return p_es;
483 484
}

485
int BaseStreamOutput::esOutSend(es_out_t *fakees, es_out_id_t *p_es, block_t *p_block)
486
{
487
    BaseStreamOutput *me = (BaseStreamOutput *) fakees->p_sys;
488
    vlc_mutex_lock(&me->lock);
489 490 491 492 493 494 495 496 497 498
    std::list<Demuxed *>::const_iterator it;
    for(it=me->queues.begin(); it!=me->queues.end();++it)
    {
        Demuxed *pair = *it;
        if(pair->es_id == p_es)
        {
            block_ChainLastAppend(&pair->pp_queue_last, p_block);
            break;
        }
    }
499
    vlc_mutex_unlock(&me->lock);
500
    return VLC_SUCCESS;
501 502
}

503
void BaseStreamOutput::esOutDel(es_out_t *fakees, es_out_id_t *p_es)
504
{
505
    BaseStreamOutput *me = (BaseStreamOutput *) fakees->p_sys;
506
    vlc_mutex_lock(&me->lock);
507 508 509 510 511
    std::list<Demuxed *>::iterator it;
    for(it=me->queues.begin(); it!=me->queues.end();++it)
    {
        if((*it)->es_id == p_es)
        {
512
            me->sendToDecoderUnlocked(INT64_MAX - VLC_TS_0);
513 514 515 516 517 518 519 520 521 522 523 524
            break;
        }
    }

    if(it != me->queues.end())
    {
        if(me->restarting)
        {
            (*it)->recycle = true;
        }
        else
        {
525 526 527 528
            delete *it;
            me->queues.erase(it);
        }
    }
529 530 531 532

    if(!me->restarting)
        me->realdemux->out->pf_del(me->realdemux->out, p_es);

533
    vlc_mutex_unlock(&me->lock);
534 535
}

536
int BaseStreamOutput::esOutControl(es_out_t *fakees, int i_query, va_list args)
537
{
538
    BaseStreamOutput *me = (BaseStreamOutput *) fakees->p_sys;
539 540 541 542 543 544 545 546 547 548 549
    if (i_query == ES_OUT_SET_PCR )
    {
        me->pcr = (int64_t)va_arg( args, int64_t );
        return VLC_SUCCESS;
    }
    else if( i_query == ES_OUT_SET_GROUP_PCR )
    {
        me->group = (int) va_arg( args, int );
        me->pcr = (int64_t)va_arg( args, int64_t );
        return VLC_SUCCESS;
    }
550 551 552 553 554 555 556 557 558 559 560
    else if( i_query == ES_OUT_GET_ES_STATE )
    {
        va_arg( args, es_out_id_t * );
        bool *pb = va_arg( args, bool * );
        *pb = true;
        return VLC_SUCCESS;
    }
    else if( me->restarting )
    {
        return VLC_EGENERIC;
    }
561 562 563 564

    return me->realdemux->out->pf_control(me->realdemux->out, i_query, args);
}

565
void BaseStreamOutput::esOutDestroy(es_out_t *fakees)
566
{
567
    BaseStreamOutput *me = (BaseStreamOutput *) fakees->p_sys;
568 569 570 571
    me->realdemux->out->pf_destroy(me->realdemux->out);
}
/* !Static callbacks */

572 573 574 575 576
AbstractStreamOutput *DefaultStreamOutputFactory::create(demux_t *demux, int format) const
{
    switch(format)
    {
        case StreamFormat::MP4:
577
            return new BaseStreamOutput(demux, "mp4");
578 579

        case StreamFormat::MPEG2TS:
580
            return new BaseStreamOutput(demux, "ts");
581 582 583 584 585 586 587 588

        default:
            throw VLC_EBADVAR;
            break;
    }
    return NULL;
}