Merge branch 'work48' into DEV

LibServer/Systems/sendsystem.cpp

@@ -2,26 +2,53 @@
 
 SendSystem::SendSystem(QObject *parent) : QObject(parent)
 {
-    isSendStopped = false; 
+    qDebug() << "SendSystem init thread ID " << QThread::currentThreadId();
+
+    client = nullptr;
+    socket = nullptr;
+    dataParser = nullptr;
+
+    globalMutex = nullptr;
+    waitCondition = nullptr;
+
+    buffer = nullptr;
+
+    type = TypeClientAutorization::TYPE_GUI;
+
+    isSendStopped = false;
+    flWaitWriten = false;
+
+    fileSize = 0;
+    countSend = 0;
+
+    buffer = new char[sendFileBlockSize];
 }
 
+SendSystem::~SendSystem()
+{
+    delete buffer;
+}
+
+
 void SendSystem::initialize(DataParser *dataParser,QMutex *globalMutex)
 {
     qDebug() << "SendSystem::initialize thread ID " << QThread::currentThreadId();
 
     this->dataParser = dataParser;
-    //connect(this,&SendSystem::sigSendXMLmessage,dataParser->ClientAnswer(),&ClientAnswerParser::message,Qt::AutoConnection); //сигнал не используется
-    connect(this,&SendSystem::sigSendNotify,dataParser->ClientAnswer(),&ClientAnswerParser::notify,Qt::DirectConnection); //потому что возвращает значение
-    //connect(this,&SendSystem::sigSendVersion,dataParser->ClientAnswer(),&ClientAnswerParser::currentVersion,Qt::AutoConnection); //сигнал не используется
+    this->globalMutex = globalMutex;
 
-    mutex = globalMutex;
+    connect(this,&SendSystem::sigSendNotify,dataParser->ClientAnswer(),&ClientAnswerParser::notify,Qt::DirectConnection); //потому что возвращает значение
 }
 
 void SendSystem::setClient(Client *client,QTcpSocket *socket)
 {
-    this->socket = socket;
-    this->type = client->getClientType();
+    qDebug() << "SendSystem::setClient thread ID " << QThread::currentThreadId();
+
+    this->socket = socket;    
     this->client = client;
+
+    this->type = client->getClientType();
+
     isSendStopped = false;
 }
 
@@ -35,50 +62,239 @@ void SendSystem::sendNotify(QString notify)
 
 void SendSystem::sendFileBlock(QString path)
 {
-    QFile file(path);
-    QFileInfo fileInfo(file);
+    //qDebug() << "SendSystem::sendFileBlock path: " << path;
 
-    if(isSendStopped)
+    if(getIsSendStopped())
     {   //Поведение на случай отключения клиента
-        file.close();
-        Logger::instance().log("UNLOCK STOP MUTEX : " + client->getLogin(),LogLevel::ERROR);
+        Logger::instance().log("Client: " + client->getLogin() + " isSendStopped", LogLevel::ERROR);
         return;
     }
 
-    QDataStream stream(socket);
-    stream.setVersion(QDataStream::Qt_DefaultCompiledVersion);
+    QFile file(path);
+    QFileInfo fileInfo(file);
 
     fileSize = file.size();
 
     if (fileSize == 0)
     {
-        Logger::instance().log("Client: " + client->getLogin() +  " WARNING! Zero size " + fileInfo.fileName(),LogLevel::ERROR);
-        Logger::instance().log(path,LogLevel::ERROR);
+        Logger::instance().log("Client: " + client->getLogin() +  " ERROR! File zero size " + fileInfo.fileName(), LogLevel::ERROR);
+        Logger::instance().log(path, LogLevel::ERROR);
         return;
     }
 
+    if(!file.open(QFile::ReadOnly))
+    {
+        Logger::instance().log("Client: " + client->getLogin() +  " ERROR! File not open: " + fileInfo.fileName(), LogLevel::ERROR);
+        Logger::instance().log(path, LogLevel::ERROR);
+        return;
+    }
+
+    countSend = 0;
+
+    QDataStream stream(socket);
+    stream.setVersion(QDataStream::Qt_DefaultCompiledVersion);
+
     stream << PacketType::TYPE_FILE; //Отправляем тип блока
     stream << path << fileSize;
 
-    if(file.open(QFile::ReadOnly))
-    {
-        while(!file.atEnd())
-        {
-            QByteArray data = file.read(sendFileBlockSize);
-            stream << data;
-            //socket->waitForBytesWritten(10);
+    waitWrittenData("sendFileBlock:header " + fileInfo.fileName());
 
-            if(socket->state() == QAbstractSocket::UnconnectedState) break;
-            countSend++;
+    while(!file.atEnd())
+    {
+        QByteArray data = file.read(sendFileBlockSize);
+        stream << data;
+
+        waitWrittenData(QString("sendFileBlock:data (size %1) ").arg(data.size()) + fileInfo.fileName());
+
+        if(socket->state() == QAbstractSocket::UnconnectedState)
+        {
+            Logger::instance().log("Client: " + client->getLogin() + " UnconnectedState", LogLevel::ERROR);
+            break;
         }
 
-        //emit sigSendToLogger(Tools::getTime() + " send file " + fileInfo.fileName());
+        countSend++;
     }
 
     file.close();
     countSend = 0;
 }
 
+void SendSystem::sendFileBlock_V2(QString path)
+{
+    QFile file(path);
+    if(!file.open(QIODevice::ReadOnly))
+    {
+        Logger::instance().log("ERROR open file: " + path, LogLevel::ERROR);
+        return;
+    }
+
+    countSend = 0;
+    fileSize = file.size();
+
+    if (fileSize == 0)
+    {
+        Logger::instance().log("Client: " + client->getLogin() +  " WARNING! Zero size " +file.fileName(), LogLevel::ERROR);
+        Logger::instance().log(path, LogLevel::ERROR);
+        return;
+    }
+
+    QString type = QString::number(PacketType::TYPE_FILE);
+    QString fileSizeStr = QString::number(fileSize);
+
+    qDebug() << "SendSystem::sendFileBlock_V2 file " << file.fileName() << " size " << fileSizeStr;
+
+    QDataStream stream(socket);
+    stream.setVersion(QDataStream::Qt_DefaultCompiledVersion);
+
+
+
+    //mutexWriten.lock();
+    //flWaitWriten = true;
+
+    stream << PacketType::TYPE_FILE; //Отправляем тип блока
+    stream << path << fileSize;
+
+    socket->waitForBytesWritten(10);
+
+    /*
+    while(flWaitWriten)
+    {
+        QCoreApplication::processEvents(); // Обеспечиваем обработку сообщений
+    }
+    */
+    //socket->waitForBytesWritten(10);
+
+    //socket->write(type.toUtf8());
+    //socket->write(fileSizeStr.toUtf8());
+
+    connect(this->socket, &QTcpSocket::bytesWritten, this, &SendSystem::slot_BytesWrittenToSocket);
+
+
+    while(true)
+    {
+        if(socket->state() == QAbstractSocket::UnconnectedState)
+            break;
+
+        countSend++;
+
+        qint64 readBytes = file.read(buffer, /*sizeof(buffer)*/sendFileBlockSize);
+        if(readBytes <= 0)
+            break;
+
+       flWaitWriten = true;
+        while(!socket->write(buffer, readBytes))
+        {
+            int i = 0;
+            i++;
+        }
+        while(flWaitWriten)
+        {
+            QCoreApplication::processEvents(); // Обеспечиваем обработку сообщений
+        }
+        /*
+        while(!socket->flush())
+        {
+            int i = 0;
+            i++;
+        }
+        */
+        // Завершаем отправку
+        //socket->flush();
+        //socket->waitForBytesWritten(10);
+    }
+
+    disconnect(this->socket, &QTcpSocket::bytesWritten, this, &SendSystem::slot_BytesWrittenToSocket);
+
+    //socket->flush();
+    /*
+    while(!socket->flush())
+    {
+        int i = 0;
+        i++;
+    }*/
+
+
+    file.close();
+}
+
+void SendSystem::sendFileBlock_V3(QString path)
+{
+    //qDebug() << "SendSystem::sendFileBlock path: " << path;
+
+    if(getIsSendStopped())
+    {   //Поведение на случай отключения клиента
+        Logger::instance().log("Client: " + client->getLogin() + " isSendStopped", LogLevel::ERROR);
+        return;
+    }
+
+    QFile file(path);
+    QFileInfo fileInfo(file);
+
+    fileSize = file.size();
+
+    if (fileSize == 0)
+    {
+        Logger::instance().log("Client: " + client->getLogin() +  " ERROR! File zero size " + fileInfo.fileName(), LogLevel::ERROR);
+        Logger::instance().log(path, LogLevel::ERROR);
+        return;
+    }
+
+    if(!file.open(QFile::ReadOnly))
+    {
+        Logger::instance().log("Client: " + client->getLogin() +  " ERROR! File not open: " + fileInfo.fileName(), LogLevel::ERROR);
+        Logger::instance().log(path, LogLevel::ERROR);
+        return;
+    }
+
+    countSend = 0;
+
+    QDataStream stream(socket);
+    stream.setVersion(QDataStream::Qt_DefaultCompiledVersion);
+
+    stream << PacketType::TYPE_FILE; //Отправляем тип блока
+    stream << path << fileSize;
+
+    waitWrittenData("sendFileBlock:header " + fileInfo.fileName());
+
+    connect(this->socket, &QTcpSocket::bytesWritten, this, &SendSystem::slot_BytesWrittenToSocket);
+
+    while(!file.atEnd())
+    {
+        if(socket->state() == QAbstractSocket::UnconnectedState)
+        {
+            Logger::instance().log("Client: " + client->getLogin() + " UnconnectedState", LogLevel::ERROR);
+            break;
+        }
+
+        qint64 readBytes = file.read(buffer, /*sizeof(buffer)*/sendFileBlockSize);
+        if(readBytes <= 0)
+            break;
+
+       flWaitWriten = true;
+        while(!socket->write(buffer, readBytes))
+        {
+            int i = 0;
+            i++;
+            qCritical() << "socket->write ERROR. size " + QString::number(readBytes);
+        }
+        while(flWaitWriten)
+        {
+            QCoreApplication::processEvents(); // Обеспечиваем обработку сообщений
+            int i = 0;
+            i++;
+        }
+
+        waitWrittenData(QString("sendFileBlock:data (readBytes %1, num %2) ").arg(QString::number(readBytes), QString::number(countSend)) + fileInfo.fileName());
+
+        countSend++;
+    }
+
+    disconnect(this->socket, &QTcpSocket::bytesWritten, this, &SendSystem::slot_BytesWrittenToSocket);
+
+    file.close();
+    countSend = 0;
+}
+
 void SendSystem::sendFileBlockByteArray(QByteArray array, PacketType packetType)
 {
     if(client->getClientType() == TypeClientAutorization::TYPE_QT_CLIENT ||
@@ -99,11 +315,15 @@ void SendSystem::sendFileBlockByteArray(QByteArray array, PacketType packetType)
          stream << packetType; //Отправляем тип блока
          stream << size;
 
+         socket->waitForBytesWritten(10);
+
          while (bytesSended < size)
          {
              QByteArray chunk = array.mid(bytesSended,sendFileBlockSize);
              stream << chunk;
 
+             socket->waitForBytesWritten(10);
+
              bytesSended += chunk.length();
          }
     }
@@ -140,6 +360,11 @@ void SendSystem::sendVersion()
     sendXmlAnswer(data);
 }
 
+void SendSystem::slot_BytesWrittenToSocket(qint64 bytes)
+{
+    flWaitWriten = false;
+}
+
 void SendSystem::sendFileBlockWithRename(QString path, QString newName)
 {
     qDebug() << "SendSystem::sendFileBlockWithRename thread ID " << QThread::currentThreadId();
@@ -162,6 +387,8 @@ void SendSystem::sendFileBlockWithRename(QString path, QString newName)
 
     stream << pathForSend << fileSize;
 
+    socket->waitForBytesWritten(10);
+
     if(isSendStopped) {             //Поведение на случай отключения клиента
 
         file.close();
@@ -169,13 +396,13 @@ void SendSystem::sendFileBlockWithRename(QString path, QString newName)
     }
 
 
-    socket->waitForBytesWritten();
+    //socket->waitForBytesWritten();
 
     if(file.open(QFile::ReadOnly)){
         while(!file.atEnd()){
             QByteArray data = file.read(sendFileBlockSize);
             stream << data;
-            socket->waitForBytesWritten();
+            socket->waitForBytesWritten(10);
             countSend++;
         }
 
@@ -184,8 +411,8 @@ void SendSystem::sendFileBlockWithRename(QString path, QString newName)
 
     file.close();
     countSend = 0;
-    socket->waitForBytesWritten();
-    socket->waitForReadyRead(100);
+    socket->waitForBytesWritten(10);
+    //socket->waitForReadyRead(100);
 
     sendNotify(commandHashCompleteClient);
 }
@@ -197,6 +424,8 @@ void SendSystem::sendFolderBlock(QString path)
 
     stream << PacketType::TYPE_FOLDER;
     stream << path;
+
+    waitWrittenData("sendFolderBlock");
 }
 
 void SendSystem::sendDeleteBlock(QString path)
@@ -206,8 +435,8 @@ void SendSystem::sendDeleteBlock(QString path)
 
     stream << PacketType::TYPE_DELETE;
     stream << path;
-    qDebug() << "DELETE: " + path;
-    socket->waitForReadyRead(10);
+
+    waitWrittenData("sendDeleteBlock");
 }
 
 void SendSystem::sendPacketType(PacketType packetType)
@@ -216,11 +445,11 @@ void SendSystem::sendPacketType(PacketType packetType)
     if (client->getClientType() == TypeClientAutorization::TYPE_QT_CLIENT ||
         client->getClientType() == TypeClientAutorization::TYPE_GUI)
     {
-
         QDataStream stream(socket);
         stream.setVersion(QDataStream::Qt_DefaultCompiledVersion);
         stream << packetType;
-        socket->waitForReadyRead(100);
+
+        waitWrittenData("sendPacketType");
     }
     else
     {
@@ -247,7 +476,8 @@ void SendSystem::sendXmlAnswer(QByteArray array, PacketType packetType)
 
         stream << packetType;
         stream << array;
-        socket->waitForBytesWritten();
+
+        waitWrittenData("sendXmlAnswer");
     }
     else
     {
@@ -269,11 +499,14 @@ void SendSystem::sendNeedUpdate(bool flag,quint64 size,quint64 fileCount,quint64
     stream << size;
     stream << fileCount;
     stream << deleteCount;
+
+    waitWrittenData("sendNeedUpdate");
 }
 
-void SendSystem::updateFiles(QList<FileData> fileSendList, QList<FileData> deleteList){
+void SendSystem::updateFiles(QList<FileData> fileSendList, QList<FileData> deleteList)
+{
+    //globalMutex->lock();
 
-    QMutexLocker locker(mutex);
     QListIterator<FileData> clientIterator(deleteList);
 
     while(clientIterator.hasNext())
@@ -283,6 +516,8 @@ void SendSystem::updateFiles(QList<FileData> fileSendList, QList<FileData> delet
         sendDeleteBlock(data.path);
         if(getIsSendStopped())
         {
+            Logger::instance().log("Client: " + client->getLogin() + " isSendStopped", LogLevel::ERROR);
+            //globalMutex->unlock();
             return;
         }
     }
@@ -296,20 +531,22 @@ void SendSystem::updateFiles(QList<FileData> fileSendList, QList<FileData> delet
         if (data.hash == "FOLDER")
         {
             sendFolderBlock(data.path);
-            socket->waitForBytesWritten();
         }
         else
         {
-            sendFileBlock(data.path);
-            socket->waitForBytesWritten();
+            sendFileBlock_V3(data.path);
         }
 
-        if(isSendStopped)
+        if(getIsSendStopped())
         {
+            //globalMutex->unlock();
             return;
         }
     }
+
     sendPacketType(PacketType::UPDATE_FILES_COMPLETE);
+
+    //globalMutex->unlock();
 }
 
 
@@ -338,7 +575,17 @@ bool SendSystem::getIsSendStopped() const
     return isSendStopped;
 }
 
-SendSystem::~SendSystem()
+bool SendSystem::waitWrittenData(QString marker, int msec)
 {
-
+    bool success = socket->waitForBytesWritten(msec);
+    if(success)
+    {
+        //qInfo() << "Data sended OK. <" + marker + ">";
+    }
+    else
+    {
+        qCritical() << "Data sended ERROR. <" + marker + ">";
+    }
+    return success;
 }
+