chrome/browser/chromeos/file_system_provider/fileapi/buffering_file_stream_writer.cc

// Copyright 2014 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 "chrome/browser/chromeos/file_system_provider/fileapi/buffering_file_stream_writer.h"

#include <algorithm>

#include "net/base/io_buffer.h"
#include "net/base/net_errors.h"

namespace chromeos {
namespace file_system_provider {

BufferingFileStreamWriter::BufferingFileStreamWriter(
    scoped_ptr<storage::FileStreamWriter> file_stream_writer,
    int intermediate_buffer_length)
    : file_stream_writer_(file_stream_writer.Pass()),
      intermediate_buffer_length_(intermediate_buffer_length),
      intermediate_buffer_(new net::IOBuffer(intermediate_buffer_length_)),
      buffered_bytes_(0),
      weak_ptr_factory_(this) {
}

BufferingFileStreamWriter::~BufferingFileStreamWriter() {
  if (buffered_bytes_)
    LOG(ERROR) << "File stream writer not flushed. Data will be lost.";
}

int BufferingFileStreamWriter::Write(net::IOBuffer* buffer,
                                     int buffer_length,
                                     const net::CompletionCallback& callback) {
  // If |buffer_length| is larger than the intermediate buffer, then call the
  // inner file stream writer directly. Note, that the intermediate buffer
  // (used for buffering) must be flushed first.
  if (buffer_length > intermediate_buffer_length_) {
    if (buffered_bytes_) {
      FlushIntermediateBuffer(
          base::Bind(&BufferingFileStreamWriter::
                         OnFlushIntermediateBufferForDirectWriteCompleted,
                     weak_ptr_factory_.GetWeakPtr(),
                     make_scoped_refptr(buffer),
                     buffer_length,
                     callback));
    } else {
      // Nothing to flush, so skip it.
      OnFlushIntermediateBufferForDirectWriteCompleted(
          make_scoped_refptr(buffer), buffer_length, callback, net::OK);
    }
    return net::ERR_IO_PENDING;
  }

  // Buffer consecutive writes to larger chunks.
  const int buffer_bytes =
      std::min(intermediate_buffer_length_ - buffered_bytes_, buffer_length);

  CopyToIntermediateBuffer(
      make_scoped_refptr(buffer), 0 /* buffer_offset */, buffer_bytes);
  const int bytes_left = buffer_length - buffer_bytes;

  if (buffered_bytes_ == intermediate_buffer_length_) {
    FlushIntermediateBuffer(
        base::Bind(&BufferingFileStreamWriter::
                       OnFlushIntermediateBufferForBufferedWriteCompleted,
                   weak_ptr_factory_.GetWeakPtr(),
                   make_scoped_refptr(buffer),
                   buffer_bytes,
                   bytes_left,
                   callback));
    return net::ERR_IO_PENDING;
  }

  // Optimistically return a success.
  return buffer_length;
}

int BufferingFileStreamWriter::Cancel(const net::CompletionCallback& callback) {
  // Since there is no any asynchronous call in this class other than on
  // |file_stream_writer_|, then there must be an in-flight operation going on.
  return file_stream_writer_->Cancel(callback);
}

int BufferingFileStreamWriter::Flush(const net::CompletionCallback& callback) {
  // Flush all the buffered bytes first, then invoke Flush() on the inner file
  // stream writer.
  FlushIntermediateBuffer(base::Bind(
      &BufferingFileStreamWriter::OnFlushIntermediateBufferForFlushCompleted,
      weak_ptr_factory_.GetWeakPtr(),
      callback));
  return net::ERR_IO_PENDING;
}

void BufferingFileStreamWriter::CopyToIntermediateBuffer(
    scoped_refptr<net::IOBuffer> buffer,
    int buffer_offset,
    int buffer_length) {
  DCHECK_GE(intermediate_buffer_length_, buffer_length + buffered_bytes_);
  memcpy(intermediate_buffer_->data() + buffered_bytes_,
         buffer->data() + buffer_offset,
         buffer_length);
  buffered_bytes_ += buffer_length;
}

void BufferingFileStreamWriter::FlushIntermediateBuffer(
    const net::CompletionCallback& callback) {
  const int result = file_stream_writer_->Write(
      intermediate_buffer_.get(),
      buffered_bytes_,
      base::Bind(&BufferingFileStreamWriter::OnFlushIntermediateBufferCompleted,
                 weak_ptr_factory_.GetWeakPtr(),
                 buffered_bytes_,
                 callback));
  DCHECK_EQ(net::ERR_IO_PENDING, result);
}

void BufferingFileStreamWriter::OnFlushIntermediateBufferCompleted(
    int length,
    const net::CompletionCallback& callback,
    int result) {
  if (result < 0) {
    callback.Run(result);
    return;
  }

  DCHECK_EQ(length, result) << "Partial writes are not supported.";
  buffered_bytes_ = 0;

  callback.Run(net::OK);
}

void
BufferingFileStreamWriter::OnFlushIntermediateBufferForDirectWriteCompleted(
    scoped_refptr<net::IOBuffer> buffer,
    int length,
    const net::CompletionCallback& callback,
    int result) {
  if (result < 0) {
    callback.Run(result);
    return;
  }

  // The following logic is only valid if the intermediate buffer is empty.
  DCHECK_EQ(0, buffered_bytes_);

  const int write_result =
      file_stream_writer_->Write(buffer.get(), length, callback);
  DCHECK_EQ(net::ERR_IO_PENDING, write_result);
}

void
BufferingFileStreamWriter::OnFlushIntermediateBufferForBufferedWriteCompleted(
    scoped_refptr<net::IOBuffer> buffer,
    int buffered_bytes,
    int bytes_left,
    const net::CompletionCallback& callback,
    int result) {
  if (result < 0) {
    callback.Run(result);
    return;
  }

  // Copy the rest of bytes to the buffer.
  DCHECK_EQ(net::OK, result);
  DCHECK_EQ(0, buffered_bytes_);
  DCHECK_GE(intermediate_buffer_length_, bytes_left);
  CopyToIntermediateBuffer(buffer, buffered_bytes, bytes_left);

  callback.Run(buffered_bytes + bytes_left);
}

void BufferingFileStreamWriter::OnFlushIntermediateBufferForFlushCompleted(
    const net::CompletionCallback& callback,
    int result) {
  if (result < 0) {
    callback.Run(result);
    return;
  }

  const int flush_result = file_stream_writer_->Flush(callback);
  DCHECK_EQ(net::ERR_IO_PENDING, flush_result);
}

}  // namespace file_system_provider
}  // namespace chromeos