upstream_loader.cc 4.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132
  1. // Copyright 2020 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 "components/speech/upstream_loader.h"
  5. #include "components/speech/upstream_loader_client.h"
  6. #include "services/network/public/mojom/url_response_head.mojom.h"
  7. namespace speech {
  8. UpstreamLoader::UpstreamLoader(
  9. std::unique_ptr<network::ResourceRequest> resource_request,
  10. net::NetworkTrafficAnnotationTag upstream_traffic_annotation,
  11. network::mojom::URLLoaderFactory* url_loader_factory,
  12. UpstreamLoaderClient* upstream_loader_client)
  13. : upstream_loader_client_(upstream_loader_client) {
  14. DCHECK(upstream_loader_client_);
  15. // Attach a chunked upload body.
  16. mojo::PendingRemote<network::mojom::ChunkedDataPipeGetter> data_remote;
  17. receiver_set_.Add(this, data_remote.InitWithNewPipeAndPassReceiver());
  18. resource_request->request_body =
  19. base::MakeRefCounted<network::ResourceRequestBody>();
  20. resource_request->request_body->SetToChunkedDataPipe(
  21. std::move(data_remote),
  22. network::ResourceRequestBody::ReadOnlyOnce(false));
  23. simple_url_loader_ = network::SimpleURLLoader::Create(
  24. std::move(resource_request), upstream_traffic_annotation);
  25. simple_url_loader_->DownloadToStringOfUnboundedSizeUntilCrashAndDie(
  26. url_loader_factory,
  27. base::BindOnce(&UpstreamLoader::OnComplete, base::Unretained(this)));
  28. }
  29. UpstreamLoader::~UpstreamLoader() = default;
  30. // Attempts to send more of the upload body, if more data is available, and
  31. // |upload_pipe_| is valid.
  32. void UpstreamLoader::SendData() {
  33. DCHECK_LE(upload_position_, upload_body_.size());
  34. if (!upload_pipe_.is_valid())
  35. return;
  36. // Nothing more to write yet, or done writing everything.
  37. if (upload_position_ == upload_body_.size())
  38. return;
  39. // Since kMaxUploadWrite is a uint32_t, no overflow occurs in this downcast.
  40. uint32_t write_bytes = std::min(upload_body_.length() - upload_position_,
  41. static_cast<size_t>(kMaxUploadWrite));
  42. MojoResult result =
  43. upload_pipe_->WriteData(upload_body_.data() + upload_position_,
  44. &write_bytes, MOJO_WRITE_DATA_FLAG_NONE);
  45. // Wait for the pipe to have more capacity available, if needed.
  46. if (result == MOJO_RESULT_SHOULD_WAIT) {
  47. upload_pipe_watcher_->ArmOrNotify();
  48. return;
  49. }
  50. // Do nothing on pipe closure - depend on the SimpleURLLoader to notice the
  51. // other pipes being closed on error. Can reach this point if there's a
  52. // retry, for instance, so cannot draw any conclusions here.
  53. if (result != MOJO_RESULT_OK)
  54. return;
  55. upload_position_ += write_bytes;
  56. // If more data is available, arm the watcher again. Don't write again in a
  57. // loop, even if WriteData would allow it, to avoid blocking the current
  58. // thread.
  59. if (upload_position_ < upload_body_.size())
  60. upload_pipe_watcher_->ArmOrNotify();
  61. }
  62. void UpstreamLoader::AppendChunkToUpload(const std::string& data,
  63. bool is_last_chunk) {
  64. DCHECK(!has_last_chunk_);
  65. upload_body_ += data;
  66. if (is_last_chunk) {
  67. // Send size before the rest of the body. While it doesn't matter much, if
  68. // the other side receives the size before the last chunk, which Mojo does
  69. // not guarantee, some protocols can merge the data and the last chunk
  70. // itself into a single frame.
  71. has_last_chunk_ = is_last_chunk;
  72. if (get_size_callback_)
  73. std::move(get_size_callback_).Run(net::OK, upload_body_.size());
  74. }
  75. SendData();
  76. }
  77. void UpstreamLoader::OnUploadPipeWriteable(MojoResult unused) {
  78. SendData();
  79. }
  80. void UpstreamLoader::OnComplete(std::unique_ptr<std::string> response_body) {
  81. int response_code = -1;
  82. if (simple_url_loader_->ResponseInfo() &&
  83. simple_url_loader_->ResponseInfo()->headers) {
  84. response_code =
  85. simple_url_loader_->ResponseInfo()->headers->response_code();
  86. }
  87. upstream_loader_client_->OnUpstreamDataComplete(response_body != nullptr,
  88. response_code);
  89. }
  90. void UpstreamLoader::GetSize(GetSizeCallback get_size_callback) {
  91. if (has_last_chunk_) {
  92. std::move(get_size_callback).Run(net::OK, upload_body_.size());
  93. } else {
  94. get_size_callback_ = std::move(get_size_callback);
  95. }
  96. }
  97. void UpstreamLoader::StartReading(mojo::ScopedDataPipeProducerHandle pipe) {
  98. // Delete any existing pipe, if any.
  99. upload_pipe_watcher_.reset();
  100. upload_pipe_ = std::move(pipe);
  101. upload_pipe_watcher_ = std::make_unique<mojo::SimpleWatcher>(
  102. FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL);
  103. upload_pipe_watcher_->Watch(
  104. upload_pipe_.get(), MOJO_HANDLE_SIGNAL_WRITABLE,
  105. base::BindRepeating(&UpstreamLoader::OnUploadPipeWriteable,
  106. base::Unretained(this)));
  107. upload_position_ = 0;
  108. // Will attempt to start sending the request body, if any data is available.
  109. SendData();
  110. }
  111. } // namespace speech