diff --git a/ccnx/ccnx-wrapper.h b/ccnx/ccnx-wrapper.h
index 2e66a2d..1127316 100644
--- a/ccnx/ccnx-wrapper.h
+++ b/ccnx/ccnx-wrapper.h
@@ -41,10 +41,10 @@
   sendInterest (const Name &interest, const Closure &closure, const Selectors &selector = Selectors());
 
   virtual int
-  publishData (const Name &name, const unsigned char *buf, size_t len, int freshness);
+  publishData (const Name &name, const unsigned char *buf, size_t len, int freshness = 2147/* max value for ccnx*/);
 
   int
-  publishData (const Name &name, const Bytes &content, int freshness);
+  publishData (const Name &name, const Bytes &content, int freshness = 2147/* max value for ccnx*/);
 
   static Name
   getLocalPrefix ();
diff --git a/src/fetch-manager.cc b/src/fetch-manager.cc
index 2b43c62..cb94992 100644
--- a/src/fetch-manager.cc
+++ b/src/fetch-manager.cc
@@ -54,8 +54,20 @@
 FetchManager::Enqueue (const Ccnx::Name &deviceName, uint32_t minSeqNo, uint32_t maxSeqNo, int priority/*=PRIORITY_NORMAL*/)
 {
   // we may need to guarantee that LookupLocator will gives an answer and not throw exception...
-  Name forwardingHint = m_sync->LookupLocator (deviceName);
-  Fetcher &fetcher = *(new Fetcher (*this, deviceName, minSeqNo, maxSeqNo, forwardingHint));
+  Name forwardingHint;
+  try {
+    forwardingHint = m_sync->LookupLocator (deviceName);
+  }
+  catch (Error::Db &exception) {
+    // just ignore for now
+  }
+
+  Fetcher &fetcher = *(new Fetcher (m_ccnx,
+                                    bind (&FetchManager::DidDataSegmentFetched, this, _1, _2, _3, _4, _5),
+                                    bind (&FetchManager::DidFetchComplete, this, _1),
+                                    bind (&FetchManager::DidNoDataTimeout, this, _1),
+                                    deviceName, minSeqNo, maxSeqNo
+                                    /* Alex: should or should not include hint initially?*/));
 
   switch (priority)
     {
@@ -81,7 +93,7 @@
        m_currentParallelFetches < m_maxParallelFetches && item != m_fetchList.end ();
        item++)
     {
-      if (item->m_active)
+      if (item->IsActive ())
         continue;
 
       m_currentParallelFetches ++;
@@ -90,10 +102,16 @@
 }
 
 void
+FetchManager::DidDataSegmentFetched (Fetcher &fetcher, uint32_t seqno, const Ccnx::Name &basename,
+                                     const Ccnx::Name &name, const Bytes &data)
+{
+  // do something
+}
+
+void
 FetchManager::DidNoDataTimeout (Fetcher &fetcher)
 {
-  fetcher.m_forwardingHint = Ccnx::Name ("/ndn/broadcast");
-  fetcher.m_active = false;
+  fetcher.SetForwardingHint (Ccnx::Name ("/ndn/broadcast"));
   {
     unique_lock<mutex> lock (m_parellelFetchMutex);
     m_currentParallelFetches --;
diff --git a/src/fetch-manager.h b/src/fetch-manager.h
index 73a0699..496d0fd 100644
--- a/src/fetch-manager.h
+++ b/src/fetch-manager.h
@@ -57,7 +57,11 @@
   inline SchedulerPtr
   GetScheduler ();
 
-  // Events called from Fetcher
+  // Fetch Events
+  void
+  DidDataSegmentFetched (Fetcher &fetcher, uint32_t seqno, const Ccnx::Name &basename,
+                         const Ccnx::Name &name, const Ccnx::Bytes &data);
+
   void
   DidNoDataTimeout (Fetcher &fetcher);
 
diff --git a/src/fetcher.cc b/src/fetcher.cc
index 6afad5b..876cab4 100644
--- a/src/fetcher.cc
+++ b/src/fetcher.cc
@@ -25,18 +25,29 @@
 #include <boost/make_shared.hpp>
 #include <boost/ref.hpp>
 #include <boost/throw_exception.hpp>
+#include <boost/date_time/posix_time/posix_time.hpp>
 
 using namespace boost;
 using namespace std;
 using namespace Ccnx;
 
-Fetcher::Fetcher (FetchManager &fetchManger,
+Fetcher::Fetcher (CcnxWrapperPtr ccnx,
+                  OnDataSegmentCallback onDataSegment,
+                  OnFetchCompleteCallback onFetchComplete, OnFetchFailedCallback onFetchFailed,
                   const Ccnx::Name &name, int32_t minSeqNo, int32_t maxSeqNo,
+                  boost::posix_time::time_duration timeout/* = boost::posix_time::seconds (30)*/,
                   const Ccnx::Name &forwardingHint/* = Ccnx::Name ()*/)
-  : m_fetchManager (fetchManger)
+  : m_ccnx (ccnx)
+
+  , m_onDataSegment (onDataSegment)
+  , m_onFetchComplete (onFetchComplete)
+  , m_onFetchFailed (onFetchFailed)
+
   , m_active (false)
   , m_name (name)
   , m_forwardingHint (forwardingHint)
+  , m_maximumNoActivityPeriod (timeout)
+
   , m_minSendSeqNo (-1)
   , m_maxInOrderRecvSeqNo (-1)
   , m_minSeqNo (minSeqNo)
@@ -56,28 +67,37 @@
 {
   m_active = true;
   m_minSendSeqNo = m_maxInOrderRecvSeqNo;
+  m_lastPositiveActivity = date_time::second_clock<boost::posix_time::ptime>::universal_time();
 
   FillPipeline ();
 }
 
 void
+Fetcher::SetForwardingHint (const Ccnx::Name &forwardingHint)
+{
+  m_forwardingHint = forwardingHint;
+}
+
+void
 Fetcher::FillPipeline ()
 {
   for (; m_minSendSeqNo < m_maxSeqNo && m_activePipeline < m_pipeline; m_minSendSeqNo++)
     {
-      m_fetchManager.GetCcnx ()
-        ->sendInterest (Name (m_name)("file")(m_minSendSeqNo+1),
-                        Closure (bind(&Fetcher::OnData, this, m_minSendSeqNo+1, _1, _2),
-                                 bind(&Fetcher::OnTimeout, this, m_minSendSeqNo+1, _1)));
+      m_ccnx->sendInterest (Name (m_name)(m_minSendSeqNo+1),
+                            Closure (bind(&Fetcher::OnData, this, m_minSendSeqNo+1, _1, _2),
+                                     bind(&Fetcher::OnTimeout, this, m_minSendSeqNo+1, _1)));
 
       m_activePipeline ++;
     }
 }
 
 void
-Fetcher::OnData (uint32_t seqno, const Ccnx::Name &name, const Ccnx::Bytes &)
+Fetcher::OnData (uint32_t seqno, const Ccnx::Name &name, const Ccnx::Bytes &content)
 {
+  m_onDataSegment (*this, seqno, m_name, name, content);
+
   m_activePipeline --;
+  m_lastPositiveActivity = date_time::second_clock<boost::posix_time::ptime>::universal_time();
 
   ////////////////////////////////////////////////////////////////////////////
   m_outOfOrderRecvSeqNo.insert (seqno);
@@ -95,18 +115,36 @@
   m_outOfOrderRecvSeqNo.erase (m_outOfOrderRecvSeqNo.begin (), inOrderSeqNo);
   ////////////////////////////////////////////////////////////////////////////
 
-  FillPipeline ();
-  // bla bla
-  if (0)
+  if (m_maxInOrderRecvSeqNo == m_maxSeqNo)
     {
       m_active = false;
-      m_fetchManager.DidFetchComplete (*this);
+      m_onFetchComplete (*this);
+    }
+  else
+    {
+      FillPipeline ();
     }
 }
 
 Closure::TimeoutCallbackReturnValue
 Fetcher::OnTimeout (uint32_t seqno, const Ccnx::Name &name)
 {
-  // Closure::RESULT_REEXPRESS
-  return Closure::RESULT_OK;
+  // cout << "Fetcher::OnTimeout: " << name << endl;
+  // cout << "Last: " << m_lastPositiveActivity << ", config: " << m_maximumNoActivityPeriod
+  //      << ", now: " << date_time::second_clock<boost::posix_time::ptime>::universal_time()
+  //      << ", oldest: " << (date_time::second_clock<boost::posix_time::ptime>::universal_time() - m_maximumNoActivityPeriod) << endl;
+
+  if (m_lastPositiveActivity <
+      (date_time::second_clock<boost::posix_time::ptime>::universal_time() - m_maximumNoActivityPeriod))
+    {
+      m_activePipeline --;
+      if (m_activePipeline == 0)
+        {
+          m_onFetchFailed (*this);
+          // this is not valid anymore, but we still should be able finish work
+        }
+      return Closure::RESULT_OK;
+    }
+  else
+    return Closure::RESULT_REEXPRESS;
 }
diff --git a/src/fetcher.h b/src/fetcher.h
index c41c23c..271db52 100644
--- a/src/fetcher.h
+++ b/src/fetcher.h
@@ -27,22 +27,38 @@
 
 #include "scheduler.h"
 #include <boost/intrusive/list.hpp>
+#include <boost/date_time/posix_time/posix_time_types.hpp>
 
 class FetchManager;
 
 class Fetcher
 {
 public:
-  Fetcher (FetchManager &fetchManger,
+  typedef boost::function<void (Fetcher &, uint32_t /*requested seqno*/, const Ccnx::Name & /*requested base name*/,
+                                const Ccnx::Name & /*actual name*/, const Ccnx::Bytes &)> OnDataSegmentCallback;
+  typedef boost::function<void (Fetcher &)> OnFetchCompleteCallback;
+  typedef boost::function<void (Fetcher &)> OnFetchFailedCallback;
+
+  Fetcher (Ccnx::CcnxWrapperPtr ccnx,
+           OnDataSegmentCallback onDataSegment,
+           OnFetchCompleteCallback onFetchComplete, OnFetchFailedCallback onFetchFailed,
            const Ccnx::Name &name, int32_t minSeqNo, int32_t maxSeqNo,
+           boost::posix_time::time_duration timeout = boost::posix_time::seconds (30), // this time is not precise, but sets min bound
+                                                                                  // actual time depends on how fast Interests timeout
            const Ccnx::Name &forwardingHint = Ccnx::Name ());
   virtual ~Fetcher ();
 
-private:
+  inline bool
+  IsActive () const;
+
   void
   RestartPipeline ();
 
   void
+  SetForwardingHint (const Ccnx::Name &forwardingHint);
+
+private:
+  void
   FillPipeline ();
 
   void
@@ -51,13 +67,23 @@
   Ccnx::Closure::TimeoutCallbackReturnValue
   OnTimeout (uint32_t seqno, const Ccnx::Name &name);
 
+public:
+  boost::intrusive::list_member_hook<> m_managerListHook;
+
 private:
-  FetchManager &m_fetchManager;
+  Ccnx::CcnxWrapperPtr m_ccnx;
+
+  OnDataSegmentCallback m_onDataSegment;
+  OnFetchCompleteCallback m_onFetchComplete;
+  OnFetchFailedCallback m_onFetchFailed;
+
   bool m_active;
 
   Ccnx::Name m_name;
   Ccnx::Name m_forwardingHint;
 
+  boost::posix_time::time_duration m_maximumNoActivityPeriod;
+
   int32_t m_minSendSeqNo;
   int32_t m_maxInOrderRecvSeqNo;
   std::set<int32_t> m_outOfOrderRecvSeqNo;
@@ -68,8 +94,8 @@
   uint32_t m_pipeline;
   uint32_t m_activePipeline;
 
-  boost::intrusive::list_member_hook<> m_managerListHook;
-  friend class FetchManager;
+  boost::posix_time::ptime m_lastPositiveActivity;
+
 };
 
 typedef boost::error_info<struct tag_errmsg, std::string> errmsg_info_str;
@@ -80,5 +106,11 @@
 
 typedef boost::shared_ptr<Fetcher> FetcherPtr;
 
+bool
+Fetcher::IsActive () const
+{
+  return m_active;
+}
+
 
 #endif // FETCHER_H
diff --git a/test/test-fetch-manager.cc b/test/test-fetch-manager.cc
new file mode 100644
index 0000000..d28570e
--- /dev/null
+++ b/test/test-fetch-manager.cc
@@ -0,0 +1,162 @@
+/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
+/*
+ * Copyright (c) 2012 University of California, Los Angeles
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 as
+ * published by the Free Software Foundation;
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ *
+ * Author: Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ *	   Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ */
+
+#include "fetch-manager.h"
+#include "fetcher.h"
+#include "ccnx-wrapper.h"
+#include <boost/test/unit_test.hpp>
+#include <boost/make_shared.hpp>
+
+using namespace Ccnx;
+using namespace std;
+using namespace boost;
+
+BOOST_AUTO_TEST_SUITE(TestFetchManager)
+
+struct FetcherTestData
+{
+  set<uint32_t> recvData;
+  set<uint32_t> recvContent;
+
+  set<Name> differentNames;
+  set<Name> segmentNames;
+
+  bool m_done;
+  bool m_failed;
+
+  FetcherTestData ()
+    : m_done (false)
+    , m_failed (false)
+  {
+  }
+
+  void
+  onData (Fetcher &fetcher, uint32_t seqno, const Ccnx::Name &basename,
+          const Ccnx::Name &name, const Ccnx::Bytes &data)
+  {
+    recvData.insert (seqno);
+    differentNames.insert (basename);
+    segmentNames.insert (name);
+
+    if (data.size () == sizeof(int))
+      {
+        recvContent.insert (*reinterpret_cast<const int*> (head(data)));
+      }
+
+    // cout << basename << ", " << name << ", " << seqno << endl;
+  }
+
+  void
+  onComplete (Fetcher &fetcher)
+  {
+    m_done = true;
+    // cout << "Done" << endl;
+  }
+
+  void
+  onFail (Fetcher &fetcher)
+  {
+    m_failed = true;
+    // cout << "Failed" << endl;
+  }
+};
+
+
+BOOST_AUTO_TEST_CASE (TestFetcher)
+{
+  CcnxWrapperPtr ccnx = make_shared<CcnxWrapper> ();
+
+  Name baseName ("/base");
+  /* publish seqnos:  0, 1, 2, 3, 4, 5, 6, 7, 8, 9, <gap 5>, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, <gap 1>, 26 */
+  // this will allow us to test our pipeline of 6
+  for (int i = 0; i < 10; i++)
+    {
+      ccnx->publishData (Name (baseName)(i), reinterpret_cast<const unsigned char*> (&i), sizeof(int), 10);
+
+      int other = 10 + i+5;
+      ccnx->publishData (Name (baseName)(other), reinterpret_cast<const unsigned char*> (&other), sizeof(int), 10);
+    }
+
+  int oneMore = 26;
+  ccnx->publishData (Name (baseName)(oneMore), reinterpret_cast<const unsigned char*> (&oneMore), sizeof(int), 10);
+
+  FetcherTestData data;
+
+  Fetcher fetcher (ccnx,
+                   bind (&FetcherTestData::onData, &data, _1, _2, _3, _4, _5),
+                   bind (&FetcherTestData::onComplete, &data, _1),
+                   bind (&FetcherTestData::onFail, &data, _1),
+                   Name ("/base"), 0, 26,
+                   boost::posix_time::seconds (5)); // this time is not precise
+
+  BOOST_CHECK_EQUAL (fetcher.IsActive (), false);
+  fetcher.RestartPipeline ();
+  BOOST_CHECK_EQUAL (fetcher.IsActive (), true);
+
+  usleep(13000000);
+  BOOST_CHECK_EQUAL (data.m_failed, true);
+  BOOST_CHECK_EQUAL (data.differentNames.size (), 1);
+  BOOST_CHECK_EQUAL (data.segmentNames.size (), 10);
+  BOOST_CHECK_EQUAL (data.recvData.size (), 10);
+  BOOST_CHECK_EQUAL (data.recvContent.size (), 10);
+
+  ostringstream recvData;
+  for (set<uint32_t>::iterator i = data.recvData.begin (); i != data.recvData.end (); i++)
+    recvData << *i << ", ";
+
+  ostringstream recvContent;
+  for (set<uint32_t>::iterator i = data.recvContent.begin (); i != data.recvContent.end (); i++)
+    recvContent << *i << ", ";
+
+  BOOST_CHECK_EQUAL (recvData.str (), recvContent.str ());
+}
+
+// BOOST_AUTO_TEST_CASE (CcnxWrapperSelector)
+// {
+
+//   Closure closure (bind(dataCallback, _1, _2), bind(timeout, _1));
+
+//   Selectors selectors;
+//   selectors.interestLifetime(1);
+
+//   string n1 = "/random/01";
+//   c1->sendInterest(Name(n1), closure, selectors);
+//   sleep(2);
+//   c2->publishData(Name(n1), (const unsigned char *)n1.c_str(), n1.size(), 4);
+//   usleep(100000);
+//   BOOST_CHECK_EQUAL(g_timeout_counter, 1);
+//   BOOST_CHECK_EQUAL(g_dataCallback_counter, 0);
+
+//   string n2 = "/random/02";
+//   selectors.interestLifetime(2);
+//   c1->sendInterest(Name(n2), closure, selectors);
+//   sleep(1);
+//   c2->publishData(Name(n2), (const unsigned char *)n2.c_str(), n2.size(), 4);
+//   usleep(100000);
+//   BOOST_CHECK_EQUAL(g_timeout_counter, 1);
+//   BOOST_CHECK_EQUAL(g_dataCallback_counter, 1);
+
+//   // reset
+//   g_dataCallback_counter = 0;
+//   g_timeout_counter = 0;
+// }
+
+BOOST_AUTO_TEST_SUITE_END()