00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021 #include "resourcebase.h"
00022 #include "agentbase_p.h"
00023
00024 #include "resourceadaptor.h"
00025 #include "collectiondeletejob.h"
00026 #include "collectionsync.h"
00027 #include "itemsync.h"
00028 #include "resourcescheduler.h"
00029 #include "tracerinterface.h"
00030 #include "xdgbasedirs_p.h"
00031
00032 #include "changerecorder.h"
00033 #include "collectionfetchjob.h"
00034 #include "collectionmodifyjob.h"
00035 #include "itemfetchjob.h"
00036 #include "itemfetchscope.h"
00037 #include "itemmodifyjob.h"
00038 #include "itemmodifyjob_p.h"
00039 #include "session.h"
00040
00041 #include <kaboutdata.h>
00042 #include <kcmdlineargs.h>
00043 #include <kdebug.h>
00044 #include <klocale.h>
00045
00046 #include <QtCore/QDebug>
00047 #include <QtCore/QDir>
00048 #include <QtCore/QHash>
00049 #include <QtCore/QSettings>
00050 #include <QtCore/QTimer>
00051 #include <QtGui/QApplication>
00052 #include <QtDBus/QtDBus>
00053
00054 using namespace Akonadi;
00055
00056 class Akonadi::ResourceBasePrivate : public AgentBasePrivate
00057 {
00058 public:
00059 ResourceBasePrivate( ResourceBase *parent )
00060 : AgentBasePrivate( parent ),
00061 scheduler( 0 ),
00062 mItemSyncer( 0 )
00063 {
00064 mStatusMessage = defaultReadyMessage();
00065 }
00066
00067 Q_DECLARE_PUBLIC( ResourceBase )
00068
00069 void delayedInit()
00070 {
00071 if ( !QDBusConnection::sessionBus().registerService( QLatin1String( "org.freedesktop.Akonadi.Resource." ) + mId ) )
00072 kFatal() << "Unable to register service at D-Bus: " << QDBusConnection::sessionBus().lastError().message();
00073 AgentBasePrivate::delayedInit();
00074 }
00075
00076 virtual void changeProcessed()
00077 {
00078 mMonitor->changeProcessed();
00079 if ( !mMonitor->isEmpty() )
00080 scheduler->scheduleChangeReplay();
00081 scheduler->taskDone();
00082 }
00083
00084 void slotDeliveryDone( KJob* job );
00085 void slotCollectionSyncDone( KJob *job );
00086 void slotLocalListDone( KJob *job );
00087 void slotSynchronizeCollection( const Collection &col );
00088 void slotCollectionListDone( KJob *job );
00089
00090 void slotItemSyncDone( KJob *job );
00091
00092 void slotPercent( KJob* job, unsigned long percent );
00093 void slotDeleteResourceCollection();
00094 void slotDeleteResourceCollectionDone( KJob *job );
00095 void slotCollectionDeletionDone( KJob *job );
00096
00097 QString mName;
00098
00099
00100 Collection currentCollection;
00101
00102 ResourceScheduler *scheduler;
00103 ItemSync *mItemSyncer;
00104 };
00105
00106 ResourceBase::ResourceBase( const QString & id )
00107 : AgentBase( new ResourceBasePrivate( this ), id )
00108 {
00109 Q_D( ResourceBase );
00110
00111 new ResourceAdaptor( this );
00112
00113 const QString name = d->mSettings->value( QLatin1String( "Resource/Name" ) ).toString();
00114 if ( !name.isEmpty() )
00115 d->mName = name;
00116
00117 d->scheduler = new ResourceScheduler( this );
00118
00119 d->mMonitor->setChangeRecordingEnabled( true );
00120 connect( d->mMonitor, SIGNAL(changesAdded()),
00121 d->scheduler, SLOT(scheduleChangeReplay()) );
00122
00123 d->mMonitor->setResourceMonitored( d->mId.toLatin1() );
00124
00125 connect( d->scheduler, SIGNAL(executeFullSync()),
00126 SLOT(retrieveCollections()) );
00127 connect( d->scheduler, SIGNAL(executeCollectionTreeSync()),
00128 SLOT(retrieveCollections()) );
00129 connect( d->scheduler, SIGNAL(executeCollectionSync(Akonadi::Collection)),
00130 SLOT(slotSynchronizeCollection(Akonadi::Collection)) );
00131 connect( d->scheduler, SIGNAL(executeItemFetch(Akonadi::Item,QSet<QByteArray>)),
00132 SLOT(retrieveItem(Akonadi::Item,QSet<QByteArray>)) );
00133 connect( d->scheduler, SIGNAL(executeResourceCollectionDeletion()),
00134 SLOT(slotDeleteResourceCollection()) );
00135 connect( d->scheduler, SIGNAL( status( int, QString ) ),
00136 SIGNAL( status( int, QString ) ) );
00137 connect( d->scheduler, SIGNAL(executeChangeReplay()),
00138 d->mMonitor, SLOT(replayNext()) );
00139
00140 d->scheduler->setOnline( d->mOnline );
00141 if ( !d->mMonitor->isEmpty() )
00142 d->scheduler->scheduleChangeReplay();
00143 }
00144
00145 ResourceBase::~ResourceBase()
00146 {
00147 }
00148
00149 void ResourceBase::synchronize()
00150 {
00151 d_func()->scheduler->scheduleFullSync();
00152 }
00153
00154 void ResourceBase::setName( const QString &name )
00155 {
00156 Q_D( ResourceBase );
00157 if ( name == d->mName )
00158 return;
00159
00160
00161 d->mName = name;
00162
00163 if ( d->mName.isEmpty() || d->mName == d->mId )
00164 d->mSettings->remove( QLatin1String( "Resource/Name" ) );
00165 else
00166 d->mSettings->setValue( QLatin1String( "Resource/Name" ), d->mName );
00167
00168 d->mSettings->sync();
00169
00170 emit nameChanged( d->mName );
00171 }
00172
00173 QString ResourceBase::name() const
00174 {
00175 Q_D( const ResourceBase );
00176 if ( d->mName.isEmpty() )
00177 return d->mId;
00178 else
00179 return d->mName;
00180 }
00181
00182 static char* sAppName = 0;
00183
00184 QString ResourceBase::parseArguments( int argc, char **argv )
00185 {
00186 QString identifier;
00187 if ( argc < 3 ) {
00188 kDebug( 5250 ) << "Not enough arguments passed...";
00189 exit( 1 );
00190 }
00191
00192 for ( int i = 1; i < argc - 1; ++i ) {
00193 if ( QLatin1String( argv[ i ] ) == QLatin1String( "--identifier" ) )
00194 identifier = QLatin1String( argv[ i + 1 ] );
00195 }
00196
00197 if ( identifier.isEmpty() ) {
00198 kDebug( 5250 ) << "Identifier argument missing";
00199 exit( 1 );
00200 }
00201
00202 sAppName = qstrdup( identifier.toLatin1().constData() );
00203 KCmdLineArgs::init( argc, argv, sAppName, 0,
00204 ki18nc("@title, application name", "Akonadi Resource"), "0.1",
00205 ki18nc("@title, application description", "Akonadi Resource") );
00206
00207 KCmdLineOptions options;
00208 options.add("identifier <argument>",
00209 ki18nc("@label, commandline option", "Resource identifier"));
00210 KCmdLineArgs::addCmdLineOptions( options );
00211
00212 return identifier;
00213 }
00214
00215 int ResourceBase::init( ResourceBase *r )
00216 {
00217 QApplication::setQuitOnLastWindowClosed( false );
00218 int rv = kapp->exec();
00219 delete r;
00220 delete[] sAppName;
00221 return rv;
00222 }
00223
00224 void ResourceBase::itemRetrieved( const Item &item )
00225 {
00226 Q_D( ResourceBase );
00227 Q_ASSERT( d->scheduler->currentTask().type == ResourceScheduler::FetchItem );
00228 if ( !item.isValid() ) {
00229 QDBusMessage reply( d->scheduler->currentTask().dbusMsg );
00230 reply << false;
00231 QDBusConnection::sessionBus().send( reply );
00232 d->scheduler->taskDone();
00233 return;
00234 }
00235
00236 Item i( item );
00237 QSet<QByteArray> requestedParts = d->scheduler->currentTask().itemParts;
00238 foreach ( const QByteArray &part, requestedParts ) {
00239 if ( !item.loadedPayloadParts().contains( part ) ) {
00240 kWarning( 5250 ) << "Item does not provide part" << part;
00241 }
00242 }
00243
00244 ItemModifyJob *job = new ItemModifyJob( i );
00245
00246 job->disableRevisionCheck();
00247 connect( job, SIGNAL(result(KJob*)), SLOT(slotDeliveryDone(KJob*)) );
00248 }
00249
00250 void ResourceBasePrivate::slotDeliveryDone(KJob * job)
00251 {
00252 Q_Q( ResourceBase );
00253 Q_ASSERT( scheduler->currentTask().type == ResourceScheduler::FetchItem );
00254 QDBusMessage reply( scheduler->currentTask().dbusMsg );
00255 if ( job->error() ) {
00256 emit q->error( QLatin1String( "Error while creating item: " ) + job->errorString() );
00257 reply << false;
00258 } else {
00259 reply << true;
00260 }
00261 QDBusConnection::sessionBus().send( reply );
00262 scheduler->taskDone();
00263 }
00264
00265 void ResourceBasePrivate::slotDeleteResourceCollection()
00266 {
00267 Q_Q( ResourceBase );
00268
00269 CollectionFetchJob *job = new CollectionFetchJob( Collection::root(), CollectionFetchJob::FirstLevel );
00270 job->setResource( q->identifier() );
00271 connect( job, SIGNAL(result(KJob*)), q, SLOT(slotDeleteResourceCollectionDone(KJob*)) );
00272 }
00273
00274 void ResourceBasePrivate::slotDeleteResourceCollectionDone( KJob *job )
00275 {
00276 Q_Q( ResourceBase );
00277 if ( job->error() ) {
00278 emit q->error( job->errorString() );
00279 scheduler->taskDone();
00280 } else {
00281 const CollectionFetchJob *fetchJob = static_cast<const CollectionFetchJob*>( job );
00282
00283 if ( !fetchJob->collections().isEmpty() ) {
00284 CollectionDeleteJob *job = new CollectionDeleteJob( fetchJob->collections().first() );
00285 connect( job, SIGNAL( result( KJob* ) ), q, SLOT( slotCollectionDeletionDone( KJob* ) ) );
00286 } else {
00287
00288 scheduler->taskDone();
00289 }
00290 }
00291 }
00292
00293 void ResourceBasePrivate::slotCollectionDeletionDone( KJob *job )
00294 {
00295 Q_Q( ResourceBase );
00296 if ( job->error() ) {
00297 emit q->error( job->errorString() );
00298 }
00299
00300 scheduler->taskDone();
00301 }
00302
00303 void ResourceBase::changeCommitted(const Item& item)
00304 {
00305 Q_D( ResourceBase );
00306 ItemModifyJob *job = new ItemModifyJob( item );
00307 job->d_func()->setClean();
00308 job->disableRevisionCheck();
00309 d->changeProcessed();
00310 }
00311
00312 void ResourceBase::changeCommitted( const Collection &collection )
00313 {
00314 Q_D( ResourceBase );
00315 CollectionModifyJob *job = new CollectionModifyJob( collection );
00316 Q_UNUSED( job );
00317
00318 d->changeProcessed();
00319 }
00320
00321 bool ResourceBase::requestItemDelivery( qint64 uid, const QString & remoteId,
00322 const QString &mimeType, const QStringList &_parts )
00323 {
00324 Q_D( ResourceBase );
00325 if ( !isOnline() ) {
00326 emit error( i18nc( "@info", "Cannot fetch item in offline mode." ) );
00327 return false;
00328 }
00329
00330 setDelayedReply( true );
00331
00332 Item item( uid );
00333 item.setMimeType( mimeType );
00334 item.setRemoteId( remoteId );
00335
00336 QSet<QByteArray> parts;
00337 Q_FOREACH( const QString &str, _parts )
00338 parts.insert( str.toLatin1() );
00339
00340 d->scheduler->scheduleItemFetch( item, parts, message().createReply() );
00341
00342 return true;
00343 }
00344
00345 void ResourceBase::collectionsRetrieved(const Collection::List & collections)
00346 {
00347 Q_D( ResourceBase );
00348 CollectionSync *syncer = new CollectionSync( d->mId );
00349 syncer->setRemoteCollections( collections );
00350 connect( syncer, SIGNAL(result(KJob*)), SLOT(slotCollectionSyncDone(KJob*)) );
00351 }
00352
00353 void ResourceBase::collectionsRetrievedIncremental(const Collection::List & changedCollections, const Collection::List & removedCollections)
00354 {
00355 Q_D( ResourceBase );
00356 CollectionSync *syncer = new CollectionSync( d->mId );
00357 syncer->setRemoteCollections( changedCollections, removedCollections );
00358 connect( syncer, SIGNAL(result(KJob*)), SLOT(slotCollectionSyncDone(KJob*)) );
00359 }
00360
00361 void ResourceBasePrivate::slotCollectionSyncDone(KJob * job)
00362 {
00363 Q_Q( ResourceBase );
00364 if ( job->error() ) {
00365 emit q->error( job->errorString() );
00366 } else {
00367 if ( scheduler->currentTask().type == ResourceScheduler::SyncAll ) {
00368 CollectionFetchJob *list = new CollectionFetchJob( Collection::root(), CollectionFetchJob::Recursive );
00369 list->setResource( mId );
00370 q->connect( list, SIGNAL(result(KJob*)), q, SLOT(slotLocalListDone(KJob*)) );
00371 return;
00372 }
00373 }
00374 scheduler->taskDone();
00375 }
00376
00377 void ResourceBasePrivate::slotLocalListDone(KJob * job)
00378 {
00379 Q_Q( ResourceBase );
00380 if ( job->error() ) {
00381 emit q->error( job->errorString() );
00382 } else {
00383 Collection::List cols = static_cast<CollectionFetchJob*>( job )->collections();
00384 foreach ( const Collection &col, cols ) {
00385 scheduler->scheduleSync( col );
00386 }
00387 }
00388 scheduler->taskDone();
00389 }
00390
00391 void ResourceBasePrivate::slotSynchronizeCollection( const Collection &col )
00392 {
00393 Q_Q( ResourceBase );
00394 currentCollection = col;
00395
00396 QStringList contentTypes = currentCollection.contentMimeTypes();
00397 contentTypes.removeAll( Collection::mimeType() );
00398 if ( !contentTypes.isEmpty() ) {
00399 emit q->status( AgentBase::Running, i18nc( "@info:status", "Syncing collection '%1'", currentCollection.name() ) );
00400 q->retrieveItems( currentCollection );
00401 return;
00402 }
00403 scheduler->taskDone();
00404 }
00405
00406 void ResourceBase::itemsRetrievalDone()
00407 {
00408 Q_D( ResourceBase );
00409
00410 if ( d->mItemSyncer ) {
00411 d->mItemSyncer->deliveryDone();
00412 }
00413
00414 else {
00415 d->scheduler->taskDone();
00416 }
00417 }
00418
00419 void ResourceBase::clearCache()
00420 {
00421 Q_D( ResourceBase );
00422 d->scheduler->scheduleResourceCollectionDeletion();
00423 }
00424
00425 Collection ResourceBase::currentCollection() const
00426 {
00427 Q_D( const ResourceBase );
00428 Q_ASSERT_X( d->scheduler->currentTask().type == ResourceScheduler::SyncCollection ,
00429 "ResourceBase::currentCollection()",
00430 "Trying to access current collection although no item retrieval is in progress" );
00431 return d->currentCollection;
00432 }
00433
00434 Item ResourceBase::currentItem() const
00435 {
00436 Q_D( const ResourceBase );
00437 Q_ASSERT_X( d->scheduler->currentTask().type == ResourceScheduler::FetchItem ,
00438 "ResourceBase::currentItem()",
00439 "Trying to access current item although no item retrieval is in progress" );
00440 return d->scheduler->currentTask().item;
00441 }
00442
00443 void ResourceBase::synchronizeCollectionTree()
00444 {
00445 d_func()->scheduler->scheduleCollectionTreeSync();
00446 }
00447
00448 void ResourceBase::cancelTask()
00449 {
00450 d_func()->changeProcessed();
00451 }
00452
00453 void ResourceBase::cancelTask( const QString &msg )
00454 {
00455 cancelTask();
00456
00457 emit error( msg );
00458 }
00459
00460 void ResourceBase::doSetOnline( bool state )
00461 {
00462 d_func()->scheduler->setOnline( state );
00463 }
00464
00465 void ResourceBase::synchronizeCollection(qint64 collectionId )
00466 {
00467 CollectionFetchJob* job = new CollectionFetchJob( Collection(collectionId), CollectionFetchJob::Base );
00468 job->setResource( identifier() );
00469 connect( job, SIGNAL(result(KJob*)), SLOT(slotCollectionListDone(KJob*)) );
00470 }
00471
00472 void ResourceBasePrivate::slotCollectionListDone( KJob *job )
00473 {
00474 if ( !job->error() ) {
00475 Collection::List list = static_cast<CollectionFetchJob*>( job )->collections();
00476 if ( !list.isEmpty() ) {
00477 Collection col = list.first();
00478 scheduler->scheduleSync( col );
00479 }
00480 }
00481
00482 }
00483
00484 void ResourceBase::setTotalItems( int amount )
00485 {
00486 kDebug() << amount;
00487 Q_D( ResourceBase );
00488 setItemStreamingEnabled( true );
00489 d->mItemSyncer->setTotalItems( amount );
00490 }
00491
00492 void ResourceBase::setItemStreamingEnabled( bool enable )
00493 {
00494 Q_D( ResourceBase );
00495 Q_ASSERT_X( d->scheduler->currentTask().type == ResourceScheduler::SyncCollection,
00496 "ResourceBase::setItemStreamingEnabled()",
00497 "Calling setItemStreamingEnabled() although no item retrieval is in progress" );
00498 if ( !d->mItemSyncer ) {
00499 d->mItemSyncer = new ItemSync( currentCollection() );
00500 connect( d->mItemSyncer, SIGNAL(percent(KJob*,unsigned long)), SLOT(slotPercent(KJob*,unsigned long)) );
00501 connect( d->mItemSyncer, SIGNAL(result(KJob*)), SLOT(slotItemSyncDone(KJob*)) );
00502 }
00503 d->mItemSyncer->setStreamingEnabled( enable );
00504 }
00505
00506 void ResourceBase::itemsRetrieved( const Item::List &items )
00507 {
00508 Q_D( ResourceBase );
00509 Q_ASSERT_X( d->scheduler->currentTask().type == ResourceScheduler::SyncCollection,
00510 "ResourceBase::itemsRetrieved()",
00511 "Calling itemsRetrieved() although no item retrieval is in progress" );
00512 if ( !d->mItemSyncer ) {
00513 d->mItemSyncer = new ItemSync( currentCollection() );
00514 connect( d->mItemSyncer, SIGNAL(percent(KJob*,unsigned long)), SLOT(slotPercent(KJob*,unsigned long)) );
00515 connect( d->mItemSyncer, SIGNAL(result(KJob*)), SLOT(slotItemSyncDone(KJob*)) );
00516 }
00517 d->mItemSyncer->setFullSyncItems( items );
00518 }
00519
00520 void ResourceBase::itemsRetrievedIncremental(const Item::List &changedItems, const Item::List &removedItems)
00521 {
00522 Q_D( ResourceBase );
00523 Q_ASSERT_X( d->scheduler->currentTask().type == ResourceScheduler::SyncCollection,
00524 "ResourceBase::itemsRetrievedIncremental()",
00525 "Calling itemsRetrievedIncremental() although no item retrieval is in progress" );
00526 if ( !d->mItemSyncer ) {
00527 d->mItemSyncer = new ItemSync( currentCollection() );
00528 connect( d->mItemSyncer, SIGNAL(percent(KJob*,unsigned long)), SLOT(slotPercent(KJob*,unsigned long)) );
00529 connect( d->mItemSyncer, SIGNAL(result(KJob*)), SLOT(slotItemSyncDone(KJob*)) );
00530 }
00531 d->mItemSyncer->setIncrementalSyncItems( changedItems, removedItems );
00532 }
00533
00534 void ResourceBasePrivate::slotItemSyncDone( KJob *job )
00535 {
00536 mItemSyncer = 0;
00537 Q_Q( ResourceBase );
00538 if ( job->error() ) {
00539 emit q->error( job->errorString() );
00540 }
00541 scheduler->taskDone();
00542 }
00543
00544 void ResourceBasePrivate::slotPercent( KJob *job, unsigned long percent )
00545 {
00546 Q_Q( ResourceBase );
00547 Q_UNUSED( job );
00548 emit q->percent( percent );
00549 }
00550
00551 #include "resourcebase.moc"