Creation of ByteStream class.

content/browser/download/byte_stream_unittest.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 "content/browser/download/byte_stream.h"
+
+#include "base/bind.h"
+#include "base/callback.h"
+#include "base/location.h"
+#include "base/memory/ref_counted.h"
+#include "base/message_loop.h"
+#include "base/task_runner.h"
+#include "net/base/io_buffer.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+using ::testing::_;
+using ::testing::Return;
+using ::testing::SaveArg;
+using ::testing::StrictMock;
+
+namespace tracked_objects {
+class Location;
+}
+
+namespace {
+
+class MockTaskRunner : public base::SequencedTaskRunner {
+ public:
+  MockTaskRunner();
+
+  // TaskRunner functions.
+  MOCK_METHOD3(PostDelayedTask, bool(const tracked_objects::Location&,
+                                     const base::Closure&, int64));
+  MOCK_METHOD3(PostDelayedTask, bool(const tracked_objects::Location&,
+                                     const base::Closure&, base::TimeDelta));
+
+  MOCK_METHOD3(PostNonNestableDelayedTask, bool(
+      const tracked_objects::Location&,
+      const base::Closure&,
+      int64 delay_ms));
+
+  MOCK_METHOD3(PostNonNestableDelayedTask, bool(
+      const tracked_objects::Location&,
+      const base::Closure&,
+      base::TimeDelta));
+
+  MOCK_CONST_METHOD0(RunsTasksOnCurrentThread, bool());
+
+ protected:
+  ~MockTaskRunner();
+};
+
+MockTaskRunner::MockTaskRunner() { }
+
+MockTaskRunner::~MockTaskRunner() { }
+
+static int null_callback_call_count = 0;
+static int alt_callback_call_count = 0;
+
+void NullCallback() {

[benjhayden, 2012/05/16 15:55:57]

Would something like this work just as well with less code and insulate between
TEST_Fs?

void CountCallbacks(int* counter) { ++*counter; }

TEST_F(...) {
  int counter_a = 0;
  int counter_b = 0;
  int counter_c = 0;
  base::Bind(&CountCallbacks, &counter_a);
  base::Bind(&CountCallbacks, &counter_b);
  base::Bind(&CountCallbacks, &counter_c);
  ...
  EXPECT_EQ(42, counter_a);
  EXPECT_EQ(42, counter_b);
  EXPECT_EQ(42, counter_c);
  counter_a = counter_b = counter_c = 0;
  ...
  EXPECT_EQ(42, counter_a);
  EXPECT_EQ(42, counter_b);
  EXPECT_EQ(42, counter_c);
}

[Randy Smith (Not in Mondays), 2012/05/16 20:30:16]

D'oh.  Good idea.  Done.

+  null_callback_call_count++;
+}
+
+void AltCallback() {
+  alt_callback_call_count++;
+}
+
+}  // namespace
+
+class ByteStreamTest : public testing::Test {
+ public:
+  // Create a new IO buffer of the given |buffer_size|, with contents
+  // dependent on the |seed_key|.  The |seed_key| is also used for comparing
+  // pointers between NewIOBuffer and ValidateIOBuffer; do not re-use any
+  // |seed_key| value within a single test.

[benjhayden, 2012/05/16 15:55:57]

Can you use an internal automatic counter instead of magic numbers and fallible
human counters?

[Randy Smith (Not in Mondays), 2012/05/16 20:30:16]

I'm a little reluctant to do that, as it implies that we're always going to
evaluate buffers in the order that they were allocated.  Having said that, that
assumption is valid for this entire file, and it would make for simpler code, so
ok.  

+  scoped_refptr<net::IOBuffer> NewIOBuffer(size_t buffer_size, int seed_key) {
+    scoped_refptr<net::IOBuffer> buffer(new net::IOBuffer(buffer_size));
+    char *bufferp = buffer->data();
+    for (size_t i = 0; i < buffer_size; i++)
+      bufferp[i] = (i + seed_key) % (1 << sizeof(char));
+    DCHECK(pointer_map_.find(seed_key) == pointer_map_.end());
+    DCHECK(length_map_.find(seed_key) == length_map_.end());
+    pointer_map_[seed_key] = bufferp;
+    length_map_[seed_key] = buffer_size;
+    return buffer;
+  }
+
+  // Create an IOBuffer of the appropriate size and add it to the
+  // ByteStream, returning the result of the ByteStream::Write.
+  // Separate function to avoid duplication of buffer_size in test
+  // calls.
+  bool Write(content::ByteStreamInput* byte_stream_input,
+             int seed_key, size_t buffer_size) {
+    return byte_stream_input->Write(NewIOBuffer(buffer_size, seed_key),
+                                    buffer_size);
+  }
+
+  // Validate that we have the IOBuffer we expect.  This IOBuffer must
+  // have been created through NewIOBuffer with the given |buffer_size|
+  // and |seed_key|.
+  bool ValidateIOBuffer(scoped_refptr<net::IOBuffer> buffer, int seed_key,
+                        size_t buffer_size) {
+    char *bufferp = buffer->data();
+    EXPECT_TRUE(pointer_map_.find(seed_key) != pointer_map_.end());
+    if (pointer_map_.find(seed_key) == pointer_map_.end())
+      return false;
+    EXPECT_EQ(bufferp, pointer_map_[seed_key]);
+    EXPECT_TRUE(length_map_.find(seed_key) != length_map_.end());
+    if (length_map_.find(seed_key) == length_map_.end())
+      return false;
+    EXPECT_EQ(buffer_size, length_map_[seed_key]);
+    for (size_t i = 0; i < buffer_size; i++) {
+      EXPECT_EQ(static_cast<int>((i + seed_key) % (1 << sizeof(char))),
+                bufferp[i]);
+      if (static_cast<int>((i + seed_key) % (1 << sizeof(char))) != bufferp[i])
+        return false;
+    }
+    return true;
+  }
+
+ protected:
+  MessageLoop message_loop_;
+ private:

[benjhayden, 2012/05/16 19:17:21]

Blank line between sections?

[Randy Smith (Not in Mondays), 2012/05/16 20:30:16]

Done.

+  std::map<int, char*> pointer_map_;
+  std::map<int, size_t> length_map_;
+};
+
+// Confirm that filling and emptying the pipe works properly, and that
+// we get full triggers when we expect.
+TEST_F(ByteStreamTest, PushBack) {

[benjhayden, 2012/05/16 19:17:21]

Would you mind prepending "ByteStream_" to all of the TEST_F case names so that
we see "ByteStream_PushBack" etc. instead of mysteriously "PushBack" on the bot
summary pages?

[Randy Smith (Not in Mondays), 2012/05/16 20:30:16]

It's annoyingly redundant :-{, but I hear your pain.  Done.

+  scoped_ptr<content::ByteStreamInput> byte_stream_input;
+  scoped_ptr<content::ByteStreamOutput> byte_stream_output;
+  content::CreateByteStream(
+      &byte_stream_input, &byte_stream_output,
+      message_loop_.message_loop_proxy(), message_loop_.message_loop_proxy(),
+      3 * 1024);
+
+  // Push a series of IO buffers on; test pushback happening and
+  // that it's advisory.
+  EXPECT_TRUE(Write(byte_stream_input.get(), 0, 1024));
+  EXPECT_TRUE(Write(byte_stream_input.get(), 1, 1024));
+  EXPECT_TRUE(Write(byte_stream_input.get(), 2, 1024));
+  EXPECT_FALSE(Write(byte_stream_input.get(), 3, 1));
+  EXPECT_FALSE(Write(byte_stream_input.get(), 4, 1024));
+  // Flush
+  byte_stream_input->Close(content::DOWNLOAD_INTERRUPT_REASON_NONE);
+  message_loop_.RunAllPending();
+
+  // Pull the IO buffers out; do we get the same buffers and do they
+  // have the same contents?
+  scoped_refptr<net::IOBuffer> output_io_buffer;
+  size_t output_length;
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 0, output_length));
+
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 1, output_length));
+
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 2, output_length));
+
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 3, output_length));
+
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 4, output_length));
+
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_COMPLETE,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+}
+
+// Same as above, only use knowledge of the internals to confirm
+// that we're getting pushback even when data's split across the two
+// objects
+TEST_F(ByteStreamTest, PushBackSplit) {
+  scoped_ptr<content::ByteStreamInput> byte_stream_input;
+  scoped_ptr<content::ByteStreamOutput> byte_stream_output;
+  content::CreateByteStream(
+      &byte_stream_input, &byte_stream_output,
+      message_loop_.message_loop_proxy(), message_loop_.message_loop_proxy(),
+      9 * 1024);
+
+  // Push a series of IO buffers on; test pushback happening and
+  // that it's advisory.
+  EXPECT_TRUE(Write(byte_stream_input.get(), 0, 1024));
+  message_loop_.RunAllPending();
+  EXPECT_TRUE(Write(byte_stream_input.get(), 1, 1024));
+  message_loop_.RunAllPending();
+  EXPECT_TRUE(Write(byte_stream_input.get(), 2, 1024));
+  message_loop_.RunAllPending();
+  EXPECT_TRUE(Write(byte_stream_input.get(), 3, 1024));
+  message_loop_.RunAllPending();
+  EXPECT_FALSE(Write(byte_stream_input.get(), 4, 6 * 1024));
+  message_loop_.RunAllPending();
+
+  // Pull the IO buffers out; do we get the same buffers and do they
+  // have the same contents?
+  scoped_refptr<net::IOBuffer> output_io_buffer;
+  size_t output_length;
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 0, output_length));
+
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 1, output_length));
+
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 2, output_length));
+
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 3, output_length));
+
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 4, output_length));
+
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_EMPTY,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+}
+
+// Confirm that a Close() notification transmits in-order
+// with data on the pipe.
+TEST_F(ByteStreamTest, CompleteTransmits) {
+  scoped_ptr<content::ByteStreamInput> byte_stream_input;
+  scoped_ptr<content::ByteStreamOutput> byte_stream_output;
+
+  scoped_refptr<net::IOBuffer> output_io_buffer;
+  size_t output_length;
+
+  // Empty stream, non-error case.
+  content::CreateByteStream(
+      &byte_stream_input, &byte_stream_output,
+      message_loop_.message_loop_proxy(), message_loop_.message_loop_proxy(),
+      3 * 1024);
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_EMPTY,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  byte_stream_input->Close(content::DOWNLOAD_INTERRUPT_REASON_NONE);
+  message_loop_.RunAllPending();
+  ASSERT_EQ(content::ByteStreamOutput::STREAM_COMPLETE,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_EQ(content::DOWNLOAD_INTERRUPT_REASON_NONE,
+            byte_stream_output->GetStatus());
+
+  // Non-empty stream, non-error case.
+  content::CreateByteStream(
+      &byte_stream_input, &byte_stream_output,
+      message_loop_.message_loop_proxy(), message_loop_.message_loop_proxy(),
+      3 * 1024);
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_EMPTY,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(Write(byte_stream_input.get(), 0, 1024));
+  byte_stream_input->Close(content::DOWNLOAD_INTERRUPT_REASON_NONE);
+  message_loop_.RunAllPending();
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 0, output_length));
+  ASSERT_EQ(content::ByteStreamOutput::STREAM_COMPLETE,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_EQ(content::DOWNLOAD_INTERRUPT_REASON_NONE,
+            byte_stream_output->GetStatus());
+
+  // Empty stream, non-error case.
+  content::CreateByteStream(
+      &byte_stream_input, &byte_stream_output,
+      message_loop_.message_loop_proxy(), message_loop_.message_loop_proxy(),
+      3 * 1024);
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_EMPTY,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  byte_stream_input->Close(
+      content::DOWNLOAD_INTERRUPT_REASON_NETWORK_DISCONNECTED);
+  message_loop_.RunAllPending();
+  ASSERT_EQ(content::ByteStreamOutput::STREAM_COMPLETE,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_EQ(content::DOWNLOAD_INTERRUPT_REASON_NETWORK_DISCONNECTED,
+            byte_stream_output->GetStatus());
+
+  // Non-empty stream, non-error case.
+  content::CreateByteStream(
+      &byte_stream_input, &byte_stream_output,
+      message_loop_.message_loop_proxy(), message_loop_.message_loop_proxy(),
+      3 * 1024);
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_EMPTY,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(Write(byte_stream_input.get(), 1, 1024));
+  byte_stream_input->Close(
+      content::DOWNLOAD_INTERRUPT_REASON_NETWORK_DISCONNECTED);
+  message_loop_.RunAllPending();
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 1, output_length));
+  ASSERT_EQ(content::ByteStreamOutput::STREAM_COMPLETE,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_EQ(content::DOWNLOAD_INTERRUPT_REASON_NETWORK_DISCONNECTED,
+            byte_stream_output->GetStatus());
+}
+
+// Confirm that callbacks on the sink side are triggered when they should be.
+TEST_F(ByteStreamTest, SinkCallback) {
+  scoped_refptr<MockTaskRunner> task_runner(new StrictMock<MockTaskRunner>());
+  int null_callback_call_count_start = 0;
+
+  scoped_ptr<content::ByteStreamInput> byte_stream_input;
+  scoped_ptr<content::ByteStreamOutput> byte_stream_output;
+  content::CreateByteStream(
+      &byte_stream_input, &byte_stream_output,
+      message_loop_.message_loop_proxy(), task_runner,
+      10000);
+
+  scoped_refptr<net::IOBuffer> output_io_buffer;
+  size_t output_length;
+  base::Closure intermediate_callback;
+
+  // Note that the specifics of when the callbacks are called with regard
+  // to how much data is pushed onto the pipe is not (currently) part
+  // of the interface contract.  If it becomes part of the contract, the
+  // tests below should get much more precise.
+
+  // Confirm callback called when you add more than 33% of the buffer.
+
+  // Setup callback
+  byte_stream_output->RegisterCallback(base::Bind(NullCallback));
+  null_callback_call_count_start = null_callback_call_count;
+  EXPECT_CALL(*task_runner.get(), PostDelayedTask(_, _, 0))
+      .WillOnce(DoAll(SaveArg<1>(&intermediate_callback),
+                      Return(true)));
+
+  EXPECT_TRUE(Write(byte_stream_input.get(), 0, 4000));
+  message_loop_.RunAllPending();
+
+  // Check callback results match expectations.
+  ::testing::Mock::VerifyAndClearExpectations(task_runner.get());
+  EXPECT_EQ(null_callback_call_count_start, null_callback_call_count);
+  intermediate_callback.Run();
+  EXPECT_EQ(null_callback_call_count_start+1, null_callback_call_count);
+
+  // Check data and stream state.
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 0, output_length));
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_EMPTY,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+
+  // Confirm callback *isn't* called at less than 33%
+  null_callback_call_count_start = null_callback_call_count;
+  EXPECT_TRUE(Write(byte_stream_input.get(), 1, 3000));
+  message_loop_.RunAllPending();
+
+  // This reflects an implementation artifact that data goes with callbacks,
+  // which should not be considered part of the interface guarantee.
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_EMPTY,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+}
+
+// Confirm that callbacks on the source side are triggered when they should
+// be.
+TEST_F(ByteStreamTest, SourceCallback) {
+  scoped_refptr<MockTaskRunner> task_runner(new StrictMock<MockTaskRunner>());
+  int null_callback_call_count_start = 0;
+
+  scoped_ptr<content::ByteStreamInput> byte_stream_input;
+  scoped_ptr<content::ByteStreamOutput> byte_stream_output;
+  content::CreateByteStream(
+      &byte_stream_input, &byte_stream_output,
+      task_runner, message_loop_.message_loop_proxy(),
+      10000);
+
+  scoped_refptr<net::IOBuffer> output_io_buffer;
+  size_t output_length;
+  base::Closure intermediate_callback;
+
+  // Note that the specifics of when the callbacks are called with regard
+  // to how much data is pulled from the pipe is not (currently) part
+  // of the interface contract.  If it becomes part of the contract, the
+  // tests below should get much more precise.
+
+  // Confirm callback called when about 33% space available, and not
+  // at other transitions.
+
+  // Setup expectations and add data.
+  byte_stream_input->RegisterCallback(base::Bind(NullCallback));
+  EXPECT_TRUE(Write(byte_stream_input.get(), 0, 2000));
+  EXPECT_TRUE(Write(byte_stream_input.get(), 1, 2001));
+  EXPECT_FALSE(Write(byte_stream_input.get(), 2, 6000));
+  null_callback_call_count_start = null_callback_call_count;
+
+  // Allow bytes to transition (needed for message passing implementation),
+  // and get and validate the data.
+  message_loop_.RunAllPending();
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 0, output_length));
+
+  // Setup expectations.
+  EXPECT_CALL(*task_runner.get(), PostDelayedTask(_, _, 0))
+      .WillOnce(DoAll(SaveArg<1>(&intermediate_callback),
+                      Return(true)));
+
+  // Grab data, triggering callback.  Recorded on dispatch, but doesn't
+  // happen because it's caught by the mock.
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  ::testing::Mock::VerifyAndClearExpectations(task_runner.get());
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 1, output_length));
+
+  // Confirm that the callback passed to the mock does what we expect.
+  EXPECT_EQ(null_callback_call_count_start, null_callback_call_count);
+  intermediate_callback.Run();
+  EXPECT_EQ(null_callback_call_count_start+1, null_callback_call_count);
+
+  // Same drill with final buffer.
+  EXPECT_CALL(*task_runner.get(), PostDelayedTask(_, _, 0))
+      .WillOnce(DoAll(SaveArg<1>(&intermediate_callback),
+                      Return(true)));
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  ::testing::Mock::VerifyAndClearExpectations(task_runner.get());
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 2, output_length));
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_EMPTY,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_EQ(null_callback_call_count_start+1, null_callback_call_count);
+  intermediate_callback.Run();
+  // Should have updated the internal structures but not called the
+  // callback.
+  EXPECT_EQ(null_callback_call_count_start+1, null_callback_call_count);
+}
+
+// Confirm that racing a change to a sink callback with a post results
+// in the new callback being called.
+TEST_F(ByteStreamTest, SinkInterrupt) {
+  scoped_refptr<MockTaskRunner> task_runner(new StrictMock<MockTaskRunner>());
+  int null_callback_call_count_start = 0;
+  int alt_callback_call_count_start = 0;
+
+  scoped_ptr<content::ByteStreamInput> byte_stream_input;
+  scoped_ptr<content::ByteStreamOutput> byte_stream_output;
+  content::CreateByteStream(
+      &byte_stream_input, &byte_stream_output,
+      message_loop_.message_loop_proxy(), task_runner,
+      10000);
+
+  scoped_refptr<net::IOBuffer> output_io_buffer;
+  size_t output_length;
+  base::Closure intermediate_callback;
+
+  // Setup expectations and record initial state.
+  byte_stream_output->RegisterCallback(base::Bind(NullCallback));
+  null_callback_call_count_start = null_callback_call_count;
+  alt_callback_call_count_start = alt_callback_call_count;
+  EXPECT_CALL(*task_runner.get(), PostDelayedTask(_, _, 0))
+      .WillOnce(DoAll(SaveArg<1>(&intermediate_callback),
+                      Return(true)));
+
+  // Add data, and pass it across.
+  EXPECT_TRUE(Write(byte_stream_input.get(), 0, 4000));
+  message_loop_.RunAllPending();
+
+  // The task runner should have been hit, but the callback count
+  // isn't changed until we actually run the callback.
+  ::testing::Mock::VerifyAndClearExpectations(task_runner.get());
+  EXPECT_EQ(null_callback_call_count_start, null_callback_call_count);
+
+  // If we change the callback now, the new one should be run
+  // (simulates race with post task).
+  byte_stream_output->RegisterCallback(base::Bind(AltCallback));
+  EXPECT_EQ(null_callback_call_count_start, null_callback_call_count);
+  EXPECT_EQ(alt_callback_call_count_start, alt_callback_call_count);
+  intermediate_callback.Run();
+  EXPECT_EQ(null_callback_call_count_start, null_callback_call_count);
+  EXPECT_EQ(alt_callback_call_count_start+1, alt_callback_call_count);
+
+  // Final cleanup.
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 0, output_length));
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_EMPTY,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+
+}
+
+// Confirm that racing a change to a source callback with a post results
+// in the new callback being called.
+TEST_F(ByteStreamTest, SourceInterrupt) {
+  scoped_refptr<MockTaskRunner> task_runner(new StrictMock<MockTaskRunner>());
+  int null_callback_call_count_start = 0;
+  int alt_callback_call_count_start = 0;
+
+  scoped_ptr<content::ByteStreamInput> byte_stream_input;
+  scoped_ptr<content::ByteStreamOutput> byte_stream_output;
+  content::CreateByteStream(
+      &byte_stream_input, &byte_stream_output,
+      task_runner, message_loop_.message_loop_proxy(),
+      10000);
+
+  scoped_refptr<net::IOBuffer> output_io_buffer;
+  size_t output_length;
+  base::Closure intermediate_callback;
+
+  // Setup state for test and record initiali expectations
+  byte_stream_input->RegisterCallback(base::Bind(NullCallback));
+  EXPECT_TRUE(Write(byte_stream_input.get(), 0, 2000));
+  EXPECT_TRUE(Write(byte_stream_input.get(), 1, 2001));
+  EXPECT_FALSE(Write(byte_stream_input.get(), 2, 6000));
+  null_callback_call_count_start = null_callback_call_count;
+  alt_callback_call_count_start = alt_callback_call_count;
+  message_loop_.RunAllPending();
+
+  // Initial get should not trigger callback.
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 0, output_length));
+  message_loop_.RunAllPending();
+
+  // Setup expectations.
+  EXPECT_CALL(*task_runner.get(), PostDelayedTask(_, _, 0))
+      .WillOnce(DoAll(SaveArg<1>(&intermediate_callback),
+                      Return(true)));
+
+  // Second get *should* trigger callback.
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  ::testing::Mock::VerifyAndClearExpectations(task_runner.get());
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 1, output_length));
+
+  // Which should do the right thing when it's run.
+  byte_stream_input->RegisterCallback(base::Bind(AltCallback));
+  EXPECT_EQ(null_callback_call_count_start, null_callback_call_count);
+  EXPECT_EQ(alt_callback_call_count_start, alt_callback_call_count);
+  intermediate_callback.Run();
+  EXPECT_EQ(null_callback_call_count_start, null_callback_call_count);
+  EXPECT_EQ(alt_callback_call_count_start+1, alt_callback_call_count);
+
+  // Third get should also trigger callback.
+  EXPECT_CALL(*task_runner.get(), PostDelayedTask(_, _, 0))
+      .WillOnce(DoAll(SaveArg<1>(&intermediate_callback),
+                      Return(true)));
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_HAS_DATA,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+  ::testing::Mock::VerifyAndClearExpectations(task_runner.get());
+  EXPECT_TRUE(ValidateIOBuffer(output_io_buffer, 2, output_length));
+  EXPECT_EQ(content::ByteStreamOutput::STREAM_EMPTY,
+            byte_stream_output->Read(&output_io_buffer, &output_length));
+}