qftrans/cworker.cpp

/***************************************************************************
 *   Copyright (C) 2008 by Oliver Groß   *
 *   z.o.gross@gmx.de   *
 *                                                                         *
 *   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, write to the                         *
 *   Free Software Foundation, Inc.,                                       *
 *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.             *
 ***************************************************************************/
#include "cworker.h"
#include <QTcpSocket>
#include <QByteArray>
#include <QFile>
#include <QDir>
#include <QThread>

namespace qftrans {
	CWorker::CWorker(QTcpSocket * socket, QObject * parent) : QObject(parent), m_Socket(socket), m_IdCounter(0) {
		connect(m_Socket, SIGNAL(error(QAbstractSocket::SocketError)), this, SLOT(error(QAbstractSocket::SocketError)));
		connect(m_Socket, SIGNAL(bytesWritten(qint64)), this, SLOT(writeData()), Qt::QueuedConnection);
		connect(m_Socket, SIGNAL(readyRead()), this, SLOT(readData()), Qt::QueuedConnection);
		qDebug() << "worker created";
	}
	
	CWorker::~CWorker() {
		qDebug() << "worker removed";
	}
	
	//TODO addUpload
	void CWorker::addUpload(TransferData * transfer) {
		transfer->status = TS_PENDING;
		
		if (!m_Uploads.contains(transfer)) {
			qDebug() << "adding upload";
			transfer->id = m_IdCounter;
			m_IdCounter++;
			m_Uploads.enqueue(transfer);
		}
		else
			qDebug() << "(re)adding upload";
		
		if (m_Socket->bytesToWrite() == 0) {
			qDebug() << "executing upload";
			writeData();
		}
	}
	//TODO removeUpload
	void CWorker::removeUpload(TransferData * transfer) {
		if (m_Uploads.contains(transfer) && transfer->status > TS_PENDING) {
			transfer->status = TS_CLOSING;
			if (m_Socket->bytesToWrite() == 0) {
				qDebug() << "executing closedown";
				writeData();
			}
		}
	}
	//TODO ? startUpload
// 	void CWorker::startUpload(TransferData * transfer) {
// 		if (m_Uploads.contains(transfer) && transfer->status != TS_PENDING)
// 			transfer->status = TS_PENDING;
// 		
// 		if (m_Socket->bytesToWrite() == 0) {
// 			qDebug() << "executing upload";
// 			writeData();
// 		}
// 	}
	
	void CWorker::startUploadData(quint32 id) {
		if (m_Uploads.isEmpty())
			return;
		
		TransferData * transfer = m_Uploads.head(); /*m_UploadDict.key(id);*/
		
		if (transfer->id != id) {
			qDebug() << "got invalid id:" << id << "... expected:" << transfer->id;
			return;
		}
		
		qDebug() << "scheduling upload for transfer";
		transfer->status = TS_TRANSFERING;
		if (m_Socket->bytesToWrite() == 0) {
			qDebug() << "executing upload";
			writeData();
		}
	}
	
	void CWorker::sendAcknowledge(quint32 id) {
		QByteArray block;
		QDataStream out(&block, QIODevice::WriteOnly);
		out.setVersion(QDataStream::Qt_4_0);
		
		TransferHeader header;
		
		header.type   = HT_ACK;
		header.id     = id;
		header.length = 0;
		
		out.writeRawData((const char *)(&header), sizeof(TransferHeader));
		
		qDebug() << "sending HT_ACK" << id;
		
		m_Socket->write(block);
		qDebug() << "sent HT_ACK" << id;
	}
	
	//TODO CTransferThread::removeDownload
	void CWorker::removeDownload(TransferData * /*transfer*/) {
		
	}
	
	void CWorker::setDestinationDir(QString & value) {
		m_DestinationDir = value;
	}
	
	void CWorker::writeData() {
		if (m_Uploads.isEmpty()) {
// 		if (m_UploadDict.isEmpty()) {
			qDebug() << "no more uploads.";
			return;
		}
		
		TransferData * transfer = m_Uploads.head();
		
		if (transfer->status == TS_PENDING)
			transfer->status = TS_ANNOUNCING;
		
		QByteArray block;
		QDataStream out(&block, QIODevice::ReadWrite);
		out.setVersion(QDataStream::Qt_4_0);
		
		TransferHeader header;
		switch (transfer->status) {
		case TS_ANNOUNCING:
			{
				qDebug() << "announcing" << transfer->id;
				if (transfer->transfered)
					transfer->transfered = 0;
				
				qint64 oldPosition = block.size();
				
				//fake write filename
				out << transfer->fileName;
				out.device()->seek(oldPosition);
				
				// ANNOUNCE header with size and filename
				header.type   = HT_ANNOUNCE;
				header.id     = transfer->id;
				header.length = quint32(block.size() - oldPosition) + sizeof(qint64);
				
				out.writeRawData((const char *)(&header), sizeof(TransferHeader));
				
				qDebug() << "sending size" << transfer->size;
				out << transfer->size;
				
				qDebug() << "sending filename" << transfer->fileName;
				out << transfer->fileName;
				
				transfer->status = TS_WAITING;
			}
			break;
		case TS_TRANSFERING:
			{
				header.type   = HT_DATA;
				header.id     = transfer->id;
				if (transfer->transfered + DATA_BLOCK_SIZE < transfer->size)
					header.length = DATA_BLOCK_SIZE;
				else
					header.length = transfer->size - transfer->transfered;
				
				out.writeRawData((const char *)(&header), sizeof(TransferHeader));
				
				char * buffer = new char[header.length];
				
				QFile file(transfer->fileDir + QDir::separator() + transfer->fileName);
				file.open(QIODevice::ReadOnly);
				QDataStream in(&file);
				if (in.skipRawData(transfer->transfered) == -1) {
					qDebug() << "filename" << file.fileName();
					qDebug() << "skip error";
					qDebug() << "file size" << transfer->size;
					qDebug() << "transfered bytes" << transfer->transfered;
				}
				if (in.readRawData(buffer, header.length) == -1) {
					qDebug() << "Read error";
					qDebug() << "buffer size" << header.length;
					qDebug() << "file size" << transfer->size;
					qDebug() << "transfered bytes" << transfer->transfered;
				}
					
				out.writeRawData((const char *)(buffer), header.length);
				transfer->transfered += header.length;
				delete[] buffer;
				
				if (transfer->transfered >= transfer->size)
					transfer->status = TS_CLOSING;
				
				if (transfer->transfered > transfer->size) {
					qDebug() << "overread?";
					qDebug() << "file size" << transfer->size;
					qDebug() << "transfered bytes" << transfer->transfered;
				}
				
				emit uploadUpdated(transfer);
			}
			break;
		case TS_CLOSING:
			header.type   = HT_CLOSE;
			header.id     = transfer->id;
			header.length = 0;
			
			out.writeRawData((const char *)(&header), sizeof(TransferHeader));
			
			transfer->status = TS_FINISHED;
			
			m_Uploads.dequeue();
			emit uploadUpdated(transfer);
			break;
		default:
			break;
		}
		
		if (!block.isEmpty())
			m_Socket->write(block);
	}
	
	void CWorker::readHeader(QDataStream & in) {
		if (m_Socket->bytesAvailable() >= (qint64)(sizeof(TransferHeader))) {
			in.readRawData((char *)(&m_LastHeader), sizeof(TransferHeader));
			m_LastHeaderValid = (qstrcmp(m_LastHeader.descriptor, DESCRIPTOR_STRING) == 0);
			
			if (!m_LastHeaderValid) {
				qDebug() << "got invalid header data";
				qDebug() << "available Bytes" << m_Socket->bytesAvailable();
			}
		}
		else
			m_LastHeaderValid = false;
	}
	
	void CWorker::readData() {
		QDataStream in(m_Socket);
		in.setVersion(QDataStream::Qt_4_0);
		
		if (!m_LastHeaderValid)
			readHeader(in);
		
		TransferData * target = NULL;
		
		while (m_LastHeaderValid && m_Socket->bytesAvailable() >= m_LastHeader.length) {
// 			qDebug() << "available Bytes" << m_Socket->bytesAvailable() << "length" << m_LastHeader.length;
			
			//look for existing id
			if (m_LastHeader.type == HT_ACK) {
				qDebug() << "got HT_ACK .. scheduling upload to start";
				startUploadData(m_LastHeader.id);
			}
			else {
				if (m_DownloadDict.contains(m_LastHeader.id)) {
					if (m_LastHeader.type == HT_CLOSE) {
						//finished or canceled by peer
						
						qDebug() << "download closed py peer";
						TransferData * closed = m_DownloadDict.take(m_LastHeader.id);
						closed->status = TS_FINISHED;
						emit downloadRemoved(closed);
					}
					else
						//assign target
						target = m_DownloadDict[m_LastHeader.id];
				}
				//id not in dict
				else if (m_LastHeader.type == HT_ANNOUNCE) {
					//new file announced
					qDebug() << "download announced";
					TransferData * newFile = new TransferData;
					m_DownloadDict.insert(m_LastHeader.id, newFile);
					newFile->transfered = 0;
					newFile->size = 0;
					newFile->localFile = false;
					newFile->status = TS_WAITING;
					newFile->fileDir = m_DestinationDir;
					target = newFile;
				}
				
				//check valid target and data available
				if (target && m_LastHeader.length) {
					switch (m_LastHeader.type) {
					case HT_ANNOUNCE:
						qDebug() << "reading size..";
						in >> target->size;
						qDebug() << "got size:" << target->size;
						qDebug() << "reading filename..";
						in >> target->fileName;
						qDebug() << "got filename:" << target->fileName;
						emit downloadUpdated(target);
						qDebug() << "trying to send HT_ACK";
						
						//TODO better do this async
						sendAcknowledge(m_LastHeader.id);
						break;
					case HT_DATA:
						{
// 							qDebug() << "got data" << target->transfered << '/' << target->size;
							QFile file(target->fileDir + QDir::separator() + target->fileName);
							if (target->status == TS_WAITING) {
								target->status = TS_TRANSFERING;
								emit downloadAdded(target);
								file.open(QIODevice::WriteOnly | QIODevice::Truncate);
							}
							else if (target->status == TS_TRANSFERING)
								file.open(QIODevice::WriteOnly | QIODevice::Append);
							else {
								in.skipRawData(m_LastHeader.length);
								qDebug() << "transfer not ready to recieve data. status is" << target->status;
								qDebug() << "skipping" << m_LastHeader.length << "bytes of data";
								break;
							}
							
							char * buffer = new char[m_LastHeader.length];
							if (in.readRawData(buffer, m_LastHeader.length) == -1) {
								qDebug() << "error reading from socket";
								qDebug() << "length" << m_LastHeader.length;
								qDebug() << "available Bytes" << m_Socket->bytesAvailable();
							}
							
							QDataStream out(&file);
							out.writeRawData((const char *)(buffer), m_LastHeader.length);
							target->transfered += m_LastHeader.length;
							delete[] buffer;
							
							emit downloadUpdated(target);
						}
						break;
					default:
						in.skipRawData(m_LastHeader.length);
						qDebug() << "unknown header";
						qDebug() << "skipping" << m_LastHeader.length << "bytes of data";
						break;
					}
				}
			}
			
			readHeader(in);
		}
	}
	
	void CWorker::writePing() {
		qDebug() << "write ping... thread id" << QThread::currentThreadId();
		if (m_IdCounter > 10)
			return;
		
		QByteArray block;
		QDataStream out(&block, QIODevice::WriteOnly);
		out.setVersion(QDataStream::Qt_4_0);
		
		TransferHeader header;
		
		header.type   = HT_ACK;
		header.id     = m_IdCounter++;
		header.length = 0;
		
		out.writeRawData((const char *)(&header), sizeof(TransferHeader));
		
		qDebug() << "sending Ping..." << header.id;
		if (m_Socket->write(block) == -1)
			qDebug() << "error";
		else
			qDebug() << "done";
	}
	
	void CWorker::readPing() {
		qDebug() << "read ping... thread id" << QThread::currentThreadId();
		
		QDataStream in(m_Socket);
		in.setVersion(QDataStream::Qt_4_0);
		
		do {
			if (m_Socket->bytesAvailable() >= (qint64)(sizeof(TransferHeader))) {
				in.readRawData((char *)(&m_LastHeader), sizeof(TransferHeader));
				m_LastHeaderValid = (qstrcmp(m_LastHeader.descriptor, DESCRIPTOR_STRING) == 0);
				
				if (m_LastHeaderValid)
					qDebug() << "got valid ping" << m_LastHeader.id;
				else
					qDebug() << "got invalid ping" << m_LastHeader.id;
				
				writePing();
			}
			else
				m_LastHeaderValid = false;
		}
		while (m_LastHeaderValid);
	}
	
	void CWorker::error(QAbstractSocket::SocketError) {
		qDebug() << m_Socket->errorString();
	}
}