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     if ( m_buffer.isEmpty() )
 172     {
 173         return false;
 174     }
 175    bool rest;
 176     emit data( this, m_buffer, rest);
 177     if ( rest )
 178     {
 179         m_segData.bytes -= m_buffer.size();
 180         m_segData.offset += m_buffer.size();
 181         m_bytesWritten += m_buffer.size();
 182         m_buffer = QByteArray();
 183 //         kDebug(5001) << "Segment::writeBuffer() updating segment record of job: " << m_getJob << " -- " << m_segData.bytes <<" bytes left";
 184     }
 185     if (!m_segData.bytes)
 186     {
 187         kDebug(5001) << "Closing transfer ...";
 188         if( m_getJob )
 189             m_getJob->kill( KJob::EmitResult );
 190         emit updateSegmentData();
 191     }
 192     return rest;
 193 }
 194
 195 void Segment::setStatus(Status stat, bool doEmit)
 196 {
 197     m_status = stat;
 198     if (doEmit)
 199         emit statusChanged(this);
 200 }
 201
 202 SegmentFactory::SegmentFactory(uint n, const QList<KUrl> Urls)
 203    : m_segments( n ), m_Urls( Urls ), m_split(true)
 204
 205 {
 206     kDebug(5001);
 207     it_Urls = m_Urls.constBegin();
 208 }
 209
 210 SegmentFactory::SegmentFactory()
 211 {
 212     kDebug(5001);
 213 }
 214
 215 SegmentFactory::~SegmentFactory()
 216 {
 217     kDebug(5001);
 218     QList<Segment *>::iterator it = m_Segments.begin();
 219     QList<Segment *>::iterator itEnd = m_Segments.end();
 220     for ( ; it!=itEnd ; ++it )
 221     {
 222         if( (*it)->status() == Segment::Running )
 223             (*it)->stopTransfer();
 224         (*it)->deleteLater();
 225     }
 226 }
 227
 228 bool SegmentFactory::startTransfer()
 229 {
 230     kDebug(5001);
 231     bool rest = false;
 232     QList<Segment *>::iterator it = m_Segments.begin();
 233     QList<Segment *>::iterator itEnd = m_Segments.end();
 234     for ( ; it!=itEnd ; ++it )
 235     {
 236         rest |= (*it)->startTransfer();
 237     }
 238     return rest;
 239 }
 240
 241 bool SegmentFactory::stopTransfer()
 242 {
 243     kDebug(5001);
 244     bool rest = false;
 245     QList<Segment *>::iterator it = m_Segments.begin();
 246     QList<Segment *>::iterator itEnd = m_Segments.end();
 247     for ( ; it!=itEnd ; ++it )
 248     {
 249         rest |= (*it)->stopTransfer();
 250     }
 251     return rest;
 252 }
 253
 254 QList<SegData> SegmentFactory::SegmentsData()
 255 {
 256     kDebug(5001);
 257     QList<SegData> tdata;
 258     QList<Segment *>::iterator it = m_Segments.begin();
 259     QList<Segment *>::iterator itEnd = m_Segments.end();
 260     for ( ; it!=itEnd ; ++it )
 261     {
 262         if( (*it)->data().bytes )
 263             tdata << (*it)->data();
 264     }
 265     return tdata;
 266 }
 267
 268 QList<Segment *> SegmentFactory::splitSegment( Segment *Seg, int n)
 269 {
 270     kDebug(5001) << "Spliting " << Seg << "in " << n;
 271     QList<Segment *> Segments;
 272
 273     KIO::TransferJob *Job = Seg->job();
 274     if(Job)
 275     {
 276         Job->suspend();
 277         kDebug(5001) << "job Suspended...";
 278     }
 279
 280     KIO::filesize_t bytes = Seg->data().bytes;
 281     KIO::filesize_t offset = Seg->data().offset;
 282
 283     uint splitSize = 50;
 284     if( MultiSegKioSettings::splitSize() )
 285     {
 286         splitSize = MultiSegKioSettings::splitSize();
 287     }
 288     int min = bytes/(splitSize*1024);
 289
 290     if( min < n )
 291     {
 292         n = min;
 293     }
 294
 295     if( n == 0 )
 296     {
 297         kDebug(5001) << "Segment can't be splited.";
 298         if(Job)
 299         {
 300             Job->resume();
 301             kDebug(5001) << "Resuming Job...";
 302         }
 303         return Segments;
 304     }
 305
 306     KIO::filesize_t segment = bytes/n;
 307
 308     kDebug(5001) << "spliting: " << Seg->data().bytes <<" in "<< n << "  and got: " << segment;
 309
 310     KIO::fileoffset_t rest_size = segment + ( bytes%n );
 311     Seg->setBytes( segment );
 312     kDebug(5001) << "Now the segment has: " << Seg->data().bytes <<" bytes.";
 313
 314     SegData data;
 315     for(int i = 1; i < n; i++)
 316     {
 317         if(i == n - 1)
 318         {
 319             data.offset = i*segment + offset;
 320             data.bytes = rest_size;
 321         }
 322         else
 323         {
 324             data.offset = i*segment + offset;
 325             data.bytes = segment;
 326         }
 327         Segments << createSegment(data, nextUrl());
 328         kDebug(5001) << "Segment created at offset: "<< data.offset <<" with "<< data.bytes << " bytes.";
 329     }
 330
 331     if(Job)
 332     {
 333         Job->resume();
 334         kDebug(5001) << "Resuming Job...";
 335     }
 336
 337     return Segments;
 338 }
 339
 340 Segment *SegmentFactory::createSegment( SegData data, const KUrl &src )
 341 {
 342     kDebug(5001);
 343     Segment *seg = new Segment(this);
 344     connect( seg, SIGNAL(statusChanged( Segment *)),
 345                   SLOT(slotStatusChanged( Segment *)));
 346     seg->setData(data);
 347     seg->createTransfer( src );
 348     m_Segments.append(seg);
 349     emit createdSegment(seg);
 350     return seg;
 351 }
 352
 353 void SegmentFactory::deleteSegment(Segment *seg)
 354 {
 355     m_Segments.removeAll(seg);
 356     kDebug(5001) << m_Segments.size() << " segments left.";
 357 }
 358
 359 void SegmentFactory::slotStatusChanged( Segment *seg)
 360 {
 361     kDebug(5001) << seg->status();
 362     switch (seg->status())
 363     {
 364     case Segment::Killed :
 365         //this site does not allow resuming. so its useless for multiseg
 366         if ( !DeleteUrl( seg->job()->url() ) )
 367         {
 368         //TODO: notify that we cant resume or manage this transfer
 369         }
 370     break;
 371     case Segment::Timeout :
 372         kDebug(5001) << "Restarting Segment in 5 seg... ";
 373         m_TimeOutSegments << seg;
 374         QTimer::singleShot(5000, this, SLOT(slotSegmentTimeOut()));
 375     break;
 376     case Segment::Finished :
 377         deleteSegment(seg);
 378         if( !m_Segments.isEmpty() )
 379         {
 380             Segment* longSeg = takeLongest();
 381             if( longSeg == 0)
 382                 break;
 383             QList<Segment*> segl = splitSegment( longSeg, 2);
 384             if( !segl.isEmpty() )
 385                 segl.takeFirst()->startTransfer();
 386         }
 387     break;
 388     default:
 389     break;
 390     }
 391 }
 392
 393 void SegmentFactory::slotSegmentTimeOut()
 394 {
 395     kDebug(5001) <<  m_TimeOutSegments.size();
 396     if(m_TimeOutSegments.isEmpty())
 397         return;
 398     QList<Segment*>::const_iterator it = m_TimeOutSegments.constBegin();
 399     QList<Segment*>::const_iterator itEnd = m_TimeOutSegments.constEnd();
 400     for ( ; it!=itEnd ; ++it )
 401     {
 402         (*it)->restartTransfer( nextUrl() );
 403     }
 404 }
 405
 406 Segment *SegmentFactory::takeLongest()
 407 {
 408     kDebug(5001);
 409
 410     Segment *longest = 0;
 411     KIO::filesize_t bytes = MultiSegKioSettings::splitSize()*1024;
 412
 413     QList<Segment*>::const_iterator it = m_Segments.constBegin();
 414     QList<Segment*>::const_iterator itEnd = m_Segments.constEnd();
 415     for ( ; it!=itEnd ; ++it )
 416     {
 417         if((*it)->data().bytes > bytes)
 418         {
 419             longest = (*it);
 420             bytes = (*it)->data().bytes;
 421         }
 422     }
 423
 424     if(longest)
 425         kDebug(5001) << "the longest segment has: " << longest->data().bytes;
 426
 427     return longest;
 428 }
 429
 430 const KUrl SegmentFactory::nextUrl()
 431 {
 432     kDebug(5001);
 433     if ( it_Urls >= m_Urls.constEnd() )
 434     {
 435         it_Urls = m_Urls.constBegin();
 436     }
 437     KUrl url(*it_Urls);
 438     it_Urls++;
 439     return url;
 440 }
 441
 442 bool SegmentFactory::DeleteUrl(const KUrl &url)
 443 {
 444     if ( m_Urls.count() == 1 )
 445     {
 446         return false;
 447     }
 448     if ( m_Urls.contains( url ) )
 449     {
 450         m_Urls.removeAll( url );
 451         return true;
 452     }
 453     return false;
 454 }
 455
 456 #include "segmentfactory.moc"