Commit a5d7067e authored by François Cartegnie's avatar François Cartegnie 🤞

demux: adaptative: add chunk backend/source instea of direct reads

because we'll need fake chunks
parent 8d8a63bc
......@@ -54,7 +54,7 @@ void SegmentTracker::resetCounter()
prevRepresentation = NULL;
}
SegmentChunk * SegmentTracker::getNextChunk(bool switch_allowed)
SegmentChunk * SegmentTracker::getNextChunk(bool switch_allowed, HTTPConnectionManager *connManager)
{
BaseRepresentation *rep;
ISegment *segment;
......@@ -92,7 +92,7 @@ SegmentChunk * SegmentTracker::getNextChunk(bool switch_allowed)
init_sent = true;
segment = rep->getSegment(BaseRepresentation::INFOTYPE_INIT);
if(segment)
return segment->toChunk(count, rep);
return segment->toChunk(count, rep, connManager);
}
if(!index_sent)
......@@ -100,7 +100,7 @@ SegmentChunk * SegmentTracker::getNextChunk(bool switch_allowed)
index_sent = true;
segment = rep->getSegment(BaseRepresentation::INFOTYPE_INDEX);
if(segment)
return segment->toChunk(count, rep);
return segment->toChunk(count, rep, connManager);
}
bool b_gap = false;
......@@ -120,7 +120,7 @@ SegmentChunk * SegmentTracker::getNextChunk(bool switch_allowed)
/* stop initializing after 1st chunk */
initializing = false;
SegmentChunk *chunk = segment->toChunk(count, rep);
SegmentChunk *chunk = segment->toChunk(count, rep, connManager);
if(chunk)
count++;
......
......@@ -29,6 +29,11 @@
namespace adaptative
{
namespace http
{
class HTTPConnectionManager;
}
namespace logic
{
class AbstractAdaptationLogic;
......@@ -44,6 +49,7 @@ namespace adaptative
using namespace playlist;
using namespace logic;
using namespace http;
class SegmentTrackerListenerInterface
{
......@@ -63,7 +69,7 @@ namespace adaptative
void setAdaptationLogic(AbstractAdaptationLogic *);
void resetCounter();
SegmentChunk* getNextChunk(bool);
SegmentChunk* getNextChunk(bool, HTTPConnectionManager *);
bool setPositionByTime(mtime_t, bool, bool);
void setPositionByNumber(uint64_t, bool);
mtime_t getSegmentStart() const;
......
......@@ -150,7 +150,7 @@ SegmentChunk * AbstractStream::getChunk()
disabled = true;
return NULL;
}
currentChunk = segmentTracker->getNextChunk(!fakeesout->restarting());
currentChunk = segmentTracker->getNextChunk(!fakeesout->restarting(), connManager);
if (currentChunk == NULL)
eof = true;
}
......@@ -294,7 +294,7 @@ AbstractStream::status AbstractStream::demux(mtime_t nz_deadline, bool send)
return AbstractStream::status_demuxed;
}
block_t * AbstractStream::readNextBlock(size_t)
block_t * AbstractStream::readNextBlock(size_t toread)
{
SegmentChunk *chunk = getChunk();
if(!chunk)
......@@ -320,62 +320,22 @@ block_t * AbstractStream::readNextBlock(size_t)
return NULL;
}
if(!chunk->getConnection())
{
if(!connManager->connectChunk(chunk))
return NULL;
}
size_t readsize = 0;
bool b_segment_head_chunk = false;
/* New chunk, do query */
if(chunk->getBytesRead() == 0)
{
if(chunk->getConnection()->query(chunk->getPath(), chunk->getBytesRange()) != VLC_SUCCESS)
{
currentChunk = NULL;
delete chunk;
return NULL;
}
chunk->setLength(chunk->getConnection()->getContentLength());
b_segment_head_chunk = true;
}
/* Because we don't know Chunk size at start, we need to get size
from content length */
readsize = chunk->getBytesToRead();
if (readsize > 32768)
readsize = 32768;
const bool b_segment_head_chunk = (chunk->getBytesRead() == 0);
block_t *block = block_Alloc(readsize);
if(!block)
return NULL;
mtime_t time = mdate();
ssize_t ret = chunk->getConnection()->read(block->p_buffer, readsize);
time = mdate() - time;
if(ret < 0)
mtime_t time;
block_t *block = chunk->read(toread, &time);
if(block == NULL)
{
block_Release(block);
currentChunk = NULL;
delete chunk;
return NULL;
}
else
{
chunk->setBytesRead(chunk->getBytesRead() + ret);
block->i_buffer = (size_t)ret;
adaptationLogic->updateDownloadRate(block->i_buffer, time);
chunk->onDownload(&block);
if (chunk->getBytesToRead() == 0)
{
currentChunk = NULL;
delete chunk;
}
adaptationLogic->updateDownloadRate(block->i_buffer, time);
if (chunk->getBytesToRead() == 0)
{
currentChunk = NULL;
delete chunk;
}
block = checkBlock(block, b_segment_head_chunk);
......
......@@ -26,17 +26,111 @@
#endif
#include "Chunk.h"
#include "HTTPConnection.hpp"
#include "HTTPConnectionManager.h"
#include <vlc_common.h>
#include <vlc_url.h>
#include <vlc_block.h>
using namespace adaptative::http;
Chunk::Chunk (const std::string& url) :
port (0),
length (0),
bytesRead (0),
connection (NULL)
AbstractChunkSource::AbstractChunkSource()
{
parentChunk = NULL;
}
AbstractChunkSource::~AbstractChunkSource()
{
}
void AbstractChunkSource::setParentChunk(AbstractChunk *parent)
{
parentChunk = parent;
}
AbstractChunk::AbstractChunk(AbstractChunkSource *source_)
{
length = 0;
bytesRead = 0;
source = source_;
source->setParentChunk(this);
}
AbstractChunk::~AbstractChunk()
{
delete source;
}
void AbstractChunk::setLength(size_t length)
{
this->length = length;
}
size_t AbstractChunk::getBytesRead() const
{
return this->bytesRead;
}
void AbstractChunk::setBytesRead(size_t bytes)
{
this->bytesRead = bytes;
}
size_t AbstractChunk::getBytesToRead() const
{
return length - bytesRead;
}
void AbstractChunk::setBytesRange(const BytesRange &range)
{
bytesRange = range;
if(bytesRange.isValid() && bytesRange.getEndByte())
setLength(bytesRange.getEndByte() - bytesRange.getStartByte());
}
const BytesRange & AbstractChunk::getBytesRange() const
{
return bytesRange;
}
block_t * AbstractChunk::read(size_t sizehint, mtime_t *time)
{
if(!source)
return NULL;
*time = mdate();
block_t *block = source->read(sizehint);
*time = mdate() - *time;
if(block)
{
setBytesRead(getBytesRead() + block->i_buffer);
onDownload(&block);
}
return block;
}
HTTPChunkSource::HTTPChunkSource(const std::string& url, HTTPConnectionManager *manager) :
AbstractChunkSource(),
port (0),
connection (NULL),
connManager (manager)
{
if(!init(url))
throw VLC_EGENERIC;
}
HTTPChunkSource::~HTTPChunkSource()
{
if(connection)
connection->setUsed(false);
}
bool HTTPChunkSource::init(const std::string &url)
{
this->url = url;
......@@ -47,7 +141,7 @@ Chunk::Chunk (const std::string& url) :
}
if(scheme != "http" && scheme != "https")
throw VLC_EGENERIC;
return false;
vlc_url_t url_components;
vlc_UrlParse(&url_components, url.c_str());
......@@ -67,73 +161,63 @@ Chunk::Chunk (const std::string& url) :
vlc_UrlClean(&url_components);
if(path.empty() || hostname.empty())
throw VLC_EGENERIC;
}
return false;
Chunk::~Chunk()
{
if(connection)
connection->setUsed(false);
return true;
}
const BytesRange & Chunk::getBytesRange() const
block_t * HTTPChunkSource::read(size_t maxread)
{
return bytesRange;
}
if(!connManager || !parentChunk)
return NULL;
const std::string& Chunk::getUrl () const
{
return url;
}
if(!connection)
{
connection = connManager->getConnection(scheme, hostname, port);
if(!connection)
return NULL;
}
void Chunk::setBytesRange(const BytesRange &range)
{
bytesRange = range;
if(bytesRange.isValid() && bytesRange.getEndByte())
length = bytesRange.getEndByte() - bytesRange.getStartByte();
}
if(parentChunk->getBytesRead() == 0)
{
if( connection->query(path, parentChunk->getBytesRange()) != VLC_SUCCESS )
return NULL;
parentChunk->setLength(connection->getContentLength());
}
const std::string& Chunk::getScheme () const
{
return scheme;
}
if( parentChunk->getBytesToRead() == 0 )
return NULL;
const std::string& Chunk::getHostname () const
{
return hostname;
}
const std::string& Chunk::getPath () const
{
return this->path;
}
int Chunk::getPort () const
{
return this->port;
}
/* Because we don't know Chunk size at start, we need to get size
from content length */
size_t readsize = parentChunk->getBytesToRead();
if (readsize > maxread)
readsize = maxread;
void Chunk::setLength (size_t length)
{
this->length = length;
}
size_t Chunk::getBytesRead () const
{
return this->bytesRead;
}
void Chunk::setBytesRead (size_t bytes)
{
this->bytesRead = bytes;
}
block_t *block = block_Alloc(readsize);
if(!block)
return NULL;
size_t Chunk::getBytesToRead () const
{
return length - bytesRead;
ssize_t ret = connection->read(block->p_buffer, readsize);
if(ret < 0)
{
block_Release(block);
return NULL;
}
block->i_buffer = (size_t)ret;
return block;
}
HTTPConnection* Chunk::getConnection () const
HTTPChunk::HTTPChunk(const std::string &url, HTTPConnectionManager *manager):
AbstractChunk(new HTTPChunkSource(url, manager))
{
return this->connection;
}
void Chunk::setConnection (HTTPConnection *connection)
HTTPChunk::~HTTPChunk()
{
this->connection = connection;
}
......@@ -37,40 +37,73 @@ namespace adaptative
namespace http
{
class HTTPConnection;
class HTTPConnectionManager;
class AbstractChunk;
class Chunk
class AbstractChunkSource
{
public:
Chunk (const std::string &url);
virtual ~Chunk ();
AbstractChunkSource();
virtual ~AbstractChunkSource();
virtual block_t * read(size_t) = 0;
void setParentChunk(AbstractChunk *);
protected:
AbstractChunk *parentChunk;
};
class AbstractChunk
{
public:
virtual ~AbstractChunk();
const BytesRange & getBytesRange () const;
const std::string& getUrl () const;
const std::string& getScheme () const;
const std::string& getHostname () const;
const std::string& getPath () const;
int getPort () const;
size_t getBytesRead () const;
size_t getBytesToRead () const;
HTTPConnection* getConnection () const;
const BytesRange & getBytesRange () const;
void setConnection (HTTPConnection *connection);
void setBytesRead (size_t bytes);
void setLength (size_t length);
void setBytesRange (const BytesRange &);
virtual void onDownload (block_t **) {}
virtual block_t * read (size_t, mtime_t *);
virtual void onDownload (block_t **) = 0;
protected:
AbstractChunk(AbstractChunkSource *);
AbstractChunkSource *source;
private:
size_t length;
size_t bytesRead;
BytesRange bytesRange;
};
class HTTPChunkSource : public AbstractChunkSource
{
public:
HTTPChunkSource(const std::string &url, HTTPConnectionManager *);
virtual ~HTTPChunkSource();
virtual block_t * read(size_t); /* impl */
private:
std::string url;
std::string scheme;
std::string path;
std::string hostname;
int port;
BytesRange bytesRange;
size_t length;
size_t bytesRead;
HTTPConnection *connection;
bool init(const std::string &);
std::string url;
std::string scheme;
std::string path;
std::string hostname;
uint16_t port;
HTTPConnection *connection;
HTTPConnectionManager *connManager;
};
class HTTPChunk : public AbstractChunk
{
public:
HTTPChunk(const std::string &url, HTTPConnectionManager *);
virtual ~HTTPChunk();
virtual void onDownload (block_t **) {} /* impl */
};
}
......
......@@ -82,6 +82,9 @@ int HTTPConnection::query(const std::string &path, const BytesRange &range)
{
queryOk = false;
msg_Dbg(stream, "Retrieving ://%s:%u%s @%zu", hostname.c_str(), port, path.c_str(),
range.isValid() ? range.getStartByte() : 0);
if(!connected() && ( hostname.empty() || !connect(hostname, port) ))
return VLC_EGENERIC;
......
......@@ -27,8 +27,8 @@
#include "HTTPConnectionManager.h"
#include "HTTPConnection.hpp"
#include "Chunk.h"
#include "Sockets.hpp"
#include <vlc_url.h>
using namespace adaptative::http;
......@@ -66,42 +66,37 @@ HTTPConnection * HTTPConnectionManager::getConnection(const std::string &hostnam
return NULL;
}
bool HTTPConnectionManager::connectChunk(Chunk *chunk)
HTTPConnection * HTTPConnectionManager::getConnection(const std::string &scheme,
const std::string &hostname,
uint16_t port)
{
if(chunk == NULL)
return false;
if(chunk->getConnection())
return true;
if((scheme != "http" && scheme != "https") || hostname.empty())
return NULL;
msg_Dbg(stream, "Retrieving %s @%zu", chunk->getUrl().c_str(),
chunk->getBytesRange().isValid() ? chunk->getBytesRange().getStartByte() : 0);
const int sockettype = (chunk->getScheme() == "https") ? TLSSocket::TLS : Socket::REGULAR;
HTTPConnection *conn = getConnection(chunk->getHostname(), chunk->getPort(), sockettype);
const int sockettype = (scheme == "https") ? TLSSocket::TLS : Socket::REGULAR;
HTTPConnection *conn = getConnection(hostname, port, sockettype);
if(!conn)
{
Socket *socket = (sockettype == TLSSocket::TLS) ? new (std::nothrow) TLSSocket()
: new (std::nothrow) Socket();
if(!socket)
return false;
return NULL;
/* disable pipelined tls until we have ticket/resume session support */
conn = new (std::nothrow) HTTPConnection(stream, socket, sockettype != TLSSocket::TLS);
if(!conn)
{
delete socket;
return false;
return NULL;
}
connectionPool.push_back(conn);
if (!conn->connect(chunk->getHostname(), chunk->getPort()))
if (!conn->connect(hostname, port))
{
return false;
return NULL;
}
}
conn->setUsed(true);
chunk->setConnection(conn);
return true;
return conn;
}
......@@ -38,7 +38,6 @@ namespace adaptative
namespace http
{
class HTTPConnection;
class Chunk;
class HTTPConnectionManager
{
......@@ -48,7 +47,9 @@ namespace adaptative
void closeAllConnections ();
void releaseAllConnections ();
bool connectChunk (Chunk *chunk);
HTTPConnection * getConnection(const std::string &scheme,
const std::string &hostname,
uint16_t port);
private:
std::vector<HTTPConnection *> connectionPool;
......
......@@ -29,6 +29,7 @@
namespace adaptative
{
namespace playlist
{
class BasePeriod;
......@@ -73,6 +74,7 @@ namespace adaptative
std::vector<std::string> baseUrls;
std::string playlistUrl;
std::string type;
};
}
}
......