/*
* libdatachannel streamer example
* Copyright (c) 2020 Filip Klembara (in2core)
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version 2
* 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; If not, see .
*/
#include "stream.hpp"
#include "helpers.hpp"
void StreamSource::stop() {
sampleTime_us = 0;
sample = {};
}
StreamSource::~StreamSource() {
stop();
}
Stream::Stream(std::shared_ptr video, std::shared_ptr audio): std::enable_shared_from_this(), video(video), audio(audio) { }
Stream::~Stream() {
stop();
}
std::pair, Stream::StreamSourceType> Stream::unsafePrepareForSample() {
std::shared_ptr ss;
StreamSourceType sst;
uint64_t nextTime;
if (audio->getSampleTime_us() < video->getSampleTime_us()) {
ss = audio;
sst = StreamSourceType::Audio;
nextTime = audio->getSampleTime_us();
} else {
ss = video;
sst = StreamSourceType::Video;
nextTime = video->getSampleTime_us();
}
auto currentTime = currentTimeInMicroSeconds();
auto elapsed = currentTime - startTime;
if (nextTime > elapsed) {
auto waitTime = nextTime - elapsed;
mutex.unlock();
usleep(waitTime);
mutex.lock();
}
return {ss, sst};
}
void Stream::sendSample() {
std::lock_guard lock(mutex);
if (!isRunning) {
return;
}
auto ssSST = unsafePrepareForSample();
auto ss = ssSST.first;
auto sst = ssSST.second;
auto sample = ss->getSample();
sampleHandler(sst, ss->getSampleTime_us(), sample);
ss->loadNextSample();
dispatchQueue.dispatch([this]() {
this->sendSample();
});
}
void Stream::onSample(std::function handler) {
sampleHandler = handler;
}
void Stream::start() {
std::lock_guard lock(mutex);
if (isRunning) {
return;
}
_isRunning = true;
startTime = currentTimeInMicroSeconds();
audio->start();
video->start();
dispatchQueue.dispatch([this]() {
this->sendSample();
});
}
void Stream::stop() {
std::lock_guard lock(mutex);
if (!isRunning) {
return;
}
_isRunning = false;
dispatchQueue.removePending();
audio->stop();
video->stop();
};