/* This file is part of Telegram Desktop, the official desktop version of Telegram messaging app, see https://telegram.org Telegram Desktop 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 3 of the License, or (at your option) any later version. It 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. Full license: https://github.com/telegramdesktop/tdesktop/blob/master/LICENSE Copyright (c) 2014 John Preston, https://desktop.telegram.org */ #include "stdafx.h" #include "fileuploader.h" FileUploader::FileUploader() : sentSize(0), uploading(0) { memset(sentSizes, 0, sizeof(sentSizes)); nextTimer.setSingleShot(true); connect(&nextTimer, SIGNAL(timeout()), this, SLOT(sendNext())); killSessionsTimer.setSingleShot(true); connect(&killSessionsTimer, SIGNAL(timeout()), this, SLOT(killSessions())); } void FileUploader::uploadMedia(MsgId msgId, const ReadyLocalMedia &media) { if (media.type == ToPreparePhoto) { App::feedPhoto(media.photo, media.photoThumbs); } else if (media.type == ToPrepareDocument) { DocumentData *document; if (media.photoThumbs.isEmpty()) { document = App::feedDocument(media.document); } else { document = App::feedDocument(media.document, media.photoThumbs.begin().value()); } document->status = FileUploading; if (!media.file.isEmpty()) { document->location = FileLocation(StorageFilePartial, media.file); } } else if (media.type == ToPrepareAudio) { AudioData *audio = App::feedAudio(media.audio); audio->status = FileUploading; audio->data = media.data; } queue.insert(msgId, File(media)); sendNext(); } void FileUploader::currentFailed() { Queue::iterator j = queue.find(uploading); if (j != queue.end()) { if (j->media.type == ToPreparePhoto) { emit photoFailed(j.key()); } else if (j->media.type == ToPrepareDocument) { DocumentData *doc = App::document(j->media.id); if (doc->status == FileUploading) { doc->status = FileFailed; } emit documentFailed(j.key()); } else if (j->media.type == ToPrepareAudio) { AudioData *audio = App::audio(j->media.id); if (audio->status == FileUploading) { audio->status = FileFailed; } emit audioFailed(j.key()); } queue.erase(j); } requestsSent.clear(); docRequestsSent.clear(); dcMap.clear(); uploading = 0; sentSize = 0; for (int i = 0; i < MTPUploadSessionsCount; ++i) { sentSizes[i] = 0; } sendNext(); } void FileUploader::killSessions() { for (int i = 0; i < MTPUploadSessionsCount; ++i) { MTP::stopSession(MTP::upl[i]); } } void FileUploader::sendNext() { if (sentSize >= MaxUploadFileParallelSize) return; bool killing = killSessionsTimer.isActive(); if (queue.isEmpty()) { if (!killing) { killSessionsTimer.start(MTPAckSendWaiting + MTPKillFileSessionTimeout); } return; } if (killing) { killSessionsTimer.stop(); } Queue::iterator i = uploading ? queue.find(uploading) : queue.begin(); if (!uploading) { uploading = i.key(); } else if (i == queue.end()) { i = queue.begin(); uploading = i.key(); } int todc = 0; for (int dc = 1; dc < MTPUploadSessionsCount; ++dc) { if (sentSizes[dc] < sentSizes[todc]) { todc = dc; } } if (i->media.parts.isEmpty()) { if (i->docSentParts >= i->docPartsCount) { if (requestsSent.isEmpty() && docRequestsSent.isEmpty()) { if (i->media.type == ToPreparePhoto) { emit photoReady(uploading, MTP_inputFile(MTP_long(i->media.id), MTP_int(i->partsCount), MTP_string(i->media.filename), MTP_string(i->media.jpeg_md5))); } else if (i->media.type == ToPrepareDocument) { QByteArray docMd5(32, Qt::Uninitialized); hashMd5Hex(i->md5Hash.result(), docMd5.data()); MTPInputFile doc = (i->docSize > UseBigFilesFrom) ? MTP_inputFileBig(MTP_long(i->media.id), MTP_int(i->docPartsCount), MTP_string(i->media.filename)) : MTP_inputFile(MTP_long(i->media.id), MTP_int(i->docPartsCount), MTP_string(i->media.filename), MTP_string(docMd5)); if (i->partsCount) { emit thumbDocumentReady(uploading, doc, MTP_inputFile(MTP_long(i->media.thumbId), MTP_int(i->partsCount), MTP_string(qsl("thumb.") + i->media.thumbExt), MTP_string(i->media.jpeg_md5))); } else { emit documentReady(uploading, doc); } } else if (i->media.type == ToPrepareAudio) { QByteArray audioMd5(32, Qt::Uninitialized); hashMd5Hex(i->md5Hash.result(), audioMd5.data()); MTPInputFile audio = (i->docSize > UseBigFilesFrom) ? MTP_inputFileBig(MTP_long(i->media.id), MTP_int(i->docPartsCount), MTP_string(i->media.filename)) : MTP_inputFile(MTP_long(i->media.id), MTP_int(i->docPartsCount), MTP_string(i->media.filename), MTP_string(audioMd5)); emit audioReady(uploading, audio); } queue.remove(uploading); uploading = 0; sendNext(); } return; } QByteArray toSend; if (i->media.data.isEmpty()) { if (!i->docFile) { i->docFile.reset(new QFile(i->media.file)); if (!i->docFile->open(QIODevice::ReadOnly)) { currentFailed(); return; } } toSend = i->docFile->read(i->docPartSize); if (i->docSize <= UseBigFilesFrom) { i->md5Hash.feed(toSend.constData(), toSend.size()); } } else { toSend = i->media.data.mid(i->docSentParts * i->docPartSize, i->docPartSize); if ((i->media.type == ToPrepareDocument || i->media.type == ToPrepareAudio) && i->docSentParts <= UseBigFilesFrom) { i->md5Hash.feed(toSend.constData(), toSend.size()); } } if (toSend.size() > i->docPartSize || (toSend.size() < i->docPartSize && i->docSentParts + 1 != i->docPartsCount)) { currentFailed(); return; } mtpRequestId requestId; if (i->docSize > UseBigFilesFrom) { requestId = MTP::send(MTPupload_SaveBigFilePart(MTP_long(i->media.id), MTP_int(i->docSentParts), MTP_int(i->docPartsCount), MTP_string(toSend)), rpcDone(&FileUploader::partLoaded), rpcFail(&FileUploader::partFailed), MTP::upl[todc]); } else { requestId = MTP::send(MTPupload_SaveFilePart(MTP_long(i->media.id), MTP_int(i->docSentParts), MTP_string(toSend)), rpcDone(&FileUploader::partLoaded), rpcFail(&FileUploader::partFailed), MTP::upl[todc]); } docRequestsSent.insert(requestId, i->docSentParts); dcMap.insert(requestId, todc); sentSize += i->docPartSize; sentSizes[todc] += i->docPartSize; i->docSentParts++; } else { LocalFileParts::iterator part = i->media.parts.begin(); mtpRequestId requestId = MTP::send(MTPupload_SaveFilePart(MTP_long(i->media.thumbId), MTP_int(part.key()), MTP_string(part.value())), rpcDone(&FileUploader::partLoaded), rpcFail(&FileUploader::partFailed), MTP::upl[todc]); requestsSent.insert(requestId, part.value()); dcMap.insert(requestId, todc); sentSize += part.value().size(); sentSizes[todc] += part.value().size(); i->media.parts.erase(part); } nextTimer.start(UploadRequestInterval); } void FileUploader::cancel(MsgId msgId) { uploaded.remove(msgId); if (uploading == msgId) { currentFailed(); } else { queue.remove(msgId); } } void FileUploader::confirm(MsgId msgId) { } void FileUploader::clear() { uploaded.clear(); queue.clear(); for (QMap::const_iterator i = requestsSent.cbegin(), e = requestsSent.cend(); i != e; ++i) { MTP::cancel(i.key()); } requestsSent.clear(); for (QMap::const_iterator i = docRequestsSent.cbegin(), e = docRequestsSent.cend(); i != e; ++i) { MTP::cancel(i.key()); } docRequestsSent.clear(); dcMap.clear(); sentSize = 0; for (int32 i = 0; i < MTPUploadSessionsCount; ++i) { MTP::stopSession(MTP::upl[i]); sentSizes[i] = 0; } killSessionsTimer.stop(); } void FileUploader::partLoaded(const MTPBool &result, mtpRequestId requestId) { QMap::iterator j = docRequestsSent.end(); QMap::iterator i = requestsSent.find(requestId); if (i == requestsSent.cend()) { j = docRequestsSent.find(requestId); } if (i != requestsSent.cend() || j != docRequestsSent.cend()) { if (!result.v) { // failed to upload current file currentFailed(); return; } else { QMap::iterator dcIt = dcMap.find(requestId); if (dcIt == dcMap.cend()) { // must not happen currentFailed(); return; } int32 dc = dcIt.value(); dcMap.erase(dcIt); Queue::const_iterator k = queue.constFind(uploading); if (i != requestsSent.cend()) { sentSize -= i.value().size(); sentSizes[dc] -= i.value().size(); requestsSent.erase(i); } else { sentSize -= k->docPartSize; sentSizes[dc] -= k->docPartSize; docRequestsSent.erase(j); } if (k->media.type == ToPreparePhoto) { emit photoProgress(k.key()); } else if (k->media.type == ToPrepareDocument) { DocumentData *doc = App::document(k->media.id); if (doc->status == FileUploading) { doc->uploadOffset = (k->docSentParts - docRequestsSent.size()) * k->docPartSize; if (doc->uploadOffset > doc->size) { doc->uploadOffset = doc->size; } } emit documentProgress(k.key()); } else if (k->media.type == ToPrepareAudio) { AudioData *audio = App::audio(k->media.id); if (audio->status == FileUploading) { audio->uploadOffset = (k->docSentParts - docRequestsSent.size()) * k->docPartSize; if (audio->uploadOffset > audio->size) { audio->uploadOffset = audio->size; } } } } } sendNext(); } bool FileUploader::partFailed(const RPCError &error, mtpRequestId requestId) { if (error.type().startsWith(qsl("FLOOD_WAIT_"))) return false; if (requestsSent.constFind(requestId) != requestsSent.cend() || docRequestsSent.constFind(requestId) != docRequestsSent.cend()) { // failed to upload current file currentFailed(); } sendNext(); return true; }