diff --git a/tools/chunks/catchunks/consumer.cpp b/tools/chunks/catchunks/consumer.cpp
index 8976f6a..fe30f75 100644
--- a/tools/chunks/catchunks/consumer.cpp
+++ b/tools/chunks/catchunks/consumer.cpp
@@ -26,7 +26,6 @@
  */
 
 #include "consumer.hpp"
-#include "discover-version.hpp"
 
 namespace ndn {
 namespace chunks {
@@ -34,35 +33,37 @@
 Consumer::Consumer(Face& face, Validator& validator, bool isVerbose, std::ostream& os)
   : m_face(face)
   , m_validator(validator)
-  , m_pipeline(nullptr)
-  , m_nextToPrint(0)
   , m_outputStream(os)
+  , m_nextToPrint(0)
   , m_isVerbose(isVerbose)
 {
 }
 
-void Consumer::run(DiscoverVersion& discover, PipelineInterests& pipeline)
+void
+Consumer::run(unique_ptr<DiscoverVersion> discover, unique_ptr<PipelineInterests> pipeline)
 {
-  m_pipeline = &pipeline;
+  m_discover = std::move(discover);
+  m_pipeline = std::move(pipeline);
   m_nextToPrint = 0;
+  m_bufferedData.clear();
 
-  discover.onDiscoverySuccess.connect(bind(&Consumer::runWithData, this, _1));
-  discover.onDiscoveryFailure.connect(bind(&Consumer::onFailure, this, _1));
+  discover->onDiscoverySuccess.connect(bind(&Consumer::startPipeline, this, _1));
+  discover->onDiscoveryFailure.connect(bind(&Consumer::onFailure, this, _1));
+  discover->run();
 
-  discover.run();
   m_face.processEvents();
 }
 
-void Consumer::runWithData(const Data& data)
+void
+Consumer::startPipeline(const Data& data)
 {
   m_validator.validate(data,
                        bind(&Consumer::onDataValidated, this, _1),
                        bind(&Consumer::onFailure, this, _2));
 
-  m_pipeline->runWithExcludedSegment(data,
-                                     bind(&Consumer::onData, this, _1, _2),
-                                     bind(&Consumer::onFailure, this, _1));
-
+  m_pipeline->run(data,
+                  bind(&Consumer::onData, this, _1, _2),
+                  bind(&Consumer::onFailure, this, _1));
 }
 
 void
@@ -100,7 +101,6 @@
   for (auto it = m_bufferedData.begin();
        it != m_bufferedData.end() && it->first == m_nextToPrint;
        it = m_bufferedData.erase(it), ++m_nextToPrint) {
-
     const Block& content = it->second->getContent();
     m_outputStream.write(reinterpret_cast<const char*>(content.value()), content.value_size());
   }
diff --git a/tools/chunks/catchunks/consumer.hpp b/tools/chunks/catchunks/consumer.hpp
index 9428a06..e2451be 100644
--- a/tools/chunks/catchunks/consumer.hpp
+++ b/tools/chunks/catchunks/consumer.hpp
@@ -25,12 +25,11 @@
  * @author Andrea Tosatto
  */
 
-
 #ifndef NDN_TOOLS_CHUNKS_CATCHUNKS_CONSUMER_HPP
 #define NDN_TOOLS_CHUNKS_CATCHUNKS_CONSUMER_HPP
 
-#include "pipeline-interests.hpp"
 #include "discover-version.hpp"
+#include "pipeline-interests.hpp"
 
 #include <ndn-cxx/security/validator.hpp>
 
@@ -66,11 +65,11 @@
    * @brief Run the consumer
    */
   void
-  run(DiscoverVersion& discover, PipelineInterests& pipeline);
+  run(unique_ptr<DiscoverVersion> discover, unique_ptr<PipelineInterests> pipeline);
 
 private:
   void
-  runWithData(const Data& data);
+  startPipeline(const Data& data);
 
   void
   onData(const Interest& interest, const Data& data);
@@ -88,9 +87,10 @@
 private:
   Face& m_face;
   Validator& m_validator;
-  PipelineInterests* m_pipeline;
-  uint64_t m_nextToPrint;
   std::ostream& m_outputStream;
+  unique_ptr<DiscoverVersion> m_discover;
+  unique_ptr<PipelineInterests> m_pipeline;
+  uint64_t m_nextToPrint;
   bool m_isVerbose;
 
 PUBLIC_WITH_TESTS_ELSE_PRIVATE:
diff --git a/tools/chunks/catchunks/ndncatchunks.cpp b/tools/chunks/catchunks/ndncatchunks.cpp
index d2b679f..98839b3 100644
--- a/tools/chunks/catchunks/ndncatchunks.cpp
+++ b/tools/chunks/catchunks/ndncatchunks.cpp
@@ -23,6 +23,8 @@
  * @author Wentao Shang
  * @author Steve DiBenedetto
  * @author Andrea Tosatto
+ * @author Davide Pesavento
+ * @author Weiwei Liu
  */
 
 #include "core/version.hpp"
@@ -30,6 +32,7 @@
 #include "consumer.hpp"
 #include "discover-version-fixed.hpp"
 #include "discover-version-iterative.hpp"
+#include "pipeline-interests-fixed-window.hpp"
 
 #include <ndn-cxx/security/validator-null.hpp>
 
@@ -42,6 +45,7 @@
   std::string programName(argv[0]);
   Options options;
   std::string discoverType("fixed");
+  std::string pipelineType("fixed");
   size_t maxPipelineSize(1);
   int maxRetriesAfterVersionFound(1);
   std::string uri;
@@ -148,15 +152,23 @@
       return 2;
     }
 
+    unique_ptr<PipelineInterests> pipeline;
+    if (pipelineType == "fixed") {
+      PipelineInterestsFixedWindow::Options optionsPipeline(options);
+      optionsPipeline.maxPipelineSize = maxPipelineSize;
+      pipeline = make_unique<PipelineInterestsFixedWindow>(face, optionsPipeline);
+    }
+    else {
+      std::cerr << "ERROR: Interest pipeline type not valid" << std::endl;
+      return 2;
+    }
+
     ValidatorNull validator;
     Consumer consumer(face, validator, options.isVerbose);
 
-    PipelineInterests::Options optionsPipeline(options);
-    optionsPipeline.maxPipelineSize = maxPipelineSize;
-    PipelineInterests pipeline(face, optionsPipeline);
-
     BOOST_ASSERT(discover != nullptr);
-    consumer.run(*discover, pipeline);
+    BOOST_ASSERT(pipeline != nullptr);
+    consumer.run(std::move(discover), std::move(pipeline));
   }
   catch (const Consumer::ApplicationNackError& e) {
     std::cerr << "ERROR: " << e.what() << std::endl;
diff --git a/tools/chunks/catchunks/pipeline-interests-fixed-window.cpp b/tools/chunks/catchunks/pipeline-interests-fixed-window.cpp
new file mode 100644
index 0000000..049d52d
--- /dev/null
+++ b/tools/chunks/catchunks/pipeline-interests-fixed-window.cpp
@@ -0,0 +1,183 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (c) 2016,  Regents of the University of California,
+ *                      Colorado State University,
+ *                      University Pierre & Marie Curie, Sorbonne University.
+ *
+ * This file is part of ndn-tools (Named Data Networking Essential Tools).
+ * See AUTHORS.md for complete list of ndn-tools authors and contributors.
+ *
+ * ndn-tools is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * ndn-tools 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
+ * ndn-tools, e.g., in COPYING.md file.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * See AUTHORS.md for complete list of ndn-cxx authors and contributors.
+ *
+ * @author Wentao Shang
+ * @author Steve DiBenedetto
+ * @author Andrea Tosatto
+ * @author Davide Pesavento
+ */
+
+#include "pipeline-interests-fixed-window.hpp"
+#include "data-fetcher.hpp"
+
+namespace ndn {
+namespace chunks {
+
+PipelineInterestsFixedWindow::PipelineInterestsFixedWindow(Face& face, const Options& options)
+  : PipelineInterests(face)
+  , m_options(options)
+  , m_nextSegmentNo(0)
+  , m_hasFailure(false)
+{
+  m_segmentFetchers.resize(m_options.maxPipelineSize);
+}
+
+PipelineInterestsFixedWindow::~PipelineInterestsFixedWindow()
+{
+  cancel();
+}
+
+void
+PipelineInterestsFixedWindow::doRun()
+{
+  // if the FinalBlockId is unknown, this could potentially request non-existent segments
+  for (size_t nRequestedSegments = 0;
+       nRequestedSegments < m_options.maxPipelineSize;
+       ++nRequestedSegments) {
+    if (!fetchNextSegment(nRequestedSegments))
+      // all segments have been requested
+      break;
+  }
+}
+
+bool
+PipelineInterestsFixedWindow::fetchNextSegment(std::size_t pipeNo)
+{
+  if (isStopping())
+    return false;
+
+  if (m_hasFailure) {
+    onFailure("Fetching terminated but no final segment number has been found");
+    return false;
+  }
+
+  if (m_nextSegmentNo == m_excludedSegmentNo)
+    m_nextSegmentNo++;
+
+  if (m_hasFinalBlockId && m_nextSegmentNo > m_lastSegmentNo)
+   return false;
+
+  // send interest for next segment
+  if (m_options.isVerbose)
+    std::cerr << "Requesting segment #" << m_nextSegmentNo << std::endl;
+
+  Interest interest(Name(m_prefix).appendSegment(m_nextSegmentNo));
+  interest.setInterestLifetime(m_options.interestLifetime);
+  interest.setMustBeFresh(m_options.mustBeFresh);
+  interest.setMaxSuffixComponents(1);
+
+  auto fetcher = DataFetcher::fetch(m_face, interest,
+                                    m_options.maxRetriesOnTimeoutOrNack,
+                                    m_options.maxRetriesOnTimeoutOrNack,
+                                    bind(&PipelineInterestsFixedWindow::handleData, this, _1, _2, pipeNo),
+                                    bind(&PipelineInterestsFixedWindow::handleFail, this, _2, pipeNo),
+                                    bind(&PipelineInterestsFixedWindow::handleFail, this, _2, pipeNo),
+                                    m_options.isVerbose);
+
+  BOOST_ASSERT(!m_segmentFetchers[pipeNo].first || !m_segmentFetchers[pipeNo].first->isRunning());
+  m_segmentFetchers[pipeNo] = make_pair(fetcher, m_nextSegmentNo);
+  m_nextSegmentNo++;
+
+  return true;
+}
+
+void
+PipelineInterestsFixedWindow::doCancel()
+{
+  for (auto& fetcher : m_segmentFetchers) {
+    if (fetcher.first)
+      fetcher.first->cancel();
+  }
+
+  m_segmentFetchers.clear();
+}
+
+void
+PipelineInterestsFixedWindow::handleData(const Interest& interest, const Data& data, size_t pipeNo)
+{
+  if (isStopping())
+    return;
+
+  BOOST_ASSERT(data.getName().equals(interest.getName()));
+
+  if (m_options.isVerbose)
+    std::cerr << "Received segment #" << data.getName()[-1].toSegment() << std::endl;
+
+  onData(interest, data);
+
+  if (!m_hasFinalBlockId && !data.getFinalBlockId().empty()) {
+    m_lastSegmentNo = data.getFinalBlockId().toSegment();
+    m_hasFinalBlockId = true;
+
+    for (auto& fetcher : m_segmentFetchers) {
+      if (fetcher.first == nullptr)
+        continue;
+
+      if (fetcher.second > m_lastSegmentNo) {
+        // stop trying to fetch segments that are beyond m_lastSegmentNo
+        fetcher.first->cancel();
+      }
+      else if (fetcher.first->hasError()) { // fetcher.second <= m_lastSegmentNo
+        // there was an error while fetching a segment that is part of the content
+        return onFailure("Failure retrieving segment #" + to_string(fetcher.second));
+      }
+    }
+  }
+
+  fetchNextSegment(pipeNo);
+}
+
+void PipelineInterestsFixedWindow::handleFail(const std::string& reason, std::size_t pipeNo)
+{
+  if (isStopping())
+    return;
+
+  // if the failed segment is definitely part of the content, raise a fatal error
+  if (m_hasFinalBlockId && m_segmentFetchers[pipeNo].second <= m_lastSegmentNo)
+    return onFailure(reason);
+
+  if (!m_hasFinalBlockId) {
+    bool areAllFetchersStopped = true;
+    for (auto& fetcher : m_segmentFetchers) {
+      if (fetcher.first == nullptr)
+        continue;
+
+      // cancel fetching all segments that follow
+      if (fetcher.second > m_segmentFetchers[pipeNo].second) {
+        fetcher.first->cancel();
+      }
+      else if (fetcher.first->isRunning()) { // fetcher.second <= m_segmentFetchers[pipeNo].second
+        areAllFetchersStopped = false;
+      }
+    }
+
+    if (areAllFetchersStopped) {
+      onFailure("Fetching terminated but no final segment number has been found");
+    }
+    else {
+      m_hasFailure = true;
+    }
+  }
+}
+
+} // namespace chunks
+} // namespace ndn
diff --git a/tools/chunks/catchunks/pipeline-interests-fixed-window.hpp b/tools/chunks/catchunks/pipeline-interests-fixed-window.hpp
new file mode 100644
index 0000000..2da40e6
--- /dev/null
+++ b/tools/chunks/catchunks/pipeline-interests-fixed-window.hpp
@@ -0,0 +1,118 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (c) 2016,  Regents of the University of California,
+ *                      Colorado State University,
+ *                      University Pierre & Marie Curie, Sorbonne University.
+ *
+ * This file is part of ndn-tools (Named Data Networking Essential Tools).
+ * See AUTHORS.md for complete list of ndn-tools authors and contributors.
+ *
+ * ndn-tools is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * ndn-tools 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
+ * ndn-tools, e.g., in COPYING.md file.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * See AUTHORS.md for complete list of ndn-cxx authors and contributors.
+ *
+ * @author Wentao Shang
+ * @author Steve DiBenedetto
+ * @author Andrea Tosatto
+ * @author Davide Pesavento
+ */
+
+#include "options.hpp"
+#include "pipeline-interests.hpp"
+
+namespace ndn {
+namespace chunks {
+
+class DataFetcher;
+
+class PipelineInterestsFixedWindowOptions : public Options
+{
+public:
+  explicit
+  PipelineInterestsFixedWindowOptions(const Options& options = Options())
+    : Options(options)
+    , maxPipelineSize(1)
+  {
+  }
+
+public:
+  size_t maxPipelineSize;
+};
+
+/**
+ * @brief Service for retrieving Data via an Interest pipeline
+ *
+ * Retrieves all segments of Data under a given prefix by maintaining a fixed-size window of
+ * N Interests in flight. A user-specified callback function is used to notify the arrival of
+ * each segment of Data.
+ *
+ * No guarantees are made as to the order in which segments are fetched or callbacks are invoked,
+ * i.e. out-of-order delivery is possible.
+ */
+class PipelineInterestsFixedWindow : public PipelineInterests
+{
+public:
+  typedef PipelineInterestsFixedWindowOptions Options;
+
+public:
+  /**
+   * @brief create a PipelineInterestsFixedWindow service
+   *
+   * Configures the pipelining service without specifying the retrieval namespace. After this
+   * configuration the method run must be called to start the Pipeline.
+   */
+  explicit
+  PipelineInterestsFixedWindow(Face& face, const Options& options = Options());
+
+  ~PipelineInterestsFixedWindow() final;
+
+private:
+  /**
+   * @brief fetch all the segments between 0 and m_lastSegmentNo
+   *
+   * Starts a fixed-window pipeline with size equal to m_options.maxPipelineSize. The pipeline
+   * will fetch every segment until the last segment is successfully received or an error occurs.
+   * The segment with segment number equal to m_excludedSegmentNo will not be fetched.
+   */
+  void
+  doRun() final;
+
+  void
+  doCancel() final;
+
+  /**
+   * @brief fetch the next segment that has not been requested yet
+   *
+   * @return false if there is an error or all the segments have been fetched, true otherwise
+   */
+  bool
+  fetchNextSegment(size_t pipeNo);
+
+  void
+  handleData(const Interest& interest, const Data& data, size_t pipeNo);
+
+  void
+  handleFail(const std::string& reason, size_t pipeNo);
+
+private:
+  const Options m_options;
+  std::vector<std::pair<shared_ptr<DataFetcher>, uint64_t>> m_segmentFetchers;
+  uint64_t m_nextSegmentNo;
+  /**
+   * true if one or more segment fetchers encountered an error; if m_hasFinalBlockId
+   * is false, this is usually not a fatal error for the pipeline
+   */
+  bool m_hasFailure;
+};
+
+} // namespace chunks
+} // namespace ndn
diff --git a/tools/chunks/catchunks/pipeline-interests.cpp b/tools/chunks/catchunks/pipeline-interests.cpp
index b07c959..ff817e4 100644
--- a/tools/chunks/catchunks/pipeline-interests.cpp
+++ b/tools/chunks/catchunks/pipeline-interests.cpp
@@ -23,179 +23,63 @@
  * @author Wentao Shang
  * @author Steve DiBenedetto
  * @author Andrea Tosatto
+ * @author Davide Pesavento
+ * @author Weiwei Liu
  */
 
 #include "pipeline-interests.hpp"
-#include "data-fetcher.hpp"
 
 namespace ndn {
 namespace chunks {
 
-PipelineInterests::PipelineInterests(Face& face, const Options& options)
+PipelineInterests::PipelineInterests(Face& face)
   : m_face(face)
-  , m_nextSegmentNo(0)
   , m_lastSegmentNo(0)
-  , m_excludeSegmentNo(0)
-  , m_options(options)
+  , m_excludedSegmentNo(0)
   , m_hasFinalBlockId(false)
-  , m_hasError(false)
-  , m_hasFailure(false)
+  , m_isStopping(false)
 {
-  m_segmentFetchers.resize(m_options.maxPipelineSize);
 }
 
-PipelineInterests::~PipelineInterests()
-{
-  cancel();
-}
+PipelineInterests::~PipelineInterests() = default;
 
 void
-PipelineInterests::runWithExcludedSegment(const Data& data, DataCallback onData,
-                                          FailureCallback onFailure)
+PipelineInterests::run(const Data& data, DataCallback onData, FailureCallback onFailure)
 {
   BOOST_ASSERT(onData != nullptr);
   m_onData = std::move(onData);
   m_onFailure = std::move(onFailure);
-
-  Name dataName = data.getName();
-  m_prefix = dataName.getPrefix(-1);
-  m_excludeSegmentNo = dataName[-1].toSegment();
+  m_prefix = data.getName().getPrefix(-1);
+  m_excludedSegmentNo = data.getName()[-1].toSegment();
 
   if (!data.getFinalBlockId().empty()) {
-    m_hasFinalBlockId = true;
     m_lastSegmentNo = data.getFinalBlockId().toSegment();
+    m_hasFinalBlockId = true;
   }
 
-  // if the FinalBlockId is unknown, this could potentially request non-existent segments
-  for (size_t nRequestedSegments = 0; nRequestedSegments < m_options.maxPipelineSize;
-       nRequestedSegments++) {
-    if (!fetchNextSegment(nRequestedSegments)) // all segments have been requested
-      break;
-  }
-}
-
-bool
-PipelineInterests::fetchNextSegment(std::size_t pipeNo)
-{
-  if (m_hasFailure) {
-    fail("Fetching terminated but no final segment number has been found");
-    return false;
-  }
-
-  if (m_nextSegmentNo == m_excludeSegmentNo)
-    m_nextSegmentNo++;
-
-  if (m_hasFinalBlockId && m_nextSegmentNo > m_lastSegmentNo)
-   return false;
-
-  // Send interest for next segment
-  if (m_options.isVerbose)
-    std::cerr << "Requesting segment #" << m_nextSegmentNo << std::endl;
-
-  Interest interest(Name(m_prefix).appendSegment(m_nextSegmentNo));
-  interest.setInterestLifetime(m_options.interestLifetime);
-  interest.setMustBeFresh(m_options.mustBeFresh);
-  interest.setMaxSuffixComponents(1);
-
-  BOOST_ASSERT(!m_segmentFetchers[pipeNo].first || !m_segmentFetchers[pipeNo].first->isRunning());
-
-  auto fetcher = DataFetcher::fetch(m_face, interest,
-                                    m_options.maxRetriesOnTimeoutOrNack,
-                                    m_options.maxRetriesOnTimeoutOrNack,
-                                    bind(&PipelineInterests::handleData, this, _1, _2, pipeNo),
-                                    bind(&PipelineInterests::handleFail, this, _2, pipeNo),
-                                    bind(&PipelineInterests::handleFail, this, _2, pipeNo),
-                                    m_options.isVerbose);
-
-  m_segmentFetchers[pipeNo] = make_pair(fetcher, m_nextSegmentNo);
-
-  m_nextSegmentNo++;
-  return true;
+  doRun();
 }
 
 void
 PipelineInterests::cancel()
 {
-  for (auto& fetcher : m_segmentFetchers)
-    if (fetcher.first)
-      fetcher.first->cancel();
+  if (m_isStopping)
+    return;
 
-  m_segmentFetchers.clear();
+  m_isStopping = true;
+  doCancel();
 }
 
 void
-PipelineInterests::fail(const std::string& reason)
+PipelineInterests::onFailure(const std::string& reason)
 {
-  if (!m_hasError) {
-    cancel();
-    m_hasError = true;
-    m_hasFailure = true;
-    if (m_onFailure)
-      m_face.getIoService().post([this, reason] { m_onFailure(reason); });
-  }
-}
-
-void
-PipelineInterests::handleData(const Interest& interest, const Data& data, size_t pipeNo)
-{
-  if (m_hasError)
+  if (m_isStopping)
     return;
 
-  BOOST_ASSERT(data.getName().equals(interest.getName()));
+  cancel();
 
-  if (m_options.isVerbose)
-    std::cerr << "Received segment #" << data.getName()[-1].toSegment() << std::endl;
-
-  m_onData(interest, data);
-
-  if (!m_hasFinalBlockId && !data.getFinalBlockId().empty()) {
-    m_lastSegmentNo = data.getFinalBlockId().toSegment();
-    m_hasFinalBlockId = true;
-
-    for (auto& fetcher : m_segmentFetchers) {
-      if (fetcher.first && fetcher.second > m_lastSegmentNo) {
-        // Stop trying to fetch segments that are not part of the content
-        fetcher.first->cancel();
-      }
-      else if (fetcher.first && fetcher.first->hasError()) { // fetcher.second <= m_lastSegmentNo
-        // there was an error while fetching a segment that is part of the content
-        fail("Failure retriving segment #" + to_string(fetcher.second));
-        return;
-      }
-    }
-  }
-
-  fetchNextSegment(pipeNo);
-}
-
-void PipelineInterests::handleFail(const std::string& reason, std::size_t pipeNo)
-{
-  if (m_hasError)
-    return;
-
-  if (m_hasFinalBlockId && m_segmentFetchers[pipeNo].second <= m_lastSegmentNo) {
-    fail(reason);
-  }
-  else if (!m_hasFinalBlockId) {
-    // don't fetch the following segments
-    bool areAllFetchersStopped = true;
-    for (auto& fetcher : m_segmentFetchers) {
-      if (fetcher.first && fetcher.second > m_segmentFetchers[pipeNo].second) {
-        fetcher.first->cancel();
-      }
-      else if (fetcher.first && fetcher.first->isRunning()) {
-        // fetcher.second <= m_segmentFetchers[pipeNo].second
-        areAllFetchersStopped = false;
-      }
-    }
-    if (areAllFetchersStopped) {
-      if (m_onFailure)
-        fail("Fetching terminated but no final segment number has been found");
-    }
-    else {
-      m_hasFailure = true;
-    }
-  }
+  if (m_onFailure)
+    m_face.getIoService().post([this, reason] { m_onFailure(reason); });
 }
 
 } // namespace chunks
diff --git a/tools/chunks/catchunks/pipeline-interests.hpp b/tools/chunks/catchunks/pipeline-interests.hpp
index 32d5b79..63d714d 100644
--- a/tools/chunks/catchunks/pipeline-interests.hpp
+++ b/tools/chunks/catchunks/pipeline-interests.hpp
@@ -23,46 +23,31 @@
  * @author Wentao Shang
  * @author Steve DiBenedetto
  * @author Andrea Tosatto
+ * @author Davide Pesavento
+ * @author Weiwei Liu
  */
 
 #ifndef NDN_TOOLS_CHUNKS_CATCHUNKS_PIPELINE_INTERESTS_HPP
 #define NDN_TOOLS_CHUNKS_CATCHUNKS_PIPELINE_INTERESTS_HPP
 
 #include "core/common.hpp"
-#include "options.hpp"
 
 namespace ndn {
 namespace chunks {
 
-class DataFetcher;
-
-class PipelineInterestsOptions : public Options
-{
-public:
-  explicit
-  PipelineInterestsOptions(const Options& options = Options())
-    : Options(options)
-    , maxPipelineSize(1)
-  {
-  }
-
-public:
-  size_t maxPipelineSize;
-};
-
 /**
  * @brief Service for retrieving Data via an Interest pipeline
  *
- * Retrieves all segmented Data under the specified prefix by maintaining a pipeline of N Interests
- * in flight.
+ * Retrieves all segments of Data under a given prefix by maintaining a (variable or fixed-size)
+ * window of N Interests in flight. A user-specified callback function is used to notify
+ * the arrival of each segment of Data.
  *
- * Provides retrieved Data on arrival with no ordering guarantees. Data is delivered to the
- * PipelineInterests' user via callback immediately upon arrival.
+ * No guarantees are made as to the order in which segments are fetched or callbacks are invoked,
+ * i.e. out-of-order delivery is possible.
  */
 class PipelineInterests
 {
 public:
-  typedef PipelineInterestsOptions Options;
   typedef function<void(const std::string& reason)> FailureCallback;
 
 public:
@@ -70,26 +55,23 @@
    * @brief create a PipelineInterests service
    *
    * Configures the pipelining service without specifying the retrieval namespace. After this
-   * configuration the method runWithExcludedSegment must be called to run the Pipeline.
+   * configuration the method run must be called to start the Pipeline.
    */
   explicit
-  PipelineInterests(Face& face, const Options& options = Options());
+  PipelineInterests(Face& face);
 
+  virtual
   ~PipelineInterests();
 
   /**
-   * @brief fetch all the segments between 0 and lastSegment of the specified prefix
+   * @brief start fetching all the segments of the specified prefix
    *
-   * Starts the pipeline of size defined inside the options. The pipeline retrieves all the segments
-   * until the last segment is received, @p data is excluded from the retrieving.
-   *
-   * @param data a segment of the segmented Data to retrive; data.getName() must end with a segment
-   *        number
+   * @param data a segment of the segmented Data to fetch; the Data name must end with a segment number
    * @param onData callback for every segment correctly received, must not be empty
-   * @param onfailure callback called if an error occurs
+   * @param onFailure callback if an error occurs, may be empty
    */
   void
-  runWithExcludedSegment(const Data& data, DataCallback onData, FailureCallback onFailure);
+  run(const Data& data, DataCallback onData, FailureCallback onFailure);
 
   /**
    * @brief stop all fetch operations
@@ -97,44 +79,52 @@
   void
   cancel();
 
+protected:
+  bool
+  isStopping() const
+  {
+    return m_isStopping;
+  }
+
+  void
+  onData(const Interest& interest, const Data& data) const
+  {
+    m_onData(interest, data);
+  }
+
+  /**
+   * @brief subclasses can call this method to signal an unrecoverable failure
+   */
+  void
+  onFailure(const std::string& reason);
+
 private:
   /**
-   * @brief fetch the next segment that has not been requested yet
+   * @brief perform subclass-specific operations to fetch all the segments
    *
-   * @return false if there is an error or all the segments have been fetched, true otherwise
+   * When overriding this function, at a minimum, the subclass should implement the retrieving
+   * of all the segments. Segment m_excludedSegmentNo can be skipped. Subclass must guarantee
+   * that onData is called at least once for every segment that is fetched successfully.
+   *
+   * @note m_lastSegmentNo contains a valid value only if m_hasFinalBlockId is true.
    */
-  bool
-  fetchNextSegment(size_t pipeNo);
+  virtual void
+  doRun() = 0;
 
-  void
-  fail(const std::string& reason);
+  virtual void
+  doCancel() = 0;
 
-  void
-  handleData(const Interest& interest, const Data& data, size_t pipeNo);
-
-  void
-  handleFail(const std::string& reason, size_t pipeNo);
+protected:
+  Face& m_face;
+  Name m_prefix;
+  uint64_t m_lastSegmentNo;
+  uint64_t m_excludedSegmentNo;
+  bool m_hasFinalBlockId; ///< true if the last segment number is known
 
 private:
-  Name m_prefix;
-  Face& m_face;
-  uint64_t m_nextSegmentNo;
-  uint64_t m_lastSegmentNo;
-  uint64_t m_excludeSegmentNo;
   DataCallback m_onData;
   FailureCallback m_onFailure;
-  const Options m_options;
-  std::vector<std::pair<shared_ptr<DataFetcher>, uint64_t>> m_segmentFetchers;
-  bool m_hasFinalBlockId;
-  /**
-   * true if there's a critical error
-   */
-  bool m_hasError;
-  /**
-   * true if one or more segmentFetcher failed, if lastSegmentNo is not set this is usually not a
-   * fatal error for the pipeline
-   */
-  bool m_hasFailure;
+  bool m_isStopping;
 };
 
 } // namespace chunks