kget/transfer-plugins/multisegmentkio/segmentfactory.cpp

   1 /* This file is part of the KDE project
   2
   3    Copyright (C) 2006 Manolo Valdes <nolis71cu@gmail.com>
   4
   5    This program is free software; you can redistribute it and/or
   6    modify it under the terms of the GNU General Public
   7    License as published by the Free Software Foundation; either
   8    version 2 of the License, or (at your option) any later version.
   9 */
  10
  11 #include "segmentfactory.h"
  12 #include "multisegkiosettings.h"
  13 #include "../../settings.h"
  14
  15 #include <QtCore/QTimer>
  16
  17 SegData::SegData ()
  18 {
  19     offset = 0;
  20     bytes = (KIO::filesize_t) 0;
  21 }
  22
  23 Segment::Segment (QObject* parent)
  24   :QObject(parent)
  25     , m_status(Stopped)
  26     , m_bytesWritten(0)
  27     , m_getJob(0)
  28     , m_canResume(true)
  29 {
  30 }
  31
  32 bool Segment::createTransfer ( const KUrl &src )
  33 {
  34     kDebug(5001) << " -- " << src;
  35     if ( m_getJob )
  36         return false;
  37     m_getJob = KIO::get(src, KIO::NoReload, KIO::HideProgressInfo);
  38     m_getJob->suspend();
  39     m_getJob->addMetaData( "errorPage", "false" );
  40     m_getJob->addMetaData( "AllowCompressedPage", "false" );
  41     if ( m_segData.offset )
  42     {
  43         m_canResume = false;
  44         m_getJob->addMetaData( "resume", KIO::number(m_segData.offset) );
  45         connect(m_getJob, SIGNAL(canResume(KIO::Job *, KIO::filesize_t)),
  46                  SLOT( slotCanResume(KIO::Job *, KIO::filesize_t)));
  47     }
  48     #if 0 //TODO: we disable that code till it's implemented in kdelibs, also we need to think, which settings we should use
  49     if(Settings::speedLimit())
  50     {
  51                 m_getJob->addMetaData( "speed-limit", KIO::number(Settings::transferSpeedLimit() * 1024) );
  52     }
  53     #endif
  54     connect( m_getJob, SIGNAL(data(KIO::Job *, const QByteArray&)),
  55                  SLOT( slotData(KIO::Job *, const QByteArray&)));
  56     connect( m_getJob, SIGNAL(result(KJob *)), SLOT(slotResult( KJob *)));
  57
  58     return true;
  59 }
  60
  61 void Segment::slotCanResume( KIO::Job* job, KIO::filesize_t offset )
  62 {
  63     Q_UNUSED(job);
  64     Q_UNUSED(offset);
  65     kDebug(5001);
  66     m_canResume = true;
  67 }
  68
  69 bool Segment::startTransfer ()
  70 {
  71     kDebug(5001);
  72     if( m_getJob && m_status != Running )
  73     {
  74         setStatus( Running, false );
  75         m_getJob->resume();
  76         return true;
  77     }
  78     return false;
  79 }
  80
  81 bool Segment::stopTransfer ()
  82 {
  83     kDebug(5001);
  84     if( m_getJob && m_status == Running )
  85     {
  86         setStatus( Stopped, false );
  87         m_getJob->suspend();
  88         if ( !m_buffer.isEmpty() )
  89         {
  90             writeBuffer();
  91         }
  92         if (m_getJob)
  93             m_getJob->kill( KJob::EmitResult );
  94         return true;
  95     }
  96     return false;
  97 }
  98
  99 bool Segment::restartTransfer ( const KUrl &url )
 100 {
 101     bool rest;
 102     rest = createTransfer( url );
 103     rest |= startTransfer();
 104     return rest;
 105 }
 106
 107 void Segment::slotResult( KJob *job )
 108 {
 109     kDebug(5001) << "job: " << job;
 110     m_getJob = 0;
 111     if ( !m_buffer.isEmpty() )
 112     {
 113         kDebug(5001) << "Looping until write the buffer ...";
 114         while(writeBuffer()) ;
 115     }
 116     if( !m_segData.bytes )
 117     {
 118         setStatus(Finished);
 119         deleteLater();
 120         return;
 121     }
 122     if( m_status == Killed )
 123     {
 124         deleteLater();
 125         return;
 126     }
 127     if( m_status == Running )
 128     {
 129         kDebug(5001) << "Conection broken " << job << " --restarting--";
 130         setStatus(Timeout);
 131     }
 132 }
 133
 134 void Segment::slotData(KIO::Job *, const QByteArray& _data)
 135 {
 136 //     kDebug(5001) << "Segment::slotData()";
 137
 138     // Check if the transfer allow resuming...
 139     if ( m_segData.offset && !m_canResume)
 140     {
 141         kDebug(5001) << "the remote site does not allow resuming ...";
 142         stopTransfer();
 143         setStatus(Killed, false );
 144         return;
 145     }
 146
 147     m_buffer.append(_data);
 148     if ( (uint)m_buffer.size() > m_segData.bytes )
 149     {
 150 //         kDebug(5001) << "Segment::slotData() buffer full. stoping transfer...";
 151         m_buffer.truncate( m_segData.bytes );
 152         m_getJob->suspend();
 153         m_getJob->kill( KJob::EmitResult );
 154         writeBuffer();
 155     }
 156     else
 157     {
 158     /*
 159      write to the local file only if the buffer has more than 8kbytes
 160      this hack try to avoid too much cpu usage. it seems to be due KIO::Filejob
 161      so remove it when it works property
 162     */
 163     if ( m_buffer.size() > 8*1024)
 164         writeBuffer();
 165     }
 166 }
 167
 168 bool Segment::writeBuffer()
 169 {
 170 //     kDebug(5001) << "Segment::writeBuffer() sending: " << m_buffer.size() << " from job: "<< m_getJob;
 171     bool rest;
 172     emit data( this, m_buffer, rest);
 173     if ( rest )
 174     {
 175         m_segData.bytes -= m_buffer.size();
 176         m_segData.offset += m_buffer.size();
 177         m_bytesWritten += m_buffer.size();
 178         m_buffer = QByteArray();
 179 //         kDebug(5001) << "Segment::writeBuffer() updating segment record of job: " << m_getJob << " -- " << m_segData.bytes <<" bytes left";
 180     }
 181     if (!m_segData.bytes)
 182     {
 183         kDebug(5001) << "Closing transfer ...";
 184         if( m_getJob )
 185             m_getJob->kill( KJob::EmitResult );
 186         emit updateSegmentData();
 187     }
 188     return rest;
 189 }
 190
 191 void Segment::setStatus(Status stat, bool doEmit)
 192 {
 193     m_status = stat;
 194     if (doEmit)
 195         emit statusChanged(this);
 196 }
 197
 198 SegmentFactory::SegmentFactory(uint n, const QList<KUrl> Urls)
 199    : m_segments( n ), m_Urls( Urls ), m_split(true)
 200
 201 {
 202     kDebug(5001);
 203     it_Urls = m_Urls.begin();
 204 }
 205
 206 SegmentFactory::SegmentFactory()
 207 {
 208     kDebug(5001);
 209 }
 210
 211 SegmentFactory::~SegmentFactory()
 212 {
 213     kDebug(5001);
 214     QList<Segment *>::iterator it = m_Segments.begin();
 215     QList<Segment *>::iterator itEnd = m_Segments.end();
 216     for ( ; it!=itEnd ; ++it )
 217     {
 218         if( (*it)->status() == Segment::Running )
 219             (*it)->stopTransfer();
 220         (*it)->deleteLater();
 221     }
 222 }
 223
 224 bool SegmentFactory::startTransfer()
 225 {
 226     kDebug(5001);
 227     bool rest = false;
 228     QList<Segment *>::iterator it = m_Segments.begin();
 229     QList<Segment *>::iterator itEnd = m_Segments.end();
 230     for ( ; it!=itEnd ; ++it )
 231     {
 232         rest |= (*it)->startTransfer();
 233     }
 234     return rest;
 235 }
 236
 237 bool SegmentFactory::stopTransfer()
 238 {
 239     kDebug(5001);
 240     bool rest = false;
 241     QList<Segment *>::iterator it = m_Segments.begin();
 242     QList<Segment *>::iterator itEnd = m_Segments.end();
 243     for ( ; it!=itEnd ; ++it )
 244     {
 245         rest |= (*it)->stopTransfer();
 246     }
 247     return rest;
 248 }
 249
 250 QList<SegData> SegmentFactory::SegmentsData()
 251 {
 252     kDebug(5001);
 253     QList<SegData> tdata;
 254     QList<Segment *>::iterator it = m_Segments.begin();
 255     QList<Segment *>::iterator itEnd = m_Segments.end();
 256     for ( ; it!=itEnd ; ++it )
 257     {
 258         if( (*it)->data().bytes )
 259             tdata << (*it)->data();
 260     }
 261     return tdata;
 262 }
 263
 264 QList<Segment *> SegmentFactory::splitSegment( Segment *Seg, int n)
 265 {
 266     kDebug(5001) << "Spliting " << Seg << "in " << n;
 267     QList<Segment *> Segments;
 268
 269     KIO::TransferJob *Job = Seg->job();
 270     if(Job)
 271     {
 272         Job->suspend();
 273         kDebug(5001) << "job Suspended...";
 274     }
 275
 276     KIO::filesize_t bytes = Seg->data().bytes;
 277     KIO::filesize_t offset = Seg->data().offset;
 278
 279     uint splitSize = 50;
 280     if( MultiSegKioSettings::splitSize() )
 281     {
 282         splitSize = MultiSegKioSettings::splitSize();
 283     }
 284     int min = bytes/(splitSize*1024);
 285
 286     if( min < n )
 287     {
 288         n = min;
 289     }
 290
 291     if( n == 0 )
 292     {
 293         kDebug(5001) << "Segment can't be splited.";
 294         if(Job)
 295         {
 296             Job->resume();
 297             kDebug(5001) << "Resuming Job...";
 298         }
 299         return Segments;
 300     }
 301
 302     KIO::filesize_t segment = bytes/n;
 303
 304     kDebug(5001) << "spliting: " << Seg->data().bytes <<" in "<< n << "  and got: " << segment;
 305
 306     KIO::fileoffset_t rest_size = segment + ( bytes%n );
 307     Seg->setBytes( segment );
 308     kDebug(5001) << "Now the segment has: " << Seg->data().bytes <<" bytes.";
 309
 310     if(Job)
 311     {
 312         Job->resume();
 313         kDebug(5001) << "Resuming Job...";
 314     }
 315
 316     SegData data;
 317     for(int i = 1; i < n; i++)
 318     {
 319         if(i == n - 1)
 320         {
 321             data.offset = i*segment + offset;
 322             data.bytes = rest_size;
 323             Segments << createSegment(data, nextUrl());
 324             kDebug(5001) << "Segment created at offset: "<< data.offset <<" with "<< data.bytes << " bytes.";
 325             continue;
 326         }
 327         data.offset = i*segment + offset;
 328         data.bytes = segment;
 329         Segments << createSegment(data, nextUrl());
 330         kDebug(5001) << "Segment created at offset: "<< data.offset <<" with "<< data.bytes << " bytes.";
 331     }
 332
 333     return Segments;
 334 }
 335
 336 Segment *SegmentFactory::createSegment( SegData data, const KUrl &src )
 337 {
 338     kDebug(5001);
 339     Segment *seg = new Segment(this);
 340     connect( seg, SIGNAL(statusChanged( Segment *)),
 341                   SLOT(slotStatusChanged( Segment *)));
 342     seg->setData(data);
 343     seg->createTransfer( src );
 344     m_Segments.append(seg);
 345     emit createdSegment(seg);
 346     return seg;
 347 }
 348
 349 void SegmentFactory::deleteSegment(Segment *seg)
 350 {
 351     m_Segments.removeAll(seg);
 352     kDebug(5001) << m_Segments.size() << " segments left.";
 353 }
 354
 355 void SegmentFactory::slotStatusChanged( Segment *seg)
 356 {
 357     kDebug(5001) << seg->status();
 358     switch (seg->status())
 359     {
 360     case Segment::Killed :
 361         //this site does not allow resuming. so its useless for multiseg
 362         if ( !DeleteUrl( seg->job()->url() ) )
 363         {
 364         //TODO: notify that we cant resume or manage this transfer
 365         }
 366     break;
 367     case Segment::Timeout :
 368         kDebug(5001) << "Restarting Segment in 5 seg... ";
 369         m_TimeOutSegments << seg;
 370         QTimer::singleShot(5000, this, SLOT(slotSegmentTimeOut()));
 371     break;
 372     case Segment::Finished :
 373         deleteSegment(seg);
 374         if( !m_Segments.isEmpty() )
 375         {
 376             Segment* longSeg = takeLongest();
 377             if( longSeg == 0)
 378                 break;
 379             QList<Segment*> segl = splitSegment( longSeg, 2);
 380             if( !segl.isEmpty() )
 381                 segl.takeFirst()->startTransfer();
 382         }
 383     break;
 384     default:
 385     break;
 386     }
 387 }
 388
 389 void SegmentFactory::slotSegmentTimeOut()
 390 {
 391     kDebug(5001) <<  m_TimeOutSegments.size();
 392     if(m_TimeOutSegments.isEmpty())
 393         return;
 394     m_TimeOutSegments.takeFirst()->restartTransfer( nextUrl() );
 395 }
 396
 397 Segment *SegmentFactory::takeLongest()
 398 {
 399     kDebug(5001);
 400
 401     Segment *longest = 0;
 402     KIO::filesize_t bytes = MultiSegKioSettings::splitSize()*1024;
 403
 404     QList<Segment*>::const_iterator it = m_Segments.begin();
 405     QList<Segment*>::const_iterator itEnd = m_Segments.end();
 406     for ( ; it!=itEnd ; ++it )
 407     {
 408         if((*it)->data().bytes > bytes)
 409         {
 410             longest = (*it);
 411             bytes = (*it)->data().bytes;
 412         }
 413     }
 414
 415     if(longest)
 416         kDebug(5001) << "the longest segment has: " << longest->data().bytes;
 417
 418     return longest;
 419 }
 420
 421 const KUrl SegmentFactory::nextUrl()
 422 {
 423     kDebug(5001);
 424     if ( it_Urls == m_Urls.end() )
 425     {
 426         it_Urls = m_Urls.begin();
 427     }
 428     KUrl url(*it_Urls);
 429     it_Urls++;
 430     return url;
 431 }
 432
 433 bool SegmentFactory::DeleteUrl(const KUrl &url)
 434 {
 435     if ( m_Urls.count() == 1 )
 436     {
 437         return false;
 438     }
 439     if ( m_Urls.contains( url ) )
 440     {
 441         m_Urls.removeAll( url );
 442         return true;
 443     }
 444     return false;
 445 }
 446
 447 #include "segmentfactory.moc"