commit da686886d6b3b82216c371f6bd19b109c1fb7052
author Zhenkai Zhu <zhenkai@cs.ucla.edu> Tue Jan 29 22:32:24 2013 -0800
committer Zhenkai Zhu <zhenkai@cs.ucla.edu> Tue Jan 29 22:32:24 2013 -0800
tree 6e2e49a329c40ee0077f71c7a06d392cfec52161
parent b8c49affc8da421620e28adfb655a79e032023e6

add fetch-task-db and integrate it with fetch-manager

src/fetch-manager.cc

diff --git a/src/fetch-manager.cc b/src/fetch-manager.cc
index 8d98207..e01df78 100644
--- a/src/fetch-manager.cc
+++ b/src/fetch-manager.cc
@@ -44,6 +44,7 @@
                 , uint32_t parallelFetches // = 3
                 , const SegmentCallback &defaultSegmentCallback
                 , const FinishCallback &defaultFinishCallback
+                , const FetchTaskDbPtr &taskDb
                 )
   : m_ccnx (ccnx)
   , m_mapping (mapping)
@@ -53,10 +54,17 @@
   , m_executor (new Executor(1))
   , m_defaultSegmentCallback(defaultSegmentCallback)
   , m_defaultFinishCallback(defaultFinishCallback)
+  , m_taskDb(taskDb)
 {
   m_scheduler->start ();
   m_executor->start();
 
+  // resume un-finished fetches if there is any
+  if (m_taskDb)
+  {
+    m_taskDb->foreachTask(bind(&FetchManager::Enqueue, this, _1, _2, _3, _4, _5));
+  }
+
   m_scheduleFetchesTask = Scheduler::schedulePeriodicTask (m_scheduler,
                                                            make_shared<SimpleIntervalGenerator> (300), // no need to check to often. if needed, will be rescheduled
                                                            bind (&FetchManager::ScheduleFetches, this), SCHEDULE_FETCHES_TAG);
@@ -94,6 +102,8 @@
   Name forwardingHint;
   forwardingHint = m_mapping (deviceName);
 
+  m_taskDb->addTask(deviceName, baseName, minSeqNo, maxSeqNo, priority);
+
   unique_lock<mutex> lock (m_parellelFetchMutex);
 
   _LOG_TRACE ("++++ Create fetcher: " << baseName);
@@ -101,7 +111,7 @@
                                   m_executor,
                                   segmentCallback,
                                   finishCallback,
-                                  bind (&FetchManager::DidFetchComplete, this, _1),
+                                  bind (&FetchManager::DidFetchComplete, this, _1, _2, _3),
                                   bind (&FetchManager::DidNoDataTimeout, this, _1),
                                   deviceName, baseName, minSeqNo, maxSeqNo,
                                   boost::posix_time::seconds (30),
@@ -205,13 +215,14 @@
 }
 
 void
-FetchManager::DidFetchComplete (Fetcher &fetcher)
+FetchManager::DidFetchComplete (Fetcher &fetcher, const Name &deviceName, const Name &baseName)
 {
   {
     unique_lock<mutex> lock (m_parellelFetchMutex);
     m_currentParallelFetches --;
     _LOG_TRACE ("+++++ removing fetcher: " << fetcher.GetName ());
     m_fetchList.erase_and_dispose (FetchList::s_iterator_to (fetcher), fetcher_disposer ());
+    m_taskDb->deleteTask(deviceName, baseName);
   }
 
   m_scheduler->rescheduleTaskAt (m_scheduleFetchesTask, 0);