Notes

/*
* Copyright (c) 2013 Citrix Online LLC
* All Rights Reserved Worldwide.
*
* THIS PROGRAM IS CONFIDENTIAL AND PROPRIETARY TO CITRIX ONLINE
* AND CONSTITUTES A VALUABLE TRADE SECRET. Any unauthorized use,
* reproduction, modification, or disclosure of this program is
* strictly prohibited. Any use of this program by an authorized
* licensee is strictly subject to the terms and conditions,
* including confidentiality obligations, set forth in the applicable
* License and Co-Branding Agreement between Citrix Online LLC
* the licensee.
*/

#include "FileTransfer.h"

#include <VTTPrecompiled.h>
#include <VTTTools.h>
#include <stdio.h>
#include <fstream>

#include <rtc/trace/Trace.h>

DECLARE_TRACING("FileTransmitter");

namespace {
const int message_size = (16384 - 1);
const int progress_interval_ms = 1000;
};

namespace citrix { namespace rtc {

FileTransmitter::FileTransmitter(::rtc::avtp::CDataChannel* pDataChannel):
m_pDataChannel(pDataChannel),
m_stopThread(false),
m_totalBytes(0),
m_transferredBytes(0),
m_startTime(0)
{
m_pTransmitBuffer = (unsigned char*)malloc(message_size + 1);
if (m_pTransmitBuffer == NULL) {
strace(2, "ERROR cant malloc buffer for data msg...");
assert(false);
}

m_progressTimer.SetPeriod(progress_interval_ms);

m_thread.start("FileTransmitter", boost::bind(&FileTransmitter::threadFunction, this));
}

FileTransmitter::~FileTransmitter()
{
removeCallback();
m_stopThread = true;
m_bytesFreeEvent.Signal(); // missuse of signal, just to wake up thread for termination
m_thread.waitForCompletion();
free(m_pTransmitBuffer);
}

void FileTransmitter::threadFunction()
{
CWaitForMultipleObjects waitObjects;
waitObjects.AddObject(&m_bytesFreeEvent);
waitObjects.AddObject(&m_startSending);
waitObjects.AddObject(&m_progressTimer);

int64_t lastProgress = 0;

for(;;) {
waitObjects.WaitFor();

CLockMutex lock(&m_mutex);

if ((waitObjects.GetSignaledObject()==&m_progressTimer) && !m_fileQueue.empty() && !(m_totalBytes == 0)) {
if(boost::shared_ptr<FileTransferProgressCallback> callback = m_progressCallback.lock())
callback->fileTransferProgressUpdate(getProgress(), getBytesPerSecond(), m_pDataChannel);
lastProgress = VTTCommonTools::GetTimeInMsSec();
}

if ((waitObjects.GetSignaledObject()==&m_bytesFreeEvent) ||
(waitObjects.GetSignaledObject()==&m_startSending)) {

doSending(lastProgress);

if (m_fileQueue.empty() &&
!((m_totalBytes==0) && (m_transferredBytes==0) )) {
if(boost::shared_ptr<FileTransferProgressCallback> callback = m_progressCallback.lock())
callback->fileTransferProgressUpdate(getProgress(), getBytesPerSecond(), m_pDataChannel);
m_totalBytes = 0;
m_transferredBytes = 0;
}
}

if (m_fileQueue.empty() && m_stopThread)
break;

}
}

int FileTransmitter::doSending(int64_t timeOfLastProgressUpdate)
{
// iterate through all files and send them (unless we get stopped because the buffers are full)
//for (std::vector<AudioCommon::FileQueueEntryPtr>::iterator it = m_fileQueue.begin(); it != m_fileQueue.end(); ++it) {
std::vector<boost::shared_ptr<FileQueueEntry> >::iterator it = m_fileQueue.begin();
while (it != m_fileQueue.end()) {
std::fstream* file = &((*it)->file);
// open file, this means we are starting to send a file now
if (!file->is_open()) {
file->open((*it)->fileName.c_str(), std::ios::binary|std::ios::in);

if (!file->is_open()) {
strace(2, "ERROR cant open file for send to server, skipping it...");
it = m_fileQueue.erase(it);
continue;
}
}

long sendStatus = ::rtc::avtp::TRA_OK;

while (!file->eof()) {
size_t previousPos = file->tellg();
file->read((char *)(m_pTransmitBuffer + 1), message_size);
size_t bytesRead = file->gcount();

// pack buffered data into a DataMsg
m_pTransmitBuffer[0] = 0;
::rtc::avtp::DataMsg msg = {bytesRead + 1, bytesRead + 1, m_pTransmitBuffer};

// send the message
sendStatus = m_pDataChannel->Send(msg);

if ((sendStatus == ::rtc::avtp::TRA_OUT_OF_MEMORY) || (sendStatus == ::rtc::avtp::TRA_NOT_ENOUGHT_BUFFERS)) {
// we will be sending the next time, have to seek back to previous position
if (file->eof()) // if we read over the end of file, reset fail/eof bits
file->clear();
file->seekg(previousPos, std::ios_base::beg);
break;
}

if (sendStatus != ::rtc::avtp::TRA_OK)
assert(false);

m_transferredBytes += bytesRead;

if (VTTCommonTools::GetTimeInMsSec() > timeOfLastProgressUpdate + progress_interval_ms) {
if(boost::shared_ptr<FileTransferProgressCallback> callback = m_progressCallback.lock())
callback->fileTransferProgressUpdate(getProgress(), getBytesPerSecond(), m_pDataChannel);
timeOfLastProgressUpdate = VTTCommonTools::GetTimeInMsSec();
}
}
// BREAK if file is not done yet and we just got thrown out by datachannel->send
if ((sendStatus == ::rtc::avtp::TRA_OUT_OF_MEMORY) || (sendStatus == ::rtc::avtp::TRA_NOT_ENOUGHT_BUFFERS))
break;

// FILE IS DONE send empty message so receiver knows
m_pTransmitBuffer[0] = 1;
::rtc::avtp::DataMsg msg = {1, 1, m_pTransmitBuffer};

// send the message
sendStatus = m_pDataChannel->Send(msg);

if ((sendStatus == ::rtc::avtp::TRA_OUT_OF_MEMORY) || (sendStatus == ::rtc::avtp::TRA_NOT_ENOUGHT_BUFFERS)) {
break; // we will be sending the empty packet the next time, just break
}

if (sendStatus != ::rtc::avtp::TRA_OK)
assert(false);

it = m_fileQueue.erase(it);
}

return 0;
}

void FileTransmitter::MsgSent(::rtc::avtp::CDataChannel* pDataChannel, std::size_t BytesFree, std::size_t bytesSent)
{
if (!m_fileQueue.empty())
m_bytesFreeEvent.Signal();
}

void FileTransmitter::MsgReceived(::rtc::avtp::CDataChannel* pDataChannel, ::rtc::avtp::DataMsg &Msg )
{}

int FileTransmitter::addFile(size_t fileSize, std::string fileName)
{
if (m_stopThread)
return -1;
CLockMutex lock(&m_mutex);
m_fileQueue.push_back(boost::shared_ptr<FileQueueEntry>(new FileQueueEntry(fileSize, fileName)));
m_startSending.Signal();
// progress/speed metering
if (m_totalBytes == 0)
m_startTime = VTTCommonTools::GetTimeInMsSec();
m_totalBytes += fileSize;
return 0;
}

bool FileTransmitter::isTransmitting()
{
return !m_fileQueue.empty();
}

double FileTransmitter::getProgress()
{
if ((m_totalBytes == 0) || (m_transferredBytes == m_totalBytes))
return 1;

return (double) m_transferredBytes / m_totalBytes;
}

double FileTransmitter::getBytesPerSecond()
{
double pastTime = std::max<double>(1, (VTTCommonTools::GetTimeInMsSec() - m_startTime) / 1000.0);

if (pastTime > 0)
return (double) m_transferredBytes / pastTime;

return 0;
}

void FileTransmitter::setCallback( boost::weak_ptr<FileTransferProgressCallback> callback )
{
m_progressCallback = callback;
}

void FileTransmitter::removeCallback()
{
m_progressCallback.reset();
}

size_t FileTransmitter::getFileSize(const std::string &filename)
{
std::ifstream in(filename.c_str(), std::ifstream::in | std::ifstream::binary);
in.seekg(0, std::ifstream::end);
return in.tellg();
}

} } //namespace citrix { namespace rtc {