data_pipe_element_reader.cc 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135
  1. // Copyright 2017 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. #include "services/network/data_pipe_element_reader.h"
  5. #include "base/bind.h"
  6. #include "base/callback.h"
  7. #include "base/check_op.h"
  8. #include "base/location.h"
  9. #include "mojo/public/c/system/types.h"
  10. #include "net/base/io_buffer.h"
  11. #include "net/base/net_errors.h"
  12. namespace network {
  13. DataPipeElementReader::DataPipeElementReader(
  14. scoped_refptr<ResourceRequestBody> resource_request_body,
  15. mojo::PendingRemote<mojom::DataPipeGetter> data_pipe_getter)
  16. : resource_request_body_(std::move(resource_request_body)),
  17. data_pipe_getter_(std::move(data_pipe_getter)),
  18. handle_watcher_(FROM_HERE,
  19. mojo::SimpleWatcher::ArmingPolicy::MANUAL,
  20. base::SequencedTaskRunnerHandle::Get()) {}
  21. DataPipeElementReader::~DataPipeElementReader() {}
  22. int DataPipeElementReader::Init(net::CompletionOnceCallback callback) {
  23. DCHECK(callback);
  24. // Init rewinds the stream. Throw away current state.
  25. read_callback_.Reset();
  26. buf_ = nullptr;
  27. buf_length_ = 0;
  28. handle_watcher_.Cancel();
  29. size_ = 0;
  30. bytes_read_ = 0;
  31. // Need to do this to prevent any previously pending ReadCallback() invocation
  32. // from running.
  33. weak_factory_.InvalidateWeakPtrs();
  34. // Get a new data pipe and start.
  35. mojo::ScopedDataPipeProducerHandle producer_handle;
  36. if (mojo::CreateDataPipe(nullptr, producer_handle, data_pipe_) !=
  37. MOJO_RESULT_OK) {
  38. return net::ERR_FAILED;
  39. }
  40. data_pipe_getter_->Read(std::move(producer_handle),
  41. base::BindOnce(&DataPipeElementReader::ReadCallback,
  42. weak_factory_.GetWeakPtr()));
  43. handle_watcher_.Watch(
  44. data_pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE,
  45. base::BindRepeating(&DataPipeElementReader::OnHandleReadable,
  46. base::Unretained(this)));
  47. init_callback_ = std::move(callback);
  48. return net::ERR_IO_PENDING;
  49. }
  50. uint64_t DataPipeElementReader::GetContentLength() const {
  51. return size_;
  52. }
  53. uint64_t DataPipeElementReader::BytesRemaining() const {
  54. return size_ - bytes_read_;
  55. }
  56. int DataPipeElementReader::Read(net::IOBuffer* buf,
  57. int buf_length,
  58. net::CompletionOnceCallback callback) {
  59. DCHECK(callback);
  60. DCHECK(!read_callback_);
  61. DCHECK(!init_callback_);
  62. DCHECK(!buf_);
  63. int result = ReadInternal(buf, buf_length);
  64. if (result == net::ERR_IO_PENDING) {
  65. buf_ = buf;
  66. buf_length_ = buf_length;
  67. read_callback_ = std::move(callback);
  68. }
  69. return result;
  70. }
  71. void DataPipeElementReader::ReadCallback(int32_t status, uint64_t size) {
  72. if (status == net::OK)
  73. size_ = size;
  74. if (init_callback_)
  75. std::move(init_callback_).Run(status);
  76. }
  77. void DataPipeElementReader::OnHandleReadable(MojoResult result) {
  78. DCHECK(read_callback_);
  79. DCHECK(buf_);
  80. // Final result of the Read() call, to be passed to the consumer.
  81. int read_result;
  82. if (result == MOJO_RESULT_OK) {
  83. read_result = ReadInternal(buf_.get(), buf_length_);
  84. } else {
  85. read_result = net::ERR_FAILED;
  86. }
  87. buf_ = nullptr;
  88. buf_length_ = 0;
  89. if (read_result != net::ERR_IO_PENDING)
  90. std::move(read_callback_).Run(read_result);
  91. }
  92. int DataPipeElementReader::ReadInternal(net::IOBuffer* buf, int buf_length) {
  93. DCHECK(buf);
  94. DCHECK_GT(buf_length, 0);
  95. if (BytesRemaining() == 0)
  96. return net::OK;
  97. uint32_t num_bytes = buf_length;
  98. MojoResult rv =
  99. data_pipe_->ReadData(buf->data(), &num_bytes, MOJO_READ_DATA_FLAG_NONE);
  100. if (rv == MOJO_RESULT_OK) {
  101. bytes_read_ += num_bytes;
  102. return num_bytes;
  103. }
  104. if (rv == MOJO_RESULT_SHOULD_WAIT) {
  105. handle_watcher_.ArmOrNotify();
  106. return net::ERR_IO_PENDING;
  107. }
  108. return net::ERR_FAILED;
  109. }
  110. } // namespace network