content/browser/streams/stream_unittest.cc

   1 // Copyright (c) 2013 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "base/message_loop/message_loop.h"
   6 #include "base/test/test_simple_task_runner.h"
   7 #include "content/browser/streams/stream.h"
   8 #include "content/browser/streams/stream_read_observer.h"
   9 #include "content/browser/streams/stream_register_observer.h"
  10 #include "content/browser/streams/stream_registry.h"
  11 #include "content/browser/streams/stream_write_observer.h"
  12 #include "testing/gtest/include/gtest/gtest.h"
  13
  14 namespace content {
  15
  16 class StreamTest : public testing::Test {
  17  public:
  18   StreamTest() : producing_seed_key_(0) {}
  19
  20   void SetUp() override { registry_.reset(new StreamRegistry()); }
  21
  22   // Create a new IO buffer of the given |buffer_size| and fill it with random
  23   // data.
  24   scoped_refptr<net::IOBuffer> NewIOBuffer(size_t buffer_size) {
  25     scoped_refptr<net::IOBuffer> buffer(new net::IOBuffer(buffer_size));
  26     char *bufferp = buffer->data();
  27     for (size_t i = 0; i < buffer_size; i++)
  28       bufferp[i] = (i + producing_seed_key_) % (1 << sizeof(char));
  29     ++producing_seed_key_;
  30     return buffer;
  31   }
  32
  33  protected:
  34   base::MessageLoop message_loop_;
  35   scoped_ptr<StreamRegistry> registry_;
  36
  37  private:
  38   int producing_seed_key_;
  39 };
  40
  41 class TestStreamReader : public StreamReadObserver {
  42  public:
  43   TestStreamReader() : buffer_(new net::GrowableIOBuffer()), completed_(false) {
  44   }
  45   ~TestStreamReader() override {}
  46
  47   void Read(Stream* stream) {
  48     const size_t kBufferSize = 32768;
  49     scoped_refptr<net::IOBuffer> buffer(new net::IOBuffer(kBufferSize));
  50
  51     int bytes_read = 0;
  52     while (true) {
  53       Stream::StreamState state =
  54           stream->ReadRawData(buffer.get(), kBufferSize, &bytes_read);
  55       switch (state) {
  56         case Stream::STREAM_HAS_DATA:
  57           // TODO(tyoshino): Move these expectations to the beginning of Read()
  58           // method once Stream::Finalize() is fixed.
  59           EXPECT_FALSE(completed_);
  60           break;
  61         case Stream::STREAM_COMPLETE:
  62           completed_ = true;
  63           return;
  64         case Stream::STREAM_EMPTY:
  65           EXPECT_FALSE(completed_);
  66           return;
  67         case Stream::STREAM_ABORTED:
  68           EXPECT_FALSE(completed_);
  69           return;
  70       }
  71       size_t old_capacity = buffer_->capacity();
  72       buffer_->SetCapacity(old_capacity + bytes_read);
  73       memcpy(buffer_->StartOfBuffer() + old_capacity,
  74              buffer->data(), bytes_read);
  75     }
  76   }
  77
  78   void OnDataAvailable(Stream* stream) override { Read(stream); }
  79
  80   scoped_refptr<net::GrowableIOBuffer> buffer() { return buffer_; }
  81
  82   bool completed() const {
  83     return completed_;
  84   }
  85
  86  private:
  87   scoped_refptr<net::GrowableIOBuffer> buffer_;
  88   bool completed_;
  89 };
  90
  91 class TestStreamWriter : public StreamWriteObserver {
  92  public:
  93   TestStreamWriter() {}
  94   ~TestStreamWriter() override {}
  95
  96   void Write(Stream* stream,
  97              scoped_refptr<net::IOBuffer> buffer,
  98              size_t buffer_size) {
  99     stream->AddData(buffer, buffer_size);
 100   }
 101
 102   void OnSpaceAvailable(Stream* stream) override {}
 103
 104   void OnClose(Stream* stream) override {}
 105 };
 106
 107 class TestStreamObserver : public StreamRegisterObserver {
 108  public:
 109   TestStreamObserver(const GURL& url, StreamRegistry* registry)
 110       : url_(url), registry_(registry), registered_(false), stream_(nullptr) {
 111     registry->SetRegisterObserver(url, this);
 112   }
 113   ~TestStreamObserver() override { registry_->RemoveRegisterObserver(url_); }
 114   void OnStreamRegistered(Stream* stream) override {
 115     registered_ = true;
 116     stream_ = stream;
 117   }
 118   bool registered() const { return registered_; }
 119   Stream* stream() const { return stream_; }
 120
 121  private:
 122   const GURL url_;
 123   StreamRegistry* registry_;
 124   bool registered_;
 125   Stream* stream_;
 126 };
 127
 128 TEST_F(StreamTest, SetAndRemoveRegisterObserver) {
 129   TestStreamWriter writer1;
 130   TestStreamWriter writer2;
 131   GURL url1("blob://stream1");
 132   GURL url2("blob://stream2");
 133   scoped_ptr<TestStreamObserver> observer1(
 134       new TestStreamObserver(url1, registry_.get()));
 135   scoped_ptr<TestStreamObserver> observer2(
 136       new TestStreamObserver(url2, registry_.get()));
 137   scoped_refptr<Stream> stream1(new Stream(registry_.get(), &writer1, url1));
 138   EXPECT_TRUE(observer1->registered());
 139   EXPECT_EQ(observer1->stream(), stream1.get());
 140   EXPECT_FALSE(observer2->registered());
 141
 142   observer2.reset();
 143   scoped_refptr<Stream> stream2(new Stream(registry_.get(), &writer2, url2));
 144 }
 145
 146 TEST_F(StreamTest, SetReadObserver) {
 147   TestStreamReader reader;
 148   TestStreamWriter writer;
 149
 150   GURL url("blob://stream");
 151   scoped_refptr<Stream> stream(
 152       new Stream(registry_.get(), &writer, url));
 153   EXPECT_TRUE(stream->SetReadObserver(&reader));
 154 }
 155
 156 TEST_F(StreamTest, SetReadObserver_SecondFails) {
 157   TestStreamReader reader1;
 158   TestStreamReader reader2;
 159   TestStreamWriter writer;
 160
 161   GURL url("blob://stream");
 162   scoped_refptr<Stream> stream(
 163       new Stream(registry_.get(), &writer, url));
 164   EXPECT_TRUE(stream->SetReadObserver(&reader1));
 165   EXPECT_FALSE(stream->SetReadObserver(&reader2));
 166 }
 167
 168 TEST_F(StreamTest, SetReadObserver_TwoReaders) {
 169   TestStreamReader reader1;
 170   TestStreamReader reader2;
 171   TestStreamWriter writer;
 172
 173   GURL url("blob://stream");
 174   scoped_refptr<Stream> stream(
 175       new Stream(registry_.get(), &writer, url));
 176   EXPECT_TRUE(stream->SetReadObserver(&reader1));
 177
 178   // Once the first read observer is removed, a new one can be added.
 179   stream->RemoveReadObserver(&reader1);
 180   EXPECT_TRUE(stream->SetReadObserver(&reader2));
 181 }
 182
 183 TEST_F(StreamTest, Stream) {
 184   TestStreamReader reader;
 185   TestStreamWriter writer;
 186
 187   GURL url("blob://stream");
 188   scoped_refptr<Stream> stream(
 189       new Stream(registry_.get(), &writer, url));
 190   EXPECT_TRUE(stream->SetReadObserver(&reader));
 191
 192   const int kBufferSize = 1000000;
 193   scoped_refptr<net::IOBuffer> buffer(NewIOBuffer(kBufferSize));
 194   writer.Write(stream.get(), buffer, kBufferSize);
 195   stream->Finalize();
 196   base::MessageLoop::current()->RunUntilIdle();
 197   EXPECT_TRUE(reader.completed());
 198
 199   ASSERT_EQ(reader.buffer()->capacity(), kBufferSize);
 200   for (int i = 0; i < kBufferSize; i++)
 201     EXPECT_EQ(buffer->data()[i], reader.buffer()->data()[i]);
 202 }
 203
 204 // Test that even if a reader receives an empty buffer, once TransferData()
 205 // method is called on it with |source_complete| = true, following Read() calls
 206 // on it never returns STREAM_EMPTY. Together with StreamTest.Stream above, this
 207 // guarantees that Reader::Read() call returns only STREAM_HAS_DATA
 208 // or STREAM_COMPLETE in |data_available_callback_| call corresponding to
 209 // Writer::Close().
 210 TEST_F(StreamTest, ClosedReaderDoesNotReturnStreamEmpty) {
 211   TestStreamReader reader;
 212   TestStreamWriter writer;
 213
 214   GURL url("blob://stream");
 215   scoped_refptr<Stream> stream(
 216       new Stream(registry_.get(), &writer, url));
 217   EXPECT_TRUE(stream->SetReadObserver(&reader));
 218
 219   const int kBufferSize = 0;
 220   scoped_refptr<net::IOBuffer> buffer(NewIOBuffer(kBufferSize));
 221   stream->AddData(buffer, kBufferSize);
 222   stream->Finalize();
 223   base::MessageLoop::current()->RunUntilIdle();
 224   EXPECT_TRUE(reader.completed());
 225   EXPECT_EQ(0, reader.buffer()->capacity());
 226 }
 227
 228 TEST_F(StreamTest, GetStream) {
 229   TestStreamWriter writer;
 230
 231   GURL url("blob://stream");
 232   scoped_refptr<Stream> stream1(
 233       new Stream(registry_.get(), &writer, url));
 234
 235   scoped_refptr<Stream> stream2 = registry_->GetStream(url);
 236   ASSERT_EQ(stream1, stream2);
 237 }
 238
 239 TEST_F(StreamTest, GetStream_Missing) {
 240   TestStreamWriter writer;
 241
 242   GURL url1("blob://stream");
 243   scoped_refptr<Stream> stream1(
 244       new Stream(registry_.get(), &writer, url1));
 245
 246   GURL url2("blob://stream2");
 247   scoped_refptr<Stream> stream2 = registry_->GetStream(url2);
 248   ASSERT_FALSE(stream2.get());
 249 }
 250
 251 TEST_F(StreamTest, CloneStream) {
 252   TestStreamWriter writer;
 253
 254   GURL url1("blob://stream");
 255   scoped_refptr<Stream> stream1(
 256       new Stream(registry_.get(), &writer, url1));
 257
 258   GURL url2("blob://stream2");
 259   ASSERT_TRUE(registry_->CloneStream(url2, url1));
 260   scoped_refptr<Stream> stream2 = registry_->GetStream(url2);
 261   ASSERT_EQ(stream1, stream2);
 262 }
 263
 264 TEST_F(StreamTest, CloneStream_Missing) {
 265   TestStreamWriter writer;
 266
 267   GURL url1("blob://stream");
 268   scoped_refptr<Stream> stream1(
 269       new Stream(registry_.get(), &writer, url1));
 270
 271   GURL url2("blob://stream2");
 272   GURL url3("blob://stream3");
 273   ASSERT_FALSE(registry_->CloneStream(url2, url3));
 274   scoped_refptr<Stream> stream2 = registry_->GetStream(url2);
 275   ASSERT_FALSE(stream2.get());
 276 }
 277
 278 TEST_F(StreamTest, UnregisterStream) {
 279   TestStreamWriter writer;
 280
 281   GURL url("blob://stream");
 282   scoped_refptr<Stream> stream1(
 283       new Stream(registry_.get(), &writer, url));
 284
 285   registry_->UnregisterStream(url);
 286   scoped_refptr<Stream> stream2 = registry_->GetStream(url);
 287   ASSERT_FALSE(stream2.get());
 288 }
 289
 290 TEST_F(StreamTest, MemoryExceedMemoryUsageLimit) {
 291   TestStreamWriter writer1;
 292   TestStreamWriter writer2;
 293
 294   GURL url1("blob://stream");
 295   scoped_refptr<Stream> stream1(
 296       new Stream(registry_.get(), &writer1, url1));
 297
 298   GURL url2("blob://stream2");
 299   scoped_refptr<Stream> stream2(
 300       new Stream(registry_.get(), &writer2, url2));
 301
 302   const int kMaxMemoryUsage = 1500000;
 303   registry_->set_max_memory_usage_for_testing(kMaxMemoryUsage);
 304
 305   const int kBufferSize = 1000000;
 306   scoped_refptr<net::IOBuffer> buffer(NewIOBuffer(kBufferSize));
 307   writer1.Write(stream1.get(), buffer, kBufferSize);
 308   // Make transfer happen.
 309   base::MessageLoop::current()->RunUntilIdle();
 310
 311   writer2.Write(stream2.get(), buffer, kBufferSize);
 312
 313   // Written data (1000000 * 2) exceeded limit (1500000). |stream2| should be
 314   // unregistered with |registry_|.
 315   EXPECT_EQ(nullptr, registry_->GetStream(url2).get());
 316
 317   writer1.Write(stream1.get(), buffer, kMaxMemoryUsage - kBufferSize);
 318   // Should be accepted since stream2 is unregistered and the new data is not
 319   // so big to exceed the limit.
 320   EXPECT_FALSE(registry_->GetStream(url1).get() == nullptr);
 321 }
 322
 323 TEST_F(StreamTest, UnderMemoryUsageLimit) {
 324   TestStreamWriter writer;
 325   TestStreamReader reader;
 326
 327   GURL url("blob://stream");
 328   scoped_refptr<Stream> stream(new Stream(registry_.get(), &writer, url));
 329   EXPECT_TRUE(stream->SetReadObserver(&reader));
 330
 331   registry_->set_max_memory_usage_for_testing(1500000);
 332
 333   const int kBufferSize = 1000000;
 334   scoped_refptr<net::IOBuffer> buffer(NewIOBuffer(kBufferSize));
 335   writer.Write(stream.get(), buffer, kBufferSize);
 336
 337   // Run loop to make |reader| consume the data.
 338   base::MessageLoop::current()->RunUntilIdle();
 339
 340   writer.Write(stream.get(), buffer, kBufferSize);
 341
 342   EXPECT_EQ(stream.get(), registry_->GetStream(url).get());
 343 }
 344
 345 TEST_F(StreamTest, Flush) {
 346   TestStreamWriter writer;
 347   TestStreamReader reader;
 348
 349   GURL url("blob://stream");
 350   scoped_refptr<Stream> stream(new Stream(registry_.get(), &writer, url));
 351   EXPECT_TRUE(stream->SetReadObserver(&reader));
 352
 353   // If the written data size is smaller than ByteStreamWriter's (total size /
 354   // kFractionBufferBeforeSending), StreamReadObserver::OnDataAvailable is not
 355   // called.
 356   const int kBufferSize = 1;
 357   scoped_refptr<net::IOBuffer> buffer(NewIOBuffer(kBufferSize));
 358   writer.Write(stream.get(), buffer, kBufferSize);
 359
 360   // Run loop to make |reader| consume the data.
 361   base::MessageLoop::current()->RunUntilIdle();
 362   EXPECT_EQ(0, reader.buffer()->capacity());
 363
 364   stream->Flush();
 365
 366   // Run loop to make |reader| consume the data.
 367   base::MessageLoop::current()->RunUntilIdle();
 368   EXPECT_EQ(kBufferSize, reader.buffer()->capacity());
 369
 370   EXPECT_EQ(stream.get(), registry_->GetStream(url).get());
 371 }
 372
 373 TEST_F(StreamTest, AbortPendingStream) {
 374   TestStreamWriter writer;
 375
 376   GURL url("blob://stream");
 377   registry_->AbortPendingStream(url);
 378   scoped_refptr<Stream> stream1(new Stream(registry_.get(), &writer, url));
 379   ASSERT_EQ(nullptr, registry_->GetStream(url).get());
 380 }
 381
 382 }  // namespace content