Rabbit Remote Control 0.1.0-alpha
Loading...
Searching...
No Matches
ChannelSFTP.cpp
1// Copyright Copyright (c) Kang Lin studio, All Rights Reserved
2// Author Kang Lin <kl222@126.com>
3
4#include <QLoggingCategory>
5#include "ChannelSFTP.h"
6#include "BackendFileTransfer.h"
7#include "RemoteFileSystemModel.h"
8#include "ListFileModel.h"
9
10#include <fcntl.h>
11#include <sys/stat.h>
12
13#ifdef _MSC_VER
14 // See: [_topen](https://learn.microsoft.com/zh-cn/cpp/c-runtime-library/reference/open-wopen?view=msvc-170)
15 #include <io.h>
16 #define S_IRUSR 0400 // Owner read permission
17 #define S_IWUSR 0200 // Owner write permission
18 #define S_IXUSR 0100 // Owner execute permission
19 #define S_IRWXU (S_IREAD | S_IWRITE | S_IEXEC) // Owner read, write, execute
20#endif
21
22static Q_LOGGING_CATEGORY(log, "Channel.SFTP")
23
24CChannelSFTP::CChannelSFTP(CBackend *pBackend, CParameterSSH *pPara,
25 bool bWakeUp, QObject *parent)
26 : CChannelSSH(pBackend, pPara, bWakeUp, parent)
27 , m_SessionSftp(nullptr)
28{
29 bool check = false;
30 CBackendFileTransfer* pB = qobject_cast<CBackendFileTransfer*>(pBackend);
31 if(pB) {
32 check = connect(this, SIGNAL(sigGetDir(CRemoteFileSystem*, QVector<QSharedPointer<CRemoteFileSystem> >, bool)),
33 pB, SIGNAL(sigGetDir(CRemoteFileSystem*, QVector<QSharedPointer<CRemoteFileSystem> >, bool)));
34 Q_ASSERT(check);
35 check = connect(this, SIGNAL(sigFileTransferUpdate(QSharedPointer<CFileTransfer>)),
36 pB, SIGNAL(sigFileTransferUpdate(QSharedPointer<CFileTransfer>)));
37 Q_ASSERT(check);
38 check = connect(this, SIGNAL(sigError(int,QString)),
39 pB, SIGNAL(sigError(int,QString)));
40 Q_ASSERT(check);
41 }
42}
43
44int CChannelSFTP::Process()
45{
46 int nRet = 0;
47
48 struct timeval timeout = {0, DEFAULT_TIMEOUT};
49 ssh_channel channels[2], channel_out[2];
50 channels[0] = nullptr; //m_SessionSftp->channel;
51 channels[1] = nullptr;
52
53 fd_set set;
54 FD_ZERO(&set);
55 socket_t fdEvent = SSH_INVALID_SOCKET;
56 if(m_pEvent)
57 fdEvent = m_pEvent->GetFd();
58 if(SSH_INVALID_SOCKET != fdEvent)
59 FD_SET(fdEvent, &set);
60
61 socket_t sshFd = ssh_get_fd(m_Session);
62 if(SSH_INVALID_SOCKET != sshFd)
63 FD_SET(sshFd, &set);
64
65 socket_t f = qMax(sshFd, fdEvent);
66 //qDebug(log) << "ssh_select:" << fdEvent;
67 nRet = ssh_select(channels, channel_out, f + 1, &set, &timeout);
68 //qDebug(log) << "ssh_select end:" << nRet;
69 if(EINTR == nRet)
70 return 0;
71
72 if(SSH_OK != nRet) {
73 QString szErr;
74 szErr = "ssh_channel_select failed: " + QString::number(nRet);
75 szErr += ssh_get_error(m_Session);
76 qCritical(log) << szErr;
77 setErrorString(szErr);
78 return -3;
79 }
80
81 if(SSH_INVALID_SOCKET != fdEvent && FD_ISSET(fdEvent, &set)) {
82 //qDebug(log) << "fires event";
83 if(m_pEvent) {
84 nRet = m_pEvent->Reset();
85 if(nRet) {
86 QString szErr = "Reset event fail";
87 qCritical(log) << szErr;
88 setErrorString(szErr);
89 return -4;
90 }
91 }
92 }
93
94 AsyncReadDir();
95
96 AsyncFile();
97
98 return 0;
99}
100
101QSharedPointer<CRemoteFileSystem> CChannelSFTP::GetFileNode(
102 const QString& szPath, sftp_attributes attributes)
103{
104 if(!attributes) return nullptr;
105 /*
106 qDebug(log) << szPath << "name:" << attributes->name
107 << "size:" << attributes->size << "type:" << attributes->type;//*/
108 QString szName = QString::fromUtf8(attributes->name);
109 if("." == szName || ".." == szName)
110 return nullptr;
111 CRemoteFileSystem::TYPE type = CRemoteFileSystem::TYPE::NO;
112 switch(attributes->type) {
113 case SSH_FILEXFER_TYPE_DIRECTORY:
114 type = CRemoteFileSystem::TYPE::DIR;
115 break;
116 case SSH_FILEXFER_TYPE_SYMLINK:
117 type = CRemoteFileSystem::TYPE::SYMLINK;
118 break;
119 case SSH_FILEXFER_TYPE_REGULAR:
120 type = CRemoteFileSystem::TYPE::FILE;
121 break;
122 case SSH_FILEXFER_TYPE_SPECIAL:
123 type = CRemoteFileSystem::TYPE::SPECIAL;
124 break;
125 default:
126 qWarning(log) << "Unsupported type:" << attributes->type;
127 return nullptr;
128 break;
129 }
130 if(szPath.right(1) == '/')
131 szName = szPath + szName;
132 else
133 szName = szPath + "/" + szName;
134 QSharedPointer<CRemoteFileSystem> p(new CRemoteFileSystem(szName, type));
135 p->SetSize(attributes->size);
136 p->SetPermissions((QFileDevice::Permissions)attributes->permissions);
137 p->SetLastModified(QDateTime::fromSecsSinceEpoch(attributes->mtime));
138 p->SetCreateTime(QDateTime::fromSecsSinceEpoch(attributes->createtime));
139 return p;
140}
141
142int CChannelSFTP::GetDir(CRemoteFileSystem* p)
143{
144 int nRet = SSH_OK;
145 if(!m_SessionSftp || !p)
146 return -1;
147 sftp_dir dir = nullptr;
148 sftp_attributes attributes = nullptr;
149
150 QString szPath = p->GetPath();
151 QVector<QSharedPointer<CRemoteFileSystem> > vFileNode;
152 dir = sftp_opendir(m_SessionSftp, szPath.toStdString().c_str());
153 if (!dir)
154 {
155 QString szErr = "Directory not opened:"
156 + QString::number(sftp_get_error(m_SessionSftp))
157 + ssh_get_error(m_Session);
158 qCritical(log) << szErr;
159 emit sigError(-1, szErr);
160 return SSH_ERROR;
161 }
162
163 while ((attributes = sftp_readdir(m_SessionSftp, dir)) != NULL)
164 {
165 qDebug(log) << "name:" << attributes->name << "size:" << attributes->size;
166 auto p = GetFileNode(szPath, attributes);
167 if(p)
168 vFileNode.append(p);
169 sftp_attributes_free(attributes);
170 }
171
172 if (!sftp_dir_eof(dir))
173 {
174 qCritical(log) << "Can't list directory:"
175 << sftp_get_error(m_SessionSftp)
176 << ssh_get_error(m_Session);
177 sftp_closedir(dir);
178 return SSH_ERROR;
179 }
180
181 nRet = sftp_closedir(dir);
182 if (nRet != SSH_OK)
183 {
184 qCritical(log) << "Can't close directory:"
185 << sftp_get_error(m_SessionSftp)
186 << ssh_get_error(m_Session);
187 return nRet;
188 }
189
190 emit sigGetDir(p, vFileNode, true);
191
192 return nRet;
193}
194
195int CChannelSFTP::MakeDir(const QString &dir)
196{
197 int nRet = SSH_FX_OK;
198 if(!m_SessionSftp)
199 return SSH_FX_NO_CONNECTION;
200 nRet = sftp_mkdir(m_SessionSftp, dir.toStdString().c_str(), S_IRWXU);
201 if (nRet != SSH_OK)
202 {
203 if (sftp_get_error(m_SessionSftp) != SSH_FX_FILE_ALREADY_EXISTS)
204 {
205 QString szErr = "Can't create directory: "
206 + dir + QString::number(nRet)
207 + ssh_get_error(m_Session);
208 qCritical(log) << szErr;
209 emit sigError(nRet, szErr);
210 } else
211 qDebug(log) << "Create directory:" << dir;
212 }
213 return nRet;
214}
215
216int CChannelSFTP::RemoveDir(const QString& dir)
217{
218 if (!m_SessionSftp)
219 return SSH_FX_NO_CONNECTION;
220
221 // 1. 打开目录
222 sftp_dir sftpDir = sftp_opendir(m_SessionSftp, dir.toStdString().c_str());
223 if (!sftpDir) {
224 qCritical(log) << "Failed to open directory for remove:" << dir;
225 // 如果打不开目录，尝试直接删除（可能是空目录或者文件）
226 int ret = sftp_rmdir(m_SessionSftp, dir.toStdString().c_str());
227 if (ret == SSH_OK) {
228 qDebug(log) << "Removed directory (directly):" << dir;
229 return SSH_OK;
230 } else {
231 QString szErr = "Can't remove directory:" + dir + ssh_get_error(m_Session);
232 qCritical(log) << szErr;
233 emit sigError(ret, szErr);
234 return ret;
235 }
236 }
237
238 // 2. 遍历目录内容
239 sftp_attributes attr;
240 while ((attr = sftp_readdir(m_SessionSftp, sftpDir)) != NULL) {
241 QString name = QString::fromUtf8(attr->name);
242 if (name == "." || name == "..") {
243 sftp_attributes_free(attr);
244 continue;
245 }
246 QString fullPath = dir + "/" + name;
247 if (attr->type == SSH_FILEXFER_TYPE_DIRECTORY) {
248 // 递归删除子目录
249 RemoveDir(fullPath);
250 } else {
251 // 删除文件
252 int ret = sftp_unlink(m_SessionSftp, fullPath.toStdString().c_str());
253 if (ret != SSH_OK) {
254 QString szErr = "Can't remove file:" + fullPath + ssh_get_error(m_Session);
255 qCritical(log) << szErr;
256 emit sigError(ret, szErr);
257 } else {
258 qDebug(log) << "Removed file:" << fullPath;
259 }
260 }
261 sftp_attributes_free(attr);
262 }
263
264 sftp_closedir(sftpDir);
265
266 // 3. 删除空目录
267 int ret = sftp_rmdir(m_SessionSftp, dir.toStdString().c_str());
268 if (ret != SSH_OK) {
269 QString szErr = "Can't remove directory:" + dir + ssh_get_error(m_Session);
270 qCritical(log) << szErr;
271 emit sigError(ret, szErr);
272 } else {
273 qDebug(log) << "Removed directory:" << dir;
274 }
275 return ret;
276}
277
278int CChannelSFTP::RemoveFile(const QString &file)
279{
280 // 删除文件
281 int ret = sftp_unlink(m_SessionSftp, file.toStdString().c_str());
282 if (ret != SSH_OK) {
283 QString szErr = "Can't remove file:" + file + ssh_get_error(m_Session);
284 qCritical(log) << szErr;
285 emit sigError(ret, szErr);
286 } else {
287 qDebug(log) << "Removed file:" << file;
288 }
289 return ret;
290}
291
292int CChannelSFTP::Rename(const QString &oldPath, const QString &newPath)
293{
294 int nRet = SSH_FX_OK;
295 if(!m_SessionSftp)
296 return SSH_FX_NO_CONNECTION;
297 nRet = sftp_rename(m_SessionSftp,
298 oldPath.toStdString().c_str(),
299 newPath.toStdString().c_str());
300 if (nRet != SSH_OK)
301 {
302 QString szErr = "Fail: Can't rename: " + QString::number(nRet)
303 + ssh_get_error(m_Session)
304 + " " + oldPath + " to " + newPath;
305 qCritical(log) << szErr;
306 emit sigError(nRet, szErr);
307 } else
308 qDebug(log) << "Rename:" << oldPath << "to" << newPath;
309 return nRet;
310}
311
312qint64 CChannelSFTP::readData(char *data, qint64 maxlen)
313{
314 qint64 nRet = 0;
315
316 return nRet;
317}
318
319qint64 CChannelSFTP::writeData(const char *data, qint64 len)
320{
321 qint64 nRet = 0;
322
323 return nRet;
324}
325
326int CChannelSFTP::OnOpen(ssh_session session)
327{
328 int nRet = SSH_OK;
329
330 m_SessionSftp = sftp_new(session);
331 if (!m_SessionSftp)
332 {
333 QString szErr = "Error allocating SFTP session: ";
334 szErr += ssh_get_error(session);
335 qCritical(log) << szErr;
336 emit sigError(nRet, szErr);
337 return SSH_ERROR;
338 }
339
340 nRet = sftp_init(m_SessionSftp);
341 if (SSH_OK != nRet)
342 {
343 QString szErr = "Error initializing SFTP session:" + sftp_get_error(m_SessionSftp);
344 qCritical(log) << szErr;
345 emit sigError(nRet, szErr);
346 sftp_free(m_SessionSftp);
347 m_SessionSftp = nullptr;
348 return nRet;
349 }
350
351 return nRet;
352}
353
354void CChannelSFTP::OnClose()
355{
356 foreach (auto d, m_vDirs) {
357 if(d->sftp) {
358 sftp_closedir(d->sftp);
359 d->sftp = nullptr;
360 }
361 }
362 m_vDirs.clear();
363 foreach(auto f, m_vFiles) {
364 if(f->remote) {
365 sftp_close(f->remote);
366 f->remote = nullptr;
367 }
368 if(-1 != f->local) {
369 ::close(f->local);
370 f->local = -1;
371 }
372 }
373 m_vFiles.clear();
374 if(m_SessionSftp) {
375 sftp_free(m_SessionSftp);
376 m_SessionSftp = nullptr;
377 }
378}
379
380void CChannelSFTP::slotGetDir(CRemoteFileSystem *p)
381{
382 if(!p) return;
383 QString szPath = p->GetPath();
384 if(szPath.isEmpty()) {
385 qCritical(log) << "The path is empty";
386 return;
387 }
388 foreach(auto d, m_vDirs) {
389 if(d->remoteFileSystem == p) {
390 qDebug(log) << szPath << "already exists";
391 return;
392 }
393 }
394 QSharedPointer<DIR_READER> dir(new DIR_READER());
395 if(!dir)
396 return;
397 dir->sftp = nullptr;
398 dir->szPath = szPath;
399 dir->state = STATE::OPEN;
400 dir->remoteFileSystem = p;
401 dir->Error = SSH_FX_OK;
402 m_vDirs.append(dir);
403 WakeUp();
404}
405
406int CChannelSFTP::AsyncReadDir()
407{
408 //qDebug(log) << Q_FUNC_INFO;
409 for(auto it = m_vDirs.begin(); it != m_vDirs.end();) {
410 auto d = *it;
411 switch (d->state) {
412 case STATE::OPEN: {
413 d->sftp = sftp_opendir(m_SessionSftp, d->szPath.toStdString().c_str());
414 if(d->sftp) {
415 d->state = STATE::READ;
416 break;
417 }
418 int err = sftp_get_error(m_SessionSftp);
419 if (err == SSH_FX_OK || err == SSH_FX_EOF) {
420 d->state = STATE::FINISH;
421 break;
422 }
423 if (err == SSH_FX_FAILURE && ssh_get_error_code(m_Session) == SSH_REQUEST_DENIED) {
424 qDebug(log) << "Request denied:" << d->szPath;
425 break;
426 }
427 qCritical(log) << "Error opening directory:" << d->szPath
428 << "Error:"
429 << sftp_get_error(m_SessionSftp)
430 << ssh_get_error(m_Session);
431 d->state = STATE::ERR;
432 d->Error = err;
433 break;
434 }
435 case STATE::READ: {
436 sftp_attributes attributes = nullptr;
437 while ((attributes = sftp_readdir(m_SessionSftp, d->sftp)) != NULL) {
438 auto p = GetFileNode(d->szPath, attributes);
439 if(p)
440 d->vFileNode.append(p);
441 sftp_attributes_free(attributes);
442 }
443 // 检查是否结束
444 if (sftp_dir_eof(d->sftp)) {
445 d->state = STATE::CLOSE;
446 break;
447 }
448 int err = sftp_get_error(m_SessionSftp);
449 if (err == SSH_FX_OK)
450 break;
451 if (err == SSH_FX_EOF) {
452 break;
453 d->state = STATE::CLOSE;
454 }
455 qCritical(log) << "Error reading directory:" << d->szPath
456 << "Error:" << err << ssh_get_error(m_Session);
457 d->state = STATE::ERR;
458 d->Error = err;
459 break;
460 }
461 case STATE::CLOSE: {
462 if(!d->sftp) {
463 d->state = STATE::FINISH;
464 break;
465 }
466 int rc = sftp_closedir(d->sftp);
467 if(SSH_NO_ERROR == rc) {
468 d->state = STATE::FINISH;
469 d->sftp = nullptr;
470 break;
471 }
472 int err = ssh_get_error_code(m_Session);
473 qCritical(log) << "Error close directory:" << d->szPath
474 << "Error:" << err << ssh_get_error(m_Session);
475 d->state = STATE::FINISH;
476 d->Error = err;
477 break;
478 }
479 case STATE::FINISH:
480 if(d && STATE::FINISH == d->state) {
481 qDebug(log) << "Remote" << d->szPath << d->vFileNode.size();
482 emit sigGetDir(d->remoteFileSystem, d->vFileNode, true);
483 it = m_vDirs.erase(it);
484 }
485 continue;
486 case STATE::ERR:
487 d->state = STATE::CLOSE;
488 break;
489 default:
490 break;
491 }
492
493 it++;
494 }
495
496 return 0;
497}
498
499void CChannelSFTP::slotStartFileTransfer(QSharedPointer<CFileTransfer> f)
500{
501 f->slotSetstate(CFileTransfer::State::Opening);
502 QSharedPointer<_AFILE> file(new _AFILE);
503 memset(file.data(), 0, sizeof(_AFILE));
504 file->local = -1;
505 file->remote = nullptr;
506 file->state = STATE::OPEN;
507 file->fileTransfer = f;
508#if LIBSSH_VERSION_INT >= SSH_VERSION_INT(0, 11, 0)
509 file->nChunkSize = BUF_SIZE;
510 file->nConcurrentCount = 5;
511#endif
512 m_vFiles.append(file);
513 emit sigFileTransferUpdate(f);
514 WakeUp();
515}
516
517void CChannelSFTP::slotStopFileTransfer(QSharedPointer<CFileTransfer> f)
518{
519 foreach(auto file, m_vFiles) {
520 if(file->fileTransfer == f) {
521 m_vFiles.removeAll(file);
522 if(-1 != file->local) {
523 ::close(file->local);
524 file->local = -1;
525 }
526 if(file->remote) {
527 sftp_close(file->remote);
528 file->remote = nullptr;
529 }
530 f->slotSetstate(CFileTransfer::State::Stop);
531 emit sigFileTransferUpdate(f);
532 break;
533 }
534 }
535}
536
537int CChannelSFTP::AsyncFile()
538{
539 for(auto it = m_vFiles.begin(); it != m_vFiles.end();) {
540 auto file = *it;
541 switch(file->state) {
542 case STATE::OPEN: {
543 int remoteFlag = O_WRONLY | O_CREAT | O_TRUNC;
544 int localFlag = O_RDONLY;
545 quint32 permission = file->fileTransfer->GetRemotePermission();
546
547 if(file->fileTransfer->GetDirection() == CFileTransfer::Direction::Download) {
548 remoteFlag = O_RDONLY;
549 localFlag = O_WRONLY | O_CREAT | O_TRUNC;
550 }
551
552 file->remote = sftp_open(
553 m_SessionSftp,
554 file->fileTransfer->GetRemoteFile().toStdString().c_str(),
555 remoteFlag, permission); // S_IRWXU);
556 if (!file->remote)
557 {
558 file->state = STATE::ERR;
559 QString szErr = "Can't open remote file: " + file->fileTransfer->GetRemoteFile()
560 + ssh_get_error(m_Session);
561 file->fileTransfer->slotSetExplanation(szErr);
562 qCritical(log) << szErr;
563 break;
564 }
565 qDebug(log) << "Open remote file:" << file->fileTransfer->GetRemoteFile();
566
567 sftp_file_set_nonblocking(file->remote);
568
569 file->local = ::open(
570 file->fileTransfer->GetLocalFile().toStdString().c_str(),
571 localFlag, permission);
572 if(-1 == file->local) {
573 file->state = STATE::ERR;
574 QString szErr = "Can't open local file: " + file->fileTransfer->GetLocalFile() + " " + strerror(errno);
575 file->fileTransfer->slotSetExplanation(szErr);
576 qCritical(log) << szErr;
577 break;
578 }
579 qDebug(log) << "Open local file:" << file->fileTransfer->GetLocalFile();
580
581 if(file->fileTransfer->GetDirection() == CFileTransfer::Direction::Download) {
582#if LIBSSH_VERSION_INT >= SSH_VERSION_INT(0, 11, 0)
583 if(ssh_version(SSH_VERSION_INT(0, 11,0))) {
584 sftp_limits_t lim = sftp_limits(m_SessionSftp);
585 if(lim) {
586 file->nChunkSize = lim->max_read_length;
587 qDebug(log) << "limits: max_open_handles:" << lim->max_open_handles
588 << "max_packet_length" << lim->max_packet_length
589 << "max_read_length" << lim->max_read_length
590 << "max_write_length:" << lim->max_write_length;
591 sftp_limits_free(lim);
592 }
593
594 quint64 nRequestBytes = 0;
595 for(int i = 0;
596 i < file->nConcurrentCount
597 && nRequestBytes < file->fileTransfer->GetFileSize();
598 i++) {
599 sftp_aio aio = nullptr;
600 quint64 nRequest = file->fileTransfer->GetFileSize() - nRequestBytes;
601 if(nRequest > file->nChunkSize)
602 nRequest = file->nChunkSize;
603 ssize_t nRet = sftp_aio_begin_read(file->remote, nRequest, &aio);
604 if(0 > nRet) {
605 int rc = ssh_get_error_code(m_Session);
606 if (rc != SSH_NO_ERROR) {
607 QString szErr = "Error during sftp aio download: " + QString::number(rc);
608 szErr += ssh_get_error(m_Session);
609 qCritical(log) << szErr;
610 break;
611 }
612 }
613 Q_ASSERT(nRequest == nRet);
614 nRequestBytes += nRet;
615 file->aio.append(aio);
616 }
617 }
618#else
619 file->asyncReadId = sftp_async_read_begin(file->remote, BUF_SIZE);
620 if(file->asyncReadId < 0) {
621 file->state = STATE::ERR;
622 QString szErr = "sftp_async_read_begin failed." + sftp_get_error(m_SessionSftp);
623 file->fileTransfer->slotSetExplanation(szErr);
624 qCritical(log) << szErr;
625 break;
626 }
627#endif
628 } else { // Upload
629#if LIBSSH_VERSION_INT >= SSH_VERSION_INT(0, 11, 0)
630 if(ssh_version(SSH_VERSION_INT(0, 11,0))) {
631 sftp_limits_t lim = sftp_limits(m_SessionSftp);
632 if(lim) {
633 file->nChunkSize = lim->max_write_length;
634 qDebug(log) << "limits: max_open_handles:" << lim->max_open_handles
635 << "max_packet_length" << lim->max_packet_length
636 << "max_read_length" << lim->max_read_length
637 << "max_write_length:" << lim->max_write_length;
638 sftp_limits_free(lim);
639 }
640
641 if(!file->buffer)
642 file->buffer = new char[file->nChunkSize];
643 if(!file->buffer) {
644 file->state = STATE::ERR;
645 break;
646 }
647
648 for(int i = 0;
649 i < file->nConcurrentCount
650 && file->nRequests < file->fileTransfer->GetFileSize();
651 i++) {
652 sftp_aio aio = nullptr;
653 quint64 nRequest = file->fileTransfer->GetFileSize() - file->nRequests;
654 if(nRequest > file->nChunkSize)
655 nRequest = file->nChunkSize;
656 ssize_t nLen = ::read(file->local, file->buffer, nRequest);
657 Q_ASSERT(nLen == nRequest);
658 ssize_t nRet = sftp_aio_begin_write(file->remote, file->buffer, nLen, &aio);
659 if(0 > nRet) {
660 int rc = ssh_get_error_code(m_Session);
661 if (rc != SSH_NO_ERROR) {
662 QString szErr = "Error during sftp aio download: " + QString::number(rc);
663 szErr += ssh_get_error(m_Session);
664 qCritical(log) << szErr;
665 break;
666 }
667 }
668 Q_ASSERT(nRequest == nRet);
669 file->nRequests += nRet;
670 file->aio.append(aio);
671 }
672 }
673#endif
674 }
675
676 file->state = STATE::READ;
677 file->fileTransfer->slotSetstate(CFileTransfer::State::Transferring);
678 emit sigFileTransferUpdate(file->fileTransfer);
679 }
680 case STATE::READ: {
681 if(file->fileTransfer->GetDirection() == CFileTransfer::Direction::Download) {
682#if LIBSSH_VERSION_INT >= SSH_VERSION_INT(0, 11, 0)
683 if(ssh_version(SSH_VERSION_INT(0, 11,0))) {
684 for(auto it = file->aio.begin(); it != file->aio.end();) {
685 auto aio = *it;
686 if(nullptr == file->buffer)
687 file->buffer = new char[file->nChunkSize];
688 if(!(aio && file->buffer))
689 break;
690 ssize_t nRet = sftp_aio_wait_read(&aio, file->buffer, file->nChunkSize);
691 if (nRet < 0) {
692 if(SSH_AGAIN == nRet)
693 break;
694 int rc = ssh_get_error_code(m_Session);
695 if (rc != SSH_NO_ERROR) {
696 file->state = STATE::ERR;
697 QString szErr = "Error during sftp aio download: " + QString::number(rc);
698 szErr += ssh_get_error(m_Session);
699 qCritical(log) << szErr;
700 break;
701 }
702 }
703 file->nTransfers += nRet;
704 if(file->fileTransfer->GetFileSize() == file->nTransfers) {
705 file->state = STATE::CLOSE;
706 }
707 it = file->aio.erase(it);
708
709 int nLen = ::write(file->local, file->buffer, nRet);
710 if(nLen < 0) {
711 if(EAGAIN != nLen)
712 file->state = STATE::ERR;
713 break;
714 }
715 if(nLen != nRet) {
716 qCritical(log) << "IO is buse, Write file error:" << file->fileTransfer->GetLocalFile();
717 Q_ASSERT(false);
718 }
719 file->fileTransfer->slotTransferSize(nRet);
720 emit sigFileTransferUpdate(file->fileTransfer);
721 }
722 if(file->fileTransfer->GetFileSize() == file->nTransfers) {
723 file->state = STATE::CLOSE;
724 break;
725 }
726 if(file->aio.size() > 0)
727 break;
728
729 quint64 nRequestBytes = file->nTransfers;
730 for(int i = 0;
731 i < file->nConcurrentCount
732 && nRequestBytes < file->fileTransfer->GetFileSize();
733 i++) {
734 sftp_aio aio = nullptr;
735 quint64 nRequest = file->fileTransfer->GetFileSize() - nRequestBytes;
736 if(nRequest > file->nChunkSize)
737 nRequest = file->nChunkSize;
738 ssize_t nRet = sftp_aio_begin_read(file->remote, nRequest, &aio);
739 if(0 > nRet) {
740 int rc = ssh_get_error_code(m_Session);
741 if (rc != SSH_NO_ERROR) {
742 QString szErr = "Error during sftp aio download: " + QString::number(rc);
743 szErr += ssh_get_error(m_Session);
744 qCritical(log) << szErr;
745 break;
746 }
747 }
748 if (nRequest != nRet) {
749 QString szErr =
750 "Error during sftp aio download: sftp_aio_begin_read() "
751 "requesting less bytes even when the number of bytes "
752 "asked to read are within the max limit";
753 qWarning(log) << szErr << nRequest << nRet;
754 }
755
756 nRequestBytes += nRet;
757 file->aio.append(aio);
758 }
759 } else {
760 file->state = STATE::ERR;
761 QString szErr = tr("Error: Asynchronous uploads are not supported");
762 file->fileTransfer->slotSetExplanation(szErr);
763 qCritical(log) << szErr;
764 break;
765 }
766#else
767 int nbytes = sftp_async_read(file->remote, file->buffer + file->offset, BUF_SIZE, file->asyncReadId);
768 if (nbytes > 0 || SSH_AGAIN == nbytes) {
769 if(nbytes > 0 ) {
770 int nLen = ::write(file->local, file->buffer + file->offset, nbytes);
771 if(nLen > 0) {
772 file->fileTransfer->slotTransferSize(nLen);
773 //TODO: add nLen < nbytes
774 emit sigFileTransferUpdate(file->fileTransfer);
775 } else {
776 file->state = STATE::ERR;
777 QString szErr = "Write local file fail:" + sftp_get_error(m_SessionSftp);
778 file->fileTransfer->slotSetExplanation(szErr);
779 }
780 }
781 // Start next async read
782 if (file->asyncReadId = sftp_async_read_begin(file->remote, BUF_SIZE) < 0) {
783 file->state = STATE::ERR;
784 QString szErr = "sftp_async_read_begin failed." + sftp_get_error(m_SessionSftp);
785 file->fileTransfer->slotSetExplanation(szErr);
786 qCritical(log) << szErr;
787 }
788 } else if (nbytes == 0) {
789 file->state = STATE::CLOSE;
790 file->asyncReadId = -1;
791 } else {
792 file->state = STATE::ERR;
793 QString szErr = "sftp_async_read failed." + sftp_get_error(m_SessionSftp);
794 file->fileTransfer->slotSetExplanation(szErr);
795 qCritical(log) << szErr;
796 }
797#endif
798 } else { // Upload
799 if(ssh_version(SSH_VERSION_INT(0, 11,0))) {
800 for(auto it = file->aio.begin(); it != file->aio.end();) {
801 auto aio = *it;
802 ssize_t nRet = sftp_aio_wait_write(&aio);
803 if (nRet < 0) {
804 if(SSH_AGAIN == nRet)
805 break;
806 int rc = ssh_get_error_code(m_Session);
807 if (rc != SSH_NO_ERROR) {
808 file->state = STATE::ERR;
809 QString szErr = "Error during sftp aio upload: " + QString::number(rc);
810 szErr += ssh_get_error(m_Session);
811 qCritical(log) << szErr;
812 break;
813 }
814 }
815 file->nTransfers += nRet;
816 if(file->fileTransfer->GetFileSize() == file->nTransfers) {
817 file->state = STATE::CLOSE;
818 }
819 it = file->aio.erase(it);
820 }
821 if(file->aio.size() >= file->nConcurrentCount)
822 break;
823 int size = file->nConcurrentCount - file->aio.size();
824 for(int i = 0;
825 i < size
826 && file->nRequests < file->fileTransfer->GetFileSize();
827 i++) {
828 sftp_aio aio = nullptr;
829 quint64 nRequest = file->fileTransfer->GetFileSize() - file->nRequests;
830 if(nRequest > file->nChunkSize)
831 nRequest = file->nChunkSize;
832 ssize_t nLen = ::read(file->local, file->buffer, nRequest);
833 Q_ASSERT(nLen == nRequest);
834 ssize_t nRet = sftp_aio_begin_write(file->remote, file->buffer, nLen, &aio);
835 if(0 > nRet) {
836 int rc = ssh_get_error_code(m_Session);
837 if (rc != SSH_NO_ERROR) {
838 QString szErr = "Error during sftp aio download: " + QString::number(rc);
839 szErr += ssh_get_error(m_Session);
840 qCritical(log) << szErr;
841 break;
842 }
843 }
844 Q_ASSERT(nRequest == nRet);
845 file->nRequests += nRet;
846 file->aio.append(aio);
847 }
848 }
849 }
850 break;
851 }
852 case STATE::CLOSE: {
853 file->state = STATE::FINISH;
854 file->fileTransfer->slotSetstate(CFileTransfer::State::Closing);
855 emit sigFileTransferUpdate(file->fileTransfer);
856 break;
857 }
858 case STATE::FINISH: {
859#if LIBSSH_VERSION_INT >= SSH_VERSION_INT(0, 11, 0)
860 CleanFileAIO(file);
861#endif
862 it = m_vFiles.erase(it);
863 file->fileTransfer->slotSetstate(CFileTransfer::State::Finish);
864 emit sigFileTransferUpdate(file->fileTransfer);
865 continue;
866 }
867 case STATE::ERR: {
868#if LIBSSH_VERSION_INT >= SSH_VERSION_INT(0, 11, 0)
869 foreach (auto aio, file->aio) {
870 sftp_aio_free(aio);
871 }
872#endif
873 CleanFileAIO(file);
874
875 it = m_vFiles.erase(it);
876 file->fileTransfer->slotSetstate(CFileTransfer::State::Fail);
877 emit sigFileTransferUpdate(file->fileTransfer);
878 continue;
879 }
880 }
881
882 it++;
883 }
884 return 0;
885}
886
887int CChannelSFTP::CleanFileAIO(QSharedPointer<_AFILE> file)
888{
889 if(file->buffer) {
890 delete []file->buffer;
891 file->buffer = nullptr;
892 }
893 if(file->remote) {
894 sftp_close(file->remote);
895 file->remote == nullptr;
896 }
897 if(-1 != file->local) {
898 ::close(file->local);
899 file->local = -1;
900 }
901 return 0;
902}
CBackend
Backend interface.
Definition Backend.h:42
CChannelSFTP::GetDir
int GetDir(CRemoteFileSystem *p)
Synchronize to get the directory.
Definition ChannelSFTP.cpp:142
CChannelSFTP::slotGetDir
void slotGetDir(CRemoteFileSystem *p)
Get the directory asynchronously.
Definition ChannelSFTP.cpp:380
CChannelSSH::open
virtual bool open(OpenMode mode) override
Definition ChannelSSH.cpp:60
CChannel::sigError
void sigError(int nErr, const QString &szErr)
emit when the channel is error
CChannelSFTP::DIR_READER
Definition ChannelSFTP.h:60
Author: Kang Lin (kl222@126.com)

Copyright (c) Kang Lin Studios All Rights Reserved