Use ByteStream in downloads system to decouple source and sink.

net/url_request/url_request_test_job.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "net/url_request/url_request_test_job.h"
 
 #include <algorithm>
 #include <list>
 
 #include "base/bind.h"
@@ skipping 229 matching lines @@
       // Must call AdvanceJob() prior to NotifyReadComplete() since that may
       // delete |this|.
       AdvanceJob();
       stage_ = DATA_AVAILABLE;
       // OK if ReadRawData wasn't called yet.
       if (async_buf_) {
         int bytes_read;
         if (!ReadRawData(async_buf_, async_buf_size_, &bytes_read))
           NOTREACHED() << "This should not return false in DATA_AVAILABLE.";
         SetStatus(URLRequestStatus());  // clear the io pending flag
+        if (NextReadAsync()) {
+          // Make all future reads return io pending until the next
+          // ProcessNextOperation().
+          stage_ = WAITING;
+        }
         NotifyReadComplete(bytes_read);
       }
       break;
     case DATA_AVAILABLE:
       AdvanceJob();
       stage_ = ALL_DATA;  // done sending data
       break;
     case ALL_DATA:
       stage_ = DONE;
       return;
     case DONE:
       return;
     default:
       NOTREACHED() << "Invalid stage";
       return;
   }
 }
 
+bool URLRequestTestJob::NextReadAsync() {
+  return false;
+}
+
 void URLRequestTestJob::AdvanceJob() {
   if (auto_advance_) {
     MessageLoop::current()->PostTask(
         FROM_HERE, base::Bind(&URLRequestTestJob::ProcessNextOperation,
                               weak_factory_.GetWeakPtr()));
     return;
   }
   g_pending_jobs.Get().push_back(this);
 }
 
 // static
 bool URLRequestTestJob::ProcessOnePendingMessage() {
   if (g_pending_jobs.Get().empty())
     return false;
 
   URLRequestTestJob* next_job(g_pending_jobs.Get().front());
   g_pending_jobs.Get().pop_front();
 
   DCHECK(!next_job->auto_advance());  // auto_advance jobs should be in this q
   next_job->ProcessNextOperation();
   return true;
 }
 
 }  // namespace net