123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232 |
- // Copyright 2018 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 "net/websockets/websocket_basic_stream_adapters.h"
- #include <algorithm>
- #include <cstring>
- #include <utility>
- #include "base/bind.h"
- #include "base/location.h"
- #include "base/task/single_thread_task_runner.h"
- #include "base/threading/thread_task_runner_handle.h"
- #include "net/base/io_buffer.h"
- #include "net/socket/client_socket_handle.h"
- #include "net/socket/socket.h"
- #include "net/spdy/spdy_buffer.h"
- namespace net {
- WebSocketClientSocketHandleAdapter::WebSocketClientSocketHandleAdapter(
- std::unique_ptr<ClientSocketHandle> connection)
- : connection_(std::move(connection)) {}
- WebSocketClientSocketHandleAdapter::~WebSocketClientSocketHandleAdapter() =
- default;
- int WebSocketClientSocketHandleAdapter::Read(IOBuffer* buf,
- int buf_len,
- CompletionOnceCallback callback) {
- return connection_->socket()->Read(buf, buf_len, std::move(callback));
- }
- int WebSocketClientSocketHandleAdapter::Write(
- IOBuffer* buf,
- int buf_len,
- CompletionOnceCallback callback,
- const NetworkTrafficAnnotationTag& traffic_annotation) {
- return connection_->socket()->Write(buf, buf_len, std::move(callback),
- traffic_annotation);
- }
- void WebSocketClientSocketHandleAdapter::Disconnect() {
- connection_->socket()->Disconnect();
- }
- bool WebSocketClientSocketHandleAdapter::is_initialized() const {
- return connection_->is_initialized();
- }
- WebSocketSpdyStreamAdapter::WebSocketSpdyStreamAdapter(
- base::WeakPtr<SpdyStream> stream,
- Delegate* delegate,
- NetLogWithSource net_log)
- : stream_(stream), delegate_(delegate), net_log_(net_log) {
- stream_->SetDelegate(this);
- }
- WebSocketSpdyStreamAdapter::~WebSocketSpdyStreamAdapter() {
- if (stream_) {
- // DetachDelegate() also cancels the stream.
- stream_->DetachDelegate();
- }
- }
- void WebSocketSpdyStreamAdapter::DetachDelegate() {
- delegate_ = nullptr;
- }
- int WebSocketSpdyStreamAdapter::Read(IOBuffer* buf,
- int buf_len,
- CompletionOnceCallback callback) {
- DCHECK(!read_callback_);
- DCHECK_LT(0, buf_len);
- read_buffer_ = buf;
- // |read_length_| is size_t and |buf_len| is a non-negative int, therefore
- // conversion is always valid.
- read_length_ = buf_len;
- if (!read_data_.IsEmpty())
- return CopySavedReadDataIntoBuffer();
- if (!stream_)
- return stream_error_;
- read_callback_ = std::move(callback);
- return ERR_IO_PENDING;
- }
- int WebSocketSpdyStreamAdapter::Write(
- IOBuffer* buf,
- int buf_len,
- CompletionOnceCallback callback,
- const NetworkTrafficAnnotationTag& traffic_annotation) {
- CHECK(headers_sent_);
- DCHECK(!write_callback_);
- DCHECK(callback);
- DCHECK_LT(0, buf_len);
- if (!stream_)
- return stream_error_;
- stream_->SendData(buf, buf_len, MORE_DATA_TO_SEND);
- write_callback_ = std::move(callback);
- write_length_ = buf_len;
- return ERR_IO_PENDING;
- }
- void WebSocketSpdyStreamAdapter::Disconnect() {
- if (stream_) {
- stream_->DetachDelegate();
- stream_ = nullptr;
- }
- }
- bool WebSocketSpdyStreamAdapter::is_initialized() const {
- return true;
- }
- // SpdyStream::Delegate methods.
- void WebSocketSpdyStreamAdapter::OnHeadersSent() {
- headers_sent_ = true;
- if (delegate_)
- delegate_->OnHeadersSent();
- }
- void WebSocketSpdyStreamAdapter::OnEarlyHintsReceived(
- const spdy::Http2HeaderBlock& headers) {
- // This callback should not be called for a WebSocket handshake.
- NOTREACHED();
- }
- void WebSocketSpdyStreamAdapter::OnHeadersReceived(
- const spdy::Http2HeaderBlock& response_headers,
- const spdy::Http2HeaderBlock* pushed_request_headers) {
- if (delegate_)
- delegate_->OnHeadersReceived(response_headers);
- }
- void WebSocketSpdyStreamAdapter::OnDataReceived(
- std::unique_ptr<SpdyBuffer> buffer) {
- if (!buffer) {
- // This is slightly wrong semantically, as it's still possible to write to
- // the stream at this point. However, if the server closes the stream
- // without waiting for a close frame from us, that means it is not
- // interested in a clean shutdown. In which case we don't need to worry
- // about sending any remaining data we might have buffered. This results in
- // a call to OnClose() which then informs our delegate.
- stream_->Close();
- return;
- }
- read_data_.Enqueue(std::move(buffer));
- if (read_callback_)
- std::move(read_callback_).Run(CopySavedReadDataIntoBuffer());
- }
- void WebSocketSpdyStreamAdapter::OnDataSent() {
- DCHECK(write_callback_);
- std::move(write_callback_).Run(write_length_);
- }
- void WebSocketSpdyStreamAdapter::OnTrailers(
- const spdy::Http2HeaderBlock& trailers) {}
- void WebSocketSpdyStreamAdapter::OnClose(int status) {
- DCHECK_NE(ERR_IO_PENDING, status);
- DCHECK_LE(status, 0);
- if (status == OK) {
- status = ERR_CONNECTION_CLOSED;
- }
- stream_error_ = status;
- stream_ = nullptr;
- auto self = weak_factory_.GetWeakPtr();
- if (read_callback_) {
- DCHECK(read_data_.IsEmpty());
- // Might destroy |this|.
- std::move(read_callback_).Run(status);
- if (!self)
- return;
- }
- if (write_callback_) {
- // Might destroy |this|.
- std::move(write_callback_).Run(status);
- if (!self)
- return;
- }
- // Delay calling delegate_->OnClose() until all buffered data are read.
- if (read_data_.IsEmpty() && delegate_) {
- // Might destroy |this|.
- delegate_->OnClose(status);
- }
- }
- bool WebSocketSpdyStreamAdapter::CanGreaseFrameType() const {
- return false;
- }
- NetLogSource WebSocketSpdyStreamAdapter::source_dependency() const {
- return net_log_.source();
- }
- int WebSocketSpdyStreamAdapter::CopySavedReadDataIntoBuffer() {
- int rv = read_data_.Dequeue(read_buffer_->data(), read_length_);
- // Stream has been destroyed earlier but delegate_->OnClose() call was
- // delayed until all buffered data are read. PostTask so that Read() can
- // return beforehand.
- if (!stream_ && delegate_ && read_data_.IsEmpty()) {
- base::ThreadTaskRunnerHandle::Get()->PostTask(
- FROM_HERE,
- base::BindOnce(&WebSocketSpdyStreamAdapter::CallDelegateOnClose,
- weak_factory_.GetWeakPtr()));
- }
- return rv;
- }
- void WebSocketSpdyStreamAdapter::CallDelegateOnClose() {
- if (delegate_)
- delegate_->OnClose(stream_error_);
- }
- } // namespace net
|