websocket_basic_stream_adapters.cc 6.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232
  1. // Copyright 2018 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 "net/websockets/websocket_basic_stream_adapters.h"
  5. #include <algorithm>
  6. #include <cstring>
  7. #include <utility>
  8. #include "base/bind.h"
  9. #include "base/location.h"
  10. #include "base/task/single_thread_task_runner.h"
  11. #include "base/threading/thread_task_runner_handle.h"
  12. #include "net/base/io_buffer.h"
  13. #include "net/socket/client_socket_handle.h"
  14. #include "net/socket/socket.h"
  15. #include "net/spdy/spdy_buffer.h"
  16. namespace net {
  17. WebSocketClientSocketHandleAdapter::WebSocketClientSocketHandleAdapter(
  18. std::unique_ptr<ClientSocketHandle> connection)
  19. : connection_(std::move(connection)) {}
  20. WebSocketClientSocketHandleAdapter::~WebSocketClientSocketHandleAdapter() =
  21. default;
  22. int WebSocketClientSocketHandleAdapter::Read(IOBuffer* buf,
  23. int buf_len,
  24. CompletionOnceCallback callback) {
  25. return connection_->socket()->Read(buf, buf_len, std::move(callback));
  26. }
  27. int WebSocketClientSocketHandleAdapter::Write(
  28. IOBuffer* buf,
  29. int buf_len,
  30. CompletionOnceCallback callback,
  31. const NetworkTrafficAnnotationTag& traffic_annotation) {
  32. return connection_->socket()->Write(buf, buf_len, std::move(callback),
  33. traffic_annotation);
  34. }
  35. void WebSocketClientSocketHandleAdapter::Disconnect() {
  36. connection_->socket()->Disconnect();
  37. }
  38. bool WebSocketClientSocketHandleAdapter::is_initialized() const {
  39. return connection_->is_initialized();
  40. }
  41. WebSocketSpdyStreamAdapter::WebSocketSpdyStreamAdapter(
  42. base::WeakPtr<SpdyStream> stream,
  43. Delegate* delegate,
  44. NetLogWithSource net_log)
  45. : stream_(stream), delegate_(delegate), net_log_(net_log) {
  46. stream_->SetDelegate(this);
  47. }
  48. WebSocketSpdyStreamAdapter::~WebSocketSpdyStreamAdapter() {
  49. if (stream_) {
  50. // DetachDelegate() also cancels the stream.
  51. stream_->DetachDelegate();
  52. }
  53. }
  54. void WebSocketSpdyStreamAdapter::DetachDelegate() {
  55. delegate_ = nullptr;
  56. }
  57. int WebSocketSpdyStreamAdapter::Read(IOBuffer* buf,
  58. int buf_len,
  59. CompletionOnceCallback callback) {
  60. DCHECK(!read_callback_);
  61. DCHECK_LT(0, buf_len);
  62. read_buffer_ = buf;
  63. // |read_length_| is size_t and |buf_len| is a non-negative int, therefore
  64. // conversion is always valid.
  65. read_length_ = buf_len;
  66. if (!read_data_.IsEmpty())
  67. return CopySavedReadDataIntoBuffer();
  68. if (!stream_)
  69. return stream_error_;
  70. read_callback_ = std::move(callback);
  71. return ERR_IO_PENDING;
  72. }
  73. int WebSocketSpdyStreamAdapter::Write(
  74. IOBuffer* buf,
  75. int buf_len,
  76. CompletionOnceCallback callback,
  77. const NetworkTrafficAnnotationTag& traffic_annotation) {
  78. CHECK(headers_sent_);
  79. DCHECK(!write_callback_);
  80. DCHECK(callback);
  81. DCHECK_LT(0, buf_len);
  82. if (!stream_)
  83. return stream_error_;
  84. stream_->SendData(buf, buf_len, MORE_DATA_TO_SEND);
  85. write_callback_ = std::move(callback);
  86. write_length_ = buf_len;
  87. return ERR_IO_PENDING;
  88. }
  89. void WebSocketSpdyStreamAdapter::Disconnect() {
  90. if (stream_) {
  91. stream_->DetachDelegate();
  92. stream_ = nullptr;
  93. }
  94. }
  95. bool WebSocketSpdyStreamAdapter::is_initialized() const {
  96. return true;
  97. }
  98. // SpdyStream::Delegate methods.
  99. void WebSocketSpdyStreamAdapter::OnHeadersSent() {
  100. headers_sent_ = true;
  101. if (delegate_)
  102. delegate_->OnHeadersSent();
  103. }
  104. void WebSocketSpdyStreamAdapter::OnEarlyHintsReceived(
  105. const spdy::Http2HeaderBlock& headers) {
  106. // This callback should not be called for a WebSocket handshake.
  107. NOTREACHED();
  108. }
  109. void WebSocketSpdyStreamAdapter::OnHeadersReceived(
  110. const spdy::Http2HeaderBlock& response_headers,
  111. const spdy::Http2HeaderBlock* pushed_request_headers) {
  112. if (delegate_)
  113. delegate_->OnHeadersReceived(response_headers);
  114. }
  115. void WebSocketSpdyStreamAdapter::OnDataReceived(
  116. std::unique_ptr<SpdyBuffer> buffer) {
  117. if (!buffer) {
  118. // This is slightly wrong semantically, as it's still possible to write to
  119. // the stream at this point. However, if the server closes the stream
  120. // without waiting for a close frame from us, that means it is not
  121. // interested in a clean shutdown. In which case we don't need to worry
  122. // about sending any remaining data we might have buffered. This results in
  123. // a call to OnClose() which then informs our delegate.
  124. stream_->Close();
  125. return;
  126. }
  127. read_data_.Enqueue(std::move(buffer));
  128. if (read_callback_)
  129. std::move(read_callback_).Run(CopySavedReadDataIntoBuffer());
  130. }
  131. void WebSocketSpdyStreamAdapter::OnDataSent() {
  132. DCHECK(write_callback_);
  133. std::move(write_callback_).Run(write_length_);
  134. }
  135. void WebSocketSpdyStreamAdapter::OnTrailers(
  136. const spdy::Http2HeaderBlock& trailers) {}
  137. void WebSocketSpdyStreamAdapter::OnClose(int status) {
  138. DCHECK_NE(ERR_IO_PENDING, status);
  139. DCHECK_LE(status, 0);
  140. if (status == OK) {
  141. status = ERR_CONNECTION_CLOSED;
  142. }
  143. stream_error_ = status;
  144. stream_ = nullptr;
  145. auto self = weak_factory_.GetWeakPtr();
  146. if (read_callback_) {
  147. DCHECK(read_data_.IsEmpty());
  148. // Might destroy |this|.
  149. std::move(read_callback_).Run(status);
  150. if (!self)
  151. return;
  152. }
  153. if (write_callback_) {
  154. // Might destroy |this|.
  155. std::move(write_callback_).Run(status);
  156. if (!self)
  157. return;
  158. }
  159. // Delay calling delegate_->OnClose() until all buffered data are read.
  160. if (read_data_.IsEmpty() && delegate_) {
  161. // Might destroy |this|.
  162. delegate_->OnClose(status);
  163. }
  164. }
  165. bool WebSocketSpdyStreamAdapter::CanGreaseFrameType() const {
  166. return false;
  167. }
  168. NetLogSource WebSocketSpdyStreamAdapter::source_dependency() const {
  169. return net_log_.source();
  170. }
  171. int WebSocketSpdyStreamAdapter::CopySavedReadDataIntoBuffer() {
  172. int rv = read_data_.Dequeue(read_buffer_->data(), read_length_);
  173. // Stream has been destroyed earlier but delegate_->OnClose() call was
  174. // delayed until all buffered data are read. PostTask so that Read() can
  175. // return beforehand.
  176. if (!stream_ && delegate_ && read_data_.IsEmpty()) {
  177. base::ThreadTaskRunnerHandle::Get()->PostTask(
  178. FROM_HERE,
  179. base::BindOnce(&WebSocketSpdyStreamAdapter::CallDelegateOnClose,
  180. weak_factory_.GetWeakPtr()));
  181. }
  182. return rv;
  183. }
  184. void WebSocketSpdyStreamAdapter::CallDelegateOnClose() {
  185. if (delegate_)
  186. delegate_->OnClose(stream_error_);
  187. }
  188. } // namespace net