// 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/byte_stream.h"

#include <set>
#include <utility>

#include "base/bind.h"
#include "base/containers/circular_deque.h"
#include "base/location.h"
#include "base/macros.h"
#include "base/memory/ref_counted.h"
#include "base/sequenced_task_runner.h"

namespace content {
namespace {

using ContentVector =
    base::circular_deque<std::pair<scoped_refptr<net::IOBuffer>, size_t>>;

class ByteStreamReaderImpl;

// A makeshift weak pointer; a RefCountedThreadSafe boolean that can be cleared
// in an object destructor and accessed to check for object existence. We can't
// use weak pointers because they're tightly tied to threads rather than task
// runners.
// TODO(rdsmith): A better solution would be extending weak pointers
// to support SequencedTaskRunners.
struct LifetimeFlag : public base::RefCountedThreadSafe<LifetimeFlag> {
 public:
  LifetimeFlag() : is_alive(true) { }
  bool is_alive;

 protected:
  friend class base::RefCountedThreadSafe<LifetimeFlag>;
  virtual ~LifetimeFlag() { }

 private:
  DISALLOW_COPY_AND_ASSIGN(LifetimeFlag);
};

// For both ByteStreamWriterImpl and ByteStreamReaderImpl, Construction and
// SetPeer may happen anywhere; all other operations on each class must
// happen in the context of their SequencedTaskRunner.
class ByteStreamWriterImpl : public ByteStreamWriter {
 public:
  ByteStreamWriterImpl(scoped_refptr<base::SequencedTaskRunner> task_runner,
                       scoped_refptr<LifetimeFlag> lifetime_flag,
                       size_t buffer_size);
  ~ByteStreamWriterImpl() override;

  // Must be called before any operations are performed.
  void SetPeer(ByteStreamReaderImpl* peer,
               scoped_refptr<base::SequencedTaskRunner> peer_task_runner,
               scoped_refptr<LifetimeFlag> peer_lifetime_flag);

  // Overridden from ByteStreamWriter.
  bool Write(scoped_refptr<net::IOBuffer> buffer, size_t byte_count) override;
  void Flush() override;
  void Close(int status) override;
  void RegisterCallback(base::RepeatingClosure source_callback) override;
  size_t GetTotalBufferedBytes() const override;

  // PostTask target from |ByteStreamReaderImpl::MaybeUpdateInput|.
  static void UpdateWindow(scoped_refptr<LifetimeFlag> lifetime_flag,
                           ByteStreamWriterImpl* target,
                           size_t bytes_consumed);

 private:
  // Called from UpdateWindow when object existence has been validated.
  void UpdateWindowInternal(size_t bytes_consumed);

  void PostToPeer(bool complete, int status);

  const size_t total_buffer_size_;

  // All data objects in this class are only valid to access on
  // this task runner except as otherwise noted.
  scoped_refptr<base::SequencedTaskRunner> my_task_runner_;

  // True while this object is alive.
  scoped_refptr<LifetimeFlag> my_lifetime_flag_;

  base::RepeatingClosure space_available_callback_;
  ContentVector input_contents_;
  size_t input_contents_size_;

  // ** Peer information.

  scoped_refptr<base::SequencedTaskRunner> peer_task_runner_;

  // How much we've sent to the output that for flow control purposes we
  // must assume hasn't been read yet.
  size_t output_size_used_;

  // Only valid to access on peer_task_runner_.
  scoped_refptr<LifetimeFlag> peer_lifetime_flag_;

  // Only valid to access on peer_task_runner_ if
  // |*peer_lifetime_flag_ == true|
  ByteStreamReaderImpl* peer_;
};

class ByteStreamReaderImpl : public ByteStreamReader {
 public:
  ByteStreamReaderImpl(scoped_refptr<base::SequencedTaskRunner> task_runner,
                       scoped_refptr<LifetimeFlag> lifetime_flag,
                       size_t buffer_size);
  ~ByteStreamReaderImpl() override;

  // Must be called before any operations are performed.
  void SetPeer(ByteStreamWriterImpl* peer,
               scoped_refptr<base::SequencedTaskRunner> peer_task_runner,
               scoped_refptr<LifetimeFlag> peer_lifetime_flag);

  // Overridden from ByteStreamReader.
  StreamState Read(scoped_refptr<net::IOBuffer>* data, size_t* length) override;
  int GetStatus() const override;
  void RegisterCallback(base::RepeatingClosure sink_callback) override;

  // PostTask target from |ByteStreamWriterImpl::Write| and
  // |ByteStreamWriterImpl::Close|.
  // Receive data from our peer.
  // static because it may be called after the object it is targeting
  // has been destroyed.  It may not access |*target|
  // if |*object_lifetime_flag| is false.
  static void TransferData(scoped_refptr<LifetimeFlag> object_lifetime_flag,
                           ByteStreamReaderImpl* target,
                           std::unique_ptr<ContentVector> transfer_buffer,
                           size_t transfer_buffer_bytes,
                           bool source_complete,
                           int status);

 private:
  // Called from TransferData once object existence has been validated.
  void TransferDataInternal(std::unique_ptr<ContentVector> transfer_buffer,
                            size_t transfer_buffer_bytes,
                            bool source_complete,
                            int status);

  void MaybeUpdateInput();

  const size_t total_buffer_size_;

  scoped_refptr<base::SequencedTaskRunner> my_task_runner_;

  // True while this object is alive.
  scoped_refptr<LifetimeFlag> my_lifetime_flag_;

  ContentVector available_contents_;

  bool received_status_;
  int status_;

  base::RepeatingClosure data_available_callback_;

  // Time of last point at which data in stream transitioned from full
  // to non-full.  Nulled when a callback is sent.
  base::Time last_non_full_time_;

  // ** Peer information

  scoped_refptr<base::SequencedTaskRunner> peer_task_runner_;

  // How much has been removed from this class that we haven't told
  // the input about yet.
  size_t unreported_consumed_bytes_;

  // Only valid to access on peer_task_runner_.
  scoped_refptr<LifetimeFlag> peer_lifetime_flag_;

  // Only valid to access on peer_task_runner_ if
  // |*peer_lifetime_flag_ == true|
  ByteStreamWriterImpl* peer_;
};

ByteStreamWriterImpl::ByteStreamWriterImpl(
    scoped_refptr<base::SequencedTaskRunner> task_runner,
    scoped_refptr<LifetimeFlag> lifetime_flag,
    size_t buffer_size)
    : total_buffer_size_(buffer_size),
      my_task_runner_(task_runner),
      my_lifetime_flag_(lifetime_flag),
      input_contents_size_(0),
      output_size_used_(0),
      peer_(nullptr) {
  DCHECK(my_lifetime_flag_.get());
  my_lifetime_flag_->is_alive = true;
}

ByteStreamWriterImpl::~ByteStreamWriterImpl() {
  // No RunsTasksInCurrentSequence() check to allow deleting a created writer
  // before we start using it. Once started, should be deleted on the specified
  // task runner.
  my_lifetime_flag_->is_alive = false;
}

void ByteStreamWriterImpl::SetPeer(
    ByteStreamReaderImpl* peer,
    scoped_refptr<base::SequencedTaskRunner> peer_task_runner,
    scoped_refptr<LifetimeFlag> peer_lifetime_flag) {
  peer_ = peer;
  peer_task_runner_ = peer_task_runner;
  peer_lifetime_flag_ = peer_lifetime_flag;
}

bool ByteStreamWriterImpl::Write(
    scoped_refptr<net::IOBuffer> buffer, size_t byte_count) {
  DCHECK(my_task_runner_->RunsTasksInCurrentSequence());

  // Check overflow.
  //
  // TODO(tyoshino): Discuss with content/browser/download developer and if
  // they're fine with, set smaller limit and make it configurable.
  size_t space_limit = std::numeric_limits<size_t>::max() -
      GetTotalBufferedBytes();
  if (byte_count > space_limit) {
    // TODO(tyoshino): Tell the user that Write() failed.
    // Ignore input.
    return false;
  }

  input_contents_.push_back(std::make_pair(buffer, byte_count));
  input_contents_size_ += byte_count;

  // Arbitrarily, we buffer to a third of the total size before sending.
  if (input_contents_size_ > total_buffer_size_ / kFractionBufferBeforeSending)
    PostToPeer(false, 0);

  return GetTotalBufferedBytes() <= total_buffer_size_;
}

void ByteStreamWriterImpl::Flush() {
  DCHECK(my_task_runner_->RunsTasksInCurrentSequence());
  if (input_contents_size_ > 0)
    PostToPeer(false, 0);
}

void ByteStreamWriterImpl::Close(int status) {
  DCHECK(my_task_runner_->RunsTasksInCurrentSequence());
  PostToPeer(true, status);
}

void ByteStreamWriterImpl::RegisterCallback(
    base::RepeatingClosure source_callback) {
  DCHECK(my_task_runner_->RunsTasksInCurrentSequence());
  space_available_callback_ = std::move(source_callback);
}

size_t ByteStreamWriterImpl::GetTotalBufferedBytes() const {
  DCHECK(my_task_runner_->RunsTasksInCurrentSequence());
  // This sum doesn't overflow since Write() fails if this sum is going to
  // overflow.
  return input_contents_size_ + output_size_used_;
}

// static
void ByteStreamWriterImpl::UpdateWindow(
    scoped_refptr<LifetimeFlag> lifetime_flag, ByteStreamWriterImpl* target,
    size_t bytes_consumed) {
  // If the target object isn't alive anymore, we do nothing.
  if (!lifetime_flag->is_alive) return;

  target->UpdateWindowInternal(bytes_consumed);
}

void ByteStreamWriterImpl::UpdateWindowInternal(size_t bytes_consumed) {
  DCHECK(my_task_runner_->RunsTasksInCurrentSequence());

  bool was_above_limit = GetTotalBufferedBytes() > total_buffer_size_;

  DCHECK_GE(output_size_used_, bytes_consumed);
  output_size_used_ -= bytes_consumed;

  // Callback if we were above the limit and we're now <= to it.
  bool no_longer_above_limit = GetTotalBufferedBytes() <= total_buffer_size_;

  if (no_longer_above_limit && was_above_limit &&
      !space_available_callback_.is_null())
    space_available_callback_.Run();
}

void ByteStreamWriterImpl::PostToPeer(bool complete, int status) {
  DCHECK(my_task_runner_->RunsTasksInCurrentSequence());
  // Valid contexts in which to call.
  DCHECK(complete || 0 != input_contents_size_);

  std::unique_ptr<ContentVector> transfer_buffer;
  size_t buffer_size = 0;
  if (0 != input_contents_size_) {
    transfer_buffer.reset(new ContentVector);
    transfer_buffer->swap(input_contents_);
    buffer_size = input_contents_size_;
    output_size_used_ += input_contents_size_;
    input_contents_size_ = 0;
  }
  peer_task_runner_->PostTask(
      FROM_HERE,
      base::BindOnce(&ByteStreamReaderImpl::TransferData, peer_lifetime_flag_,
                     peer_, std::move(transfer_buffer), buffer_size, complete,
                     status));
}

ByteStreamReaderImpl::ByteStreamReaderImpl(
    scoped_refptr<base::SequencedTaskRunner> task_runner,
    scoped_refptr<LifetimeFlag> lifetime_flag,
    size_t buffer_size)
    : total_buffer_size_(buffer_size),
      my_task_runner_(task_runner),
      my_lifetime_flag_(lifetime_flag),
      received_status_(false),
      status_(0),
      unreported_consumed_bytes_(0),
      peer_(nullptr) {
  DCHECK(my_lifetime_flag_.get());
  my_lifetime_flag_->is_alive = true;
}

ByteStreamReaderImpl::~ByteStreamReaderImpl() {
  // No RunsTasksInCurrentSequence() check to allow deleting a created writer
  // before we start using it. Once started, should be deleted on the specified
  // task runner.
  my_lifetime_flag_->is_alive = false;
}

void ByteStreamReaderImpl::SetPeer(
    ByteStreamWriterImpl* peer,
    scoped_refptr<base::SequencedTaskRunner> peer_task_runner,
    scoped_refptr<LifetimeFlag> peer_lifetime_flag) {
  peer_ = peer;
  peer_task_runner_ = peer_task_runner;
  peer_lifetime_flag_ = peer_lifetime_flag;
}

ByteStreamReaderImpl::StreamState
ByteStreamReaderImpl::Read(scoped_refptr<net::IOBuffer>* data,
                           size_t* length) {
  DCHECK(my_task_runner_->RunsTasksInCurrentSequence());

  if (available_contents_.size()) {
    *data = available_contents_.front().first;
    *length = available_contents_.front().second;
    available_contents_.pop_front();
    unreported_consumed_bytes_ += *length;

    MaybeUpdateInput();
    return STREAM_HAS_DATA;
  }
  if (received_status_) {
    return STREAM_COMPLETE;
  }
  return STREAM_EMPTY;
}

int ByteStreamReaderImpl::GetStatus() const {
  DCHECK(my_task_runner_->RunsTasksInCurrentSequence());
  DCHECK(received_status_);
  return status_;
}

void ByteStreamReaderImpl::RegisterCallback(
    base::RepeatingClosure sink_callback) {
  DCHECK(my_task_runner_->RunsTasksInCurrentSequence());

  data_available_callback_ = std::move(sink_callback);
}

// static
void ByteStreamReaderImpl::TransferData(
    scoped_refptr<LifetimeFlag> object_lifetime_flag,
    ByteStreamReaderImpl* target,
    std::unique_ptr<ContentVector> transfer_buffer,
    size_t buffer_size,
    bool source_complete,
    int status) {
  // If our target is no longer alive, do nothing.
  if (!object_lifetime_flag->is_alive) return;

  target->TransferDataInternal(std::move(transfer_buffer), buffer_size,
                               source_complete, status);
}

void ByteStreamReaderImpl::TransferDataInternal(
    std::unique_ptr<ContentVector> transfer_buffer,
    size_t buffer_size,
    bool source_complete,
    int status) {
  DCHECK(my_task_runner_->RunsTasksInCurrentSequence());

  bool was_empty = available_contents_.empty();

  if (transfer_buffer) {
    available_contents_.insert(available_contents_.end(),
                               transfer_buffer->begin(),
                               transfer_buffer->end());
  }

  if (source_complete) {
    received_status_ = true;
    status_ = status;
  }

  // Callback on transition from empty to non-empty, or
  // source complete.
  if (((was_empty && !available_contents_.empty()) ||
       source_complete) &&
      !data_available_callback_.is_null())
    data_available_callback_.Run();
}

// Decide whether or not to send the input a window update.
// Currently we do that whenever we've got unreported consumption
// greater than 1/3 of total size.
void ByteStreamReaderImpl::MaybeUpdateInput() {
  DCHECK(my_task_runner_->RunsTasksInCurrentSequence());

  if (unreported_consumed_bytes_ <=
      total_buffer_size_ / kFractionReadBeforeWindowUpdate)
    return;

  peer_task_runner_->PostTask(
      FROM_HERE,
      base::BindOnce(&ByteStreamWriterImpl::UpdateWindow, peer_lifetime_flag_,
                     peer_, unreported_consumed_bytes_));
  unreported_consumed_bytes_ = 0;
}

}  // namespace

const int ByteStreamWriter::kFractionBufferBeforeSending = 3;
const int ByteStreamReader::kFractionReadBeforeWindowUpdate = 3;

ByteStreamReader::~ByteStreamReader() { }

ByteStreamWriter::~ByteStreamWriter() { }

void CreateByteStream(
    scoped_refptr<base::SequencedTaskRunner> input_task_runner,
    scoped_refptr<base::SequencedTaskRunner> output_task_runner,
    size_t buffer_size,
    std::unique_ptr<ByteStreamWriter>* input,
    std::unique_ptr<ByteStreamReader>* output) {
  scoped_refptr<LifetimeFlag> input_flag(new LifetimeFlag());
  scoped_refptr<LifetimeFlag> output_flag(new LifetimeFlag());

  ByteStreamWriterImpl* in = new ByteStreamWriterImpl(
      input_task_runner, input_flag, buffer_size);
  ByteStreamReaderImpl* out = new ByteStreamReaderImpl(
      output_task_runner, output_flag, buffer_size);

  in->SetPeer(out, output_task_runner, output_flag);
  out->SetPeer(in, input_task_runner, input_flag);
  input->reset(in);
  output->reset(out);
}

}  // namespace content