commit 1e5e6a48f02d224f406f542ee4ab57a583cea854
Author: Stephen Kelly <steveire@gmail.com>
Date:   Sat Apr 23 10:38:59 2011 +0200

    Prefetch Collections if we are monitoring non-root collections
    
    We need to ensure that we have their ancestors before we test
    for acceptance.

diff --git a/akonadi/changerecorder.cpp b/akonadi/changerecorder.cpp
index 6988442..a5fc27e 100644
--- a/akonadi/changerecorder.cpp
+++ b/akonadi/changerecorder.cpp
@@ -59,6 +59,17 @@ void ChangeRecorder::replayNext()
   Q_D( ChangeRecorder );
   if ( !d->pendingNotifications.isEmpty() ) {
     const NotificationMessage msg = d->pendingNotifications.head();
+    if ( d->needPrefetch() && !d->ensurePrefetched( msg ) ) {
+      d->saveNotifications();
+      // acceptNotification results in either a message emission or
+      // replayNext being called again.
+      return;
+    }
+    if ( !d->acceptNotification( msg ) ) {
+      d->pendingNotifications.removeFirst();
+      // replayNext invoked asyncronously by ChangeRecorderPrivate::acceptNotification
+      return;
+    }
     if ( d->ensureDataAvailable( msg ) )
       d->emitNotification( msg );
     else if ( !d->translateAndCompress( d->pipeline, msg ) ) {
diff --git a/akonadi/changerecorder_p.h b/akonadi/changerecorder_p.h
index 4314b40..89d99cd 100644
--- a/akonadi/changerecorder_p.h
+++ b/akonadi/changerecorder_p.h
@@ -44,6 +44,14 @@ class AKONADI_TESTS_EXPORT Akonadi::ChangeRecorderPrivate : public Akonadi::Moni
       return MonitorPrivate::pipelineSize();
     }
 
+    virtual bool acceptNotification( const NotificationMessage &msg )
+    {
+      const bool accepted = MonitorPrivate::acceptNotification( msg );
+      if ( enableChangeRecording && !accepted )
+        QMetaObject::invokeMethod( q_ptr, "replayNext", Qt::QueuedConnection ); // skip notifications noone was listening to
+      return accepted;
+    }
+
     virtual void slotNotify( const NotificationMessage::List &msgs )
     {
       Q_Q( ChangeRecorder );
diff --git a/akonadi/monitor.cpp b/akonadi/monitor.cpp
index 54cbc39..4fc85bb 100644
--- a/akonadi/monitor.cpp
+++ b/akonadi/monitor.cpp
@@ -66,6 +66,9 @@ void Monitor::setCollectionMonitored( const Collection &collection, bool monitor
   Q_D( Monitor );
   if ( monitored ) {
     d->collections << collection;
+    if ( collection != Collection::root() ) {
+      d->mCollectionFetchScope.setAncestorRetrieval( Akonadi::CollectionFetchScope::All );
+    }
   } else {
     d->collections.removeAll( collection );
     d->cleanOldNotifications();
diff --git a/akonadi/monitor_p.cpp b/akonadi/monitor_p.cpp
index e65e975..ce6019e 100644
--- a/akonadi/monitor_p.cpp
+++ b/akonadi/monitor_p.cpp
@@ -87,8 +87,8 @@ MonitorPrivate::MonitorPrivate( MonitorDependeciesFactory *dependenciesFactory_,
 
 void MonitorPrivate::init()
 {
-  // needs to be at least 3x pipeline size for the collection move case
-  collectionCache = dependenciesFactory->createCollectionCache(3*PipelineSize, session);
+  // needs to be at least 6x pipeline size for the collection move case
+  collectionCache = dependenciesFactory->createCollectionCache(6*PipelineSize, session);
   // needs to be at least 1x pipeline size
   itemCache = dependenciesFactory->createItemCache(PipelineSize, session);
 
@@ -204,10 +204,21 @@ bool MonitorPrivate::acceptNotification( const NotificationMessage & msg )
         return false;
       }
 
+      if ( needPrefetch() && msg.operation() == NotificationMessage::Remove ) {
+        // The parent collection is already invalid in the case of leaves which are children of
+        // a collection which is also being removed. We can ignore everything with an invalid
+        // parent and we will later get at least one notification resulting from this removal which has
+        // a non-invalid parentCollection. That case is handled normally by removing
+        // the (ancestor) collection which was actually removed.
+        if ( !collectionCache->retrieve( msg.parentCollection() ).isValid() ) {
+          return false;
+        }
+      }
+
       // we explicitly monitor that item or the collections it's in
       return items.contains( msg.uid() )
-          || isCollectionMonitored( msg.parentCollection() )
-          || isCollectionMonitored( msg.parentDestCollection() );
+          || isCollectionMonitored( msg.parentCollection(), true )
+          || isCollectionMonitored( msg.parentDestCollection(), true );
 
     case NotificationMessage::Collection:
       // we have a resource filter
@@ -220,10 +231,27 @@ bool MonitorPrivate::acceptNotification( const NotificationMessage & msg )
         // else continue
       }
 
+      if ( needPrefetch() && msg.operation() == NotificationMessage::Remove ) {
+        // If an Item, Collection or subtree is removed, we need to notify about it if it is a
+        // collection we explicitly monitor...
+        if ( collections.contains( Collection( msg.uid() ) ) ) {
+          Q_Q( Monitor );
+          q->setCollectionMonitored( Collection( msg.uid() ), false );
+          return true;
+        }
+        // ... Otherwise, the parent collection is already invalid in the case of leaves which are children of
+        // a collection which is also being removed. We can ignore everything with an invalid
+        // parent and we will later get at least one notification resulting from this removal which has
+        // a non-invalid parentCollection. That case is handled normally below.
+        if ( !collectionCache->retrieve( msg.parentCollection() ).isValid() ) {
+          return false;
+        }
+      }
+
       // we explicitly monitor that colleciton, or all of them
       return isCollectionMonitored( msg.uid() )
-          || isCollectionMonitored( msg.parentCollection() )
-          || isCollectionMonitored( msg.parentDestCollection() );
+          || isCollectionMonitored( msg.parentCollection(), true )
+          || isCollectionMonitored( msg.parentDestCollection(), true );
   }
   Q_ASSERT( false );
   return false;
@@ -357,9 +385,9 @@ bool MonitorPrivate::translateAndCompress( QQueue<NotificationMessage> &notifica
       destWatched = isMoveDestinationResourceMonitored( msg );
     }
     if ( !sourceWatched )
-      sourceWatched = isCollectionMonitored( msg.parentCollection() );
+      sourceWatched = isCollectionMonitored( msg.parentCollection(), true );
     if ( !destWatched )
-      destWatched = isCollectionMonitored( msg.parentDestCollection() );
+      destWatched = isCollectionMonitored( msg.parentDestCollection(), true );
   }
 
   if ( !sourceWatched && !destWatched )
@@ -393,27 +421,51 @@ bool MonitorPrivate::translateAndCompress( QQueue<NotificationMessage> &notifica
 }
 
 /*
+  If a message is not accepted it is discarded and the process continues.
 
-  server notification --> ?accepted --> pendingNotifications --> ?dataAvailable --> emit
-                                  |                                           |
-                                  x --> discard                               x --> pipeline
+  serverNotify or job completed  --> pipeline --> ?dataAvailable --> emit --\
+                                                                            |
+     ,----------------------------------------------------------------------'
+     |
+     '-->   pendingNotifications --> ?prefetched --> ?accepted --> pipeline --> ?dataAvailable --> emit
 
-  fetchJobDone --> pipeline ?dataAvailable --> emit
- */
+           '--------------------------------------------- loop ----------------------------------------'
+*/
 
 void MonitorPrivate::slotNotify( const NotificationMessage::List &msgs )
 {
   foreach ( const NotificationMessage &msg, msgs ) {
     invalidateCaches( msg );
-    updatePendingStatistics( msg );
-    if ( acceptNotification( msg ) ) {
-      translateAndCompress( pendingNotifications, msg );
+    if ( !sessions.contains( msg.sessionId() ) && !isLazilyIgnored( msg ) ) {
+      if ( needPrefetch() )
+        ensurePrefetched( msg );
+      pendingNotifications.enqueue( msg );
     }
   }
-
   dispatchNotifications();
 }
 
+void MonitorPrivate::filterMessages()
+{
+  while ( pipeline.size() < pipelineSize() && !pendingNotifications.isEmpty() ) {
+    const NotificationMessage msg = pendingNotifications.head();
+    if ( needPrefetch() && !ensurePrefetched( msg ) )
+      return;
+    pendingNotifications.dequeue();
+    updatePendingStatistics( msg );
+    if ( acceptNotification( msg ) )
+      translateAndCompress( pipeline, msg );
+  }
+}
+
+bool MonitorPrivate::ensurePrefetched( const NotificationMessage &msg )
+{
+  return collectionCache->ensureCached( msg.parentCollection(), mCollectionFetchScope )
+      && ( msg.parentDestCollection() < 0 ? true :
+          collectionCache->ensureCached( msg.parentDestCollection(), mCollectionFetchScope ) );
+
+}
+
 void MonitorPrivate::flushPipeline()
 {
   while ( !pipeline.isEmpty() ) {
@@ -437,12 +489,24 @@ void MonitorPrivate::dataAvailable()
 
 void MonitorPrivate::dispatchNotifications()
 {
-  while ( pipeline.size() < pipelineSize() && !pendingNotifications.isEmpty() ) {
-    const NotificationMessage msg = pendingNotifications.dequeue();
-    if ( ensureDataAvailable( msg ) && pipeline.isEmpty() )
-      emitNotification( msg );
-    else
-      pipeline.enqueue( msg );
+  const bool piplineWasEmpty = pipeline.isEmpty();
+  if ( !pendingNotifications.isEmpty() )
+    filterMessages();
+
+  // If the pipeline was non-empty then head() still needs data to be fetched.
+  // Otherwise, the filterMessages call may have populated it with stuff we can emit.
+  if ( piplineWasEmpty ) {
+    const bool pipelineWasFull = !pipeline.isEmpty() && pipeline.size() == pipelineSize();
+    flushPipeline();
+
+    // We could have a situation where all data required for messages in pendingNotifications
+    // is available and filterMessages causes the pipeline to be filled. It then gets flushed
+    // by flushPipeline(), but because all data for messages in pendingNotifications is already
+    // available, there are no running jobs, and dataAvailable won't be emitted again. To avoid
+    // data loss we handle the special case by filtering and flushing again if needed.
+    if ( pipelineWasFull && !pendingNotifications.isEmpty() ) {
+      dispatchNotifications();
+    }
   }
 }
 
@@ -717,7 +781,7 @@ Akonadi::Collection::Id MonitorPrivate::PurgeBuffer::buffer( Collection::Id id )
 }
 
 void MonitorPrivate::notifyCollectionStatisticsWatchers(Entity::Id collection, const QByteArray& resource) {
-  if ( collection > 0 && (monitorAll || isCollectionMonitored( collection ) || resources.contains( resource ) ) ) {
+  if ( collection > 0 && (monitorAll || isCollectionMonitored( collection, true ) || resources.contains( resource ) ) ) {
     recentlyChangedCollections.insert( collection );
     if ( !statisticsCompressionTimer.isActive() )
       statisticsCompressionTimer.start();
diff --git a/akonadi/monitor_p.h b/akonadi/monitor_p.h
index a473d16..1c794dd 100644
--- a/akonadi/monitor_p.h
+++ b/akonadi/monitor_p.h
@@ -91,7 +91,7 @@ class AKONADI_TESTS_EXPORT MonitorPrivate
 
     // Virtual so it can be overridden in FakeMonitor.
     virtual bool connectToNotificationManager();
-    bool acceptNotification( const NotificationMessage &msg );
+    virtual bool acceptNotification( const NotificationMessage &msg );
     void dispatchNotifications();
     void flushPipeline();
 
@@ -113,6 +113,13 @@ class AKONADI_TESTS_EXPORT MonitorPrivate
     */
     void invalidateCache( const Collection &col );
 
+    inline bool needPrefetch() const {
+      return !(collections.isEmpty() || collections.contains(Collection::root()));
+    }
+
+    bool ensurePrefetched( const NotificationMessage &msg );
+    void filterMessages();
+
     virtual int pipelineSize() const;
 
     // private slots
@@ -206,14 +213,26 @@ class AKONADI_TESTS_EXPORT MonitorPrivate
     */
     bool isLazilyIgnored( const NotificationMessage & msg ) const;
 
-    bool isCollectionMonitored( Collection::Id collection ) const
+    bool isCollectionMonitored( Collection::Id collection, bool recurs = false ) const
     {
       if (collection < 0)
         return false;
-      if ( collections.contains( Collection( collection ) ) )
-        return true;
       if ( collections.contains( Collection::root() ) )
         return true;
+      if ( collections.contains( Collection( collection ) ) )
+        return true;
+      if ( recurs && needPrefetch() ) {
+        Collection col = collectionCache->retrieve( collection );
+        col = col.parentCollection();
+        while ( col.isValid() && col != Collection::root() ) {
+          if ( collections.contains( col ) )
+            return true;
+          col = col.parentCollection();
+        }
+        if ( !col.isValid() ) {
+          qWarning() << "Invalid collection in prefetch";
+        }
+      }
       return false;
     }
 
diff --git a/akonadi/tests/monitornotificationtest.cpp b/akonadi/tests/monitornotificationtest.cpp
index 3ff308d..ba2c48b 100644
--- a/akonadi/tests/monitornotificationtest.cpp
+++ b/akonadi/tests/monitornotificationtest.cpp
@@ -42,15 +42,23 @@ public:
 private Q_SLOTS:
   void testSingleMessage();
   void testFillPipeline();
+  void testPrefetch();
+  void testPrefetchFillPipeline();
 
   void testSingleMessage_data();
   void testFillPipeline_data();
+  void testPrefetch_data();
+  void testPrefetchFillPipeline_data();
 
 private:
   template<typename MonitorImpl>
   void testSingleMessage_impl(MonitorImpl *monitor, FakeCollectionCache *collectionCache, FakeItemCache *itemCache);
   template<typename MonitorImpl>
   void testFillPipeline_impl(MonitorImpl *monitor, FakeCollectionCache *collectionCache, FakeItemCache *itemCache);
+  template<typename MonitorImpl>
+  void testPrefetch_impl(MonitorImpl *monitor, FakeCollectionCache *collectionCache, FakeItemCache *itemCache);
+  template<typename MonitorImpl>
+  void testPrefetchFillPipeline_impl(MonitorImpl *monitor, FakeCollectionCache *collectionCache, FakeItemCache *itemCache);
 
 private:
   FakeSession *m_fakeSession;
@@ -200,5 +208,251 @@ void MonitorNotificationTest::testFillPipeline_impl(MonitorImpl *monitor, FakeCo
   QVERIFY(monitor->pendingNotifications().isEmpty());
 }
 
+void MonitorNotificationTest::testPrefetch_data()
+{
+  QTest::addColumn<bool>("useChangeRecorder");
+
+  QTest::newRow("useChangeRecorder") << true;
+  QTest::newRow("useMonitor") << false;
+}
+
+void MonitorNotificationTest::testPrefetch()
+{
+  QFETCH(bool, useChangeRecorder);
+
+  FakeCollectionCache *collectionCache = new FakeCollectionCache(m_fakeSession);
+  FakeItemCache *itemCache = new FakeItemCache(m_fakeSession);
+  FakeMonitorDependeciesFactory *depsFactory = new FakeMonitorDependeciesFactory(itemCache, collectionCache);
+
+  if (!useChangeRecorder)
+  {
+    testPrefetch_impl(new InspectableMonitor(depsFactory, this), collectionCache, itemCache);
+  } else {
+    InspectableChangeRecorder *changeRecorder = new InspectableChangeRecorder(depsFactory, this);
+    changeRecorder->setChangeRecordingEnabled(false);
+    testPrefetch_impl(changeRecorder, collectionCache, itemCache);
+  }
+}
+
+template <typename MonitorImpl>
+void MonitorNotificationTest::testPrefetch_impl(MonitorImpl *monitor, FakeCollectionCache *collectionCache, FakeItemCache *itemCache)
+{
+  Q_UNUSED(itemCache)
+
+  monitor->setSession(m_fakeSession);
+  monitor->fetchCollection(true);
+
+  monitor->setCollectionMonitored(Collection(2));
+
+  NotificationMessage::List list;
+
+  Collection col2(2);
+  col2.setParentCollection(Collection::root());
+
+  collectionCache->insert(col2);
+
+  int i = 4;
+
+  while (i < 10) {
+    Collection added(i++);
+
+    NotificationMessage msg;
+    msg.setParentCollection(added.id() % 2 == 0 ? 2 : added.id() - 1);
+    msg.setOperation(Akonadi::NotificationMessage::Add);
+    msg.setType(Akonadi::NotificationMessage::Collection);
+    msg.setUid(added.id());
+
+    list << msg;
+  }
+
+  QVERIFY(monitor->pipeline().isEmpty());
+  QVERIFY(monitor->pendingNotifications().isEmpty());
+
+  Collection col4(4);
+  col4.setParentCollection(col2);
+  Collection col6(6);
+  col6.setParentCollection(col2);
+
+  collectionCache->insert(col4);
+  collectionCache->insert(col6);
+
+  collectionCache->emitDataAvailable();
+
+  monitor->notifier()->emitNotify(list);
+
+  QVERIFY(monitor->pipeline().size() == 4);
+  QVERIFY(monitor->pendingNotifications().size() == 1);
+
+  Collection col7(7);
+  col7.setParentCollection(col6);
+
+  collectionCache->insert(col7);
+  collectionCache->emitDataAvailable();
+
+  QVERIFY(monitor->pipeline().size() == 4);
+  QVERIFY(monitor->pendingNotifications().size() == 1);
+
+  Collection col5(5);
+  col5.setParentCollection(col4);
+
+  collectionCache->insert(col5);
+  collectionCache->emitDataAvailable();
+
+  QVERIFY(monitor->pipeline().size() == 1);
+  QVERIFY(monitor->pendingNotifications().size() == 1);
+
+  list.clear();
+
+  QHash<Collection::Id, Collection> extraCollections;
+
+  while (i < 20) {
+    Collection added(i++);
+
+    NotificationMessage msg;
+    msg.setParentCollection(added.id() % 2 == 0 ? 2 : added.id() - 1);
+    msg.setOperation(Akonadi::NotificationMessage::Add);
+    msg.setType(Akonadi::NotificationMessage::Collection);
+    msg.setUid(added.id());
+
+    added.setParentCollection(added.id() % 2 == 0 ? col2 : extraCollections.value(added.id() - 1));
+
+    extraCollections.insert(added.id(), added);
+
+    list << msg;
+  }
+  monitor->notifier()->emitNotify(list);
+
+  Collection col8(8);
+  col8.setParentCollection(col2);
+
+  collectionCache->insert(col8);
+  collectionCache->emitDataAvailable();
+
+  QVERIFY(monitor->pipeline().size() == 2);
+  QVERIFY(monitor->pendingNotifications().size() == 9);
+
+  Collection col9(9);
+  col9.setParentCollection(col8);
+  collectionCache->insert(col9);
+  collectionCache->emitDataAvailable();
+
+  QVERIFY(monitor->pipeline().size() == 1);
+  QVERIFY(monitor->pendingNotifications().size() == 9);
+
+  {
+    QHash<Collection::Id, Collection>::const_iterator it = extraCollections.constBegin();
+    const QHash<Collection::Id, Collection>::const_iterator end = extraCollections.constEnd();
+
+    for (; it != end; ++it)
+      collectionCache->insert(*it);
+    collectionCache->emitDataAvailable();
+  }
+  QVERIFY(monitor->pipeline().isEmpty());
+  QVERIFY(monitor->pendingNotifications().isEmpty());
+
+}
+
+void MonitorNotificationTest::testPrefetchFillPipeline_data()
+{
+  QTest::addColumn<bool>("useChangeRecorder");
+
+  QTest::newRow("useChangeRecorder") << true;
+  QTest::newRow("useMonitor") << false;
+}
+
+void MonitorNotificationTest::testPrefetchFillPipeline()
+{
+  QFETCH(bool, useChangeRecorder);
+
+  FakeCollectionCache *collectionCache = new FakeCollectionCache(m_fakeSession);
+  FakeItemCache *itemCache = new FakeItemCache(m_fakeSession);
+  FakeMonitorDependeciesFactory *depsFactory = new FakeMonitorDependeciesFactory(itemCache, collectionCache);
+
+  if (!useChangeRecorder)
+  {
+    testPrefetchFillPipeline_impl(new InspectableMonitor(depsFactory, this), collectionCache, itemCache);
+  } else {
+    InspectableChangeRecorder *changeRecorder = new InspectableChangeRecorder(depsFactory, this);
+    changeRecorder->setChangeRecordingEnabled(false);
+    testPrefetchFillPipeline_impl(changeRecorder, collectionCache, itemCache);
+  }
+}
+
+template <typename MonitorImpl>
+void MonitorNotificationTest::testPrefetchFillPipeline_impl(MonitorImpl *monitor, FakeCollectionCache *collectionCache, FakeItemCache *itemCache)
+{
+  Q_UNUSED(itemCache)
+
+  monitor->setSession(m_fakeSession);
+  monitor->fetchCollection(true);
+
+  monitor->setCollectionMonitored(Collection(2));
+
+  NotificationMessage::List list;
+  QHash<Collection::Id, Collection> data;
+
+  Collection col2(2);
+  col2.setParentCollection(Collection::root());
+
+  collectionCache->insert(col2);
+
+  data.insert(col2.id(), col2);
+
+  NotificationMessage col4Msg;
+  col4Msg.setParentCollection(3);
+  col4Msg.setOperation(Akonadi::NotificationMessage::Add);
+  col4Msg.setType(Akonadi::NotificationMessage::Collection);
+  col4Msg.setUid(4);
+
+  list << col4Msg;
+
+  int i = 5;
+
+  while (i < 30) {
+    Collection added(i++);
+
+    NotificationMessage msg;
+    msg.setParentCollection(col2.id());
+    msg.setOperation(Akonadi::NotificationMessage::Add);
+    msg.setType(Akonadi::NotificationMessage::Collection);
+    msg.setUid(added.id());
+
+    added.setParentCollection(col2);
+    collectionCache->insert(added);
+
+    list << msg;
+  }
+
+  QVERIFY(monitor->pipeline().isEmpty());
+  QVERIFY(monitor->pendingNotifications().isEmpty());
+
+  monitor->notifier()->emitNotify(list);
+
+  QVERIFY(monitor->pipeline().isEmpty());
+  QVERIFY(monitor->pendingNotifications().size() == 26);
+  QVERIFY(monitor->pendingNotifications().first().uid() == col4Msg.uid());
+
+  Collection col3(3);
+  col3.setParentCollection(col2);
+  collectionCache->insert(col3);
+
+  collectionCache->emitDataAvailable();
+
+  // Full pipeline
+  QVERIFY(monitor->pipeline().size() == 5);
+  QVERIFY(monitor->pipeline().first().uid() == col4Msg.uid());
+  QVERIFY(monitor->pendingNotifications().size() == 21);
+
+  Collection col4(4);
+  col4.setParentCollection(col3);
+  collectionCache->insert(col4);
+
+  collectionCache->emitDataAvailable();
+
+  QVERIFY(monitor->pipeline().isEmpty());
+  QVERIFY(monitor->pendingNotifications().isEmpty());
+}
+
+
 QTEST_MAIN( MonitorNotificationTest )
 #include "monitornotificationtest.moc"

