| // Copyright (c) 2012 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_job.h" |
| |
| #include <algorithm> |
| |
| #include "base/bind.h" |
| #include "base/lazy_instance.h" |
| #include "net/base/io_buffer.h" |
| #include "net/base/net_errors.h" |
| #include "net/base/net_log.h" |
| #include "net/cookies/cookie_store.h" |
| #include "net/http/http_network_session.h" |
| #include "net/http/http_transaction_factory.h" |
| #include "net/http/http_util.h" |
| #include "net/spdy/spdy_session.h" |
| #include "net/spdy/spdy_session_pool.h" |
| #include "net/url_request/url_request_context.h" |
| #include "net/websockets/websocket_handshake_handler.h" |
| #include "net/websockets/websocket_net_log_params.h" |
| #include "net/websockets/websocket_throttle.h" |
| #include "url/gurl.h" |
| |
| static const int kMaxPendingSendAllowed = 32768; // 32 kilobytes. |
| |
| namespace { |
| |
| // lower-case header names. |
| const char* const kCookieHeaders[] = { |
| "cookie", "cookie2" |
| }; |
| const char* const kSetCookieHeaders[] = { |
| "set-cookie", "set-cookie2" |
| }; |
| |
| net::SocketStreamJob* WebSocketJobFactory( |
| const GURL& url, net::SocketStream::Delegate* delegate, |
| net::URLRequestContext* context, net::CookieStore* cookie_store) { |
| net::WebSocketJob* job = new net::WebSocketJob(delegate); |
| job->InitSocketStream(new net::SocketStream(url, job, context, cookie_store)); |
| return job; |
| } |
| |
| class WebSocketJobInitSingleton { |
| private: |
| friend struct base::DefaultLazyInstanceTraits<WebSocketJobInitSingleton>; |
| WebSocketJobInitSingleton() { |
| net::SocketStreamJob::RegisterProtocolFactory("ws", WebSocketJobFactory); |
| net::SocketStreamJob::RegisterProtocolFactory("wss", WebSocketJobFactory); |
| } |
| }; |
| |
| static base::LazyInstance<WebSocketJobInitSingleton> g_websocket_job_init = |
| LAZY_INSTANCE_INITIALIZER; |
| |
| } // anonymous namespace |
| |
| namespace net { |
| |
| // static |
| void WebSocketJob::EnsureInit() { |
| g_websocket_job_init.Get(); |
| } |
| |
| WebSocketJob::WebSocketJob(SocketStream::Delegate* delegate) |
| : delegate_(delegate), |
| state_(INITIALIZED), |
| waiting_(false), |
| handshake_request_(new WebSocketHandshakeRequestHandler), |
| handshake_response_(new WebSocketHandshakeResponseHandler), |
| started_to_send_handshake_request_(false), |
| handshake_request_sent_(0), |
| response_cookies_save_index_(0), |
| spdy_protocol_version_(0), |
| save_next_cookie_running_(false), |
| callback_pending_(false), |
| weak_ptr_factory_(this), |
| weak_ptr_factory_for_send_pending_(this) { |
| } |
| |
| WebSocketJob::~WebSocketJob() { |
| DCHECK_EQ(CLOSED, state_); |
| DCHECK(!delegate_); |
| DCHECK(!socket_.get()); |
| } |
| |
| void WebSocketJob::Connect() { |
| DCHECK(socket_.get()); |
| DCHECK_EQ(state_, INITIALIZED); |
| state_ = CONNECTING; |
| socket_->Connect(); |
| } |
| |
| bool WebSocketJob::SendData(const char* data, int len) { |
| switch (state_) { |
| case INITIALIZED: |
| return false; |
| |
| case CONNECTING: |
| return SendHandshakeRequest(data, len); |
| |
| case OPEN: |
| { |
| scoped_refptr<IOBufferWithSize> buffer = new IOBufferWithSize(len); |
| memcpy(buffer->data(), data, len); |
| if (current_send_buffer_.get() || !send_buffer_queue_.empty()) { |
| send_buffer_queue_.push_back(buffer); |
| return true; |
| } |
| current_send_buffer_ = new DrainableIOBuffer(buffer.get(), len); |
| return SendDataInternal(current_send_buffer_->data(), |
| current_send_buffer_->BytesRemaining()); |
| } |
| |
| case CLOSING: |
| case CLOSED: |
| return false; |
| } |
| return false; |
| } |
| |
| void WebSocketJob::Close() { |
| if (state_ == CLOSED) |
| return; |
| |
| state_ = CLOSING; |
| if (current_send_buffer_.get()) { |
| // Will close in SendPending. |
| return; |
| } |
| state_ = CLOSED; |
| CloseInternal(); |
| } |
| |
| void WebSocketJob::RestartWithAuth(const AuthCredentials& credentials) { |
| state_ = CONNECTING; |
| socket_->RestartWithAuth(credentials); |
| } |
| |
| void WebSocketJob::DetachDelegate() { |
| state_ = CLOSED; |
| WebSocketThrottle::GetInstance()->RemoveFromQueue(this); |
| |
| scoped_refptr<WebSocketJob> protect(this); |
| weak_ptr_factory_.InvalidateWeakPtrs(); |
| weak_ptr_factory_for_send_pending_.InvalidateWeakPtrs(); |
| |
| delegate_ = NULL; |
| if (socket_.get()) |
| socket_->DetachDelegate(); |
| socket_ = NULL; |
| if (!callback_.is_null()) { |
| waiting_ = false; |
| callback_.Reset(); |
| Release(); // Balanced with OnStartOpenConnection(). |
| } |
| } |
| |
| int WebSocketJob::OnStartOpenConnection( |
| SocketStream* socket, const CompletionCallback& callback) { |
| DCHECK(callback_.is_null()); |
| state_ = CONNECTING; |
| |
| addresses_ = socket->address_list(); |
| if (!WebSocketThrottle::GetInstance()->PutInQueue(this)) { |
| return ERR_WS_THROTTLE_QUEUE_TOO_LARGE; |
| } |
| |
| if (delegate_) { |
| int result = delegate_->OnStartOpenConnection(socket, callback); |
| DCHECK_EQ(OK, result); |
| } |
| if (waiting_) { |
| // PutInQueue() may set |waiting_| true for throttling. In this case, |
| // Wakeup() will be called later. |
| callback_ = callback; |
| AddRef(); // Balanced when callback_ is cleared. |
| return ERR_IO_PENDING; |
| } |
| return TrySpdyStream(); |
| } |
| |
| void WebSocketJob::OnConnected( |
| SocketStream* socket, int max_pending_send_allowed) { |
| if (state_ == CLOSED) |
| return; |
| DCHECK_EQ(CONNECTING, state_); |
| if (delegate_) |
| delegate_->OnConnected(socket, max_pending_send_allowed); |
| } |
| |
| void WebSocketJob::OnSentData(SocketStream* socket, int amount_sent) { |
| DCHECK_NE(INITIALIZED, state_); |
| DCHECK_GT(amount_sent, 0); |
| if (state_ == CLOSED) |
| return; |
| if (state_ == CONNECTING) { |
| OnSentHandshakeRequest(socket, amount_sent); |
| return; |
| } |
| if (delegate_) { |
| DCHECK(state_ == OPEN || state_ == CLOSING); |
| if (!current_send_buffer_.get()) { |
| VLOG(1) |
| << "OnSentData current_send_buffer=NULL amount_sent=" << amount_sent; |
| return; |
| } |
| current_send_buffer_->DidConsume(amount_sent); |
| if (current_send_buffer_->BytesRemaining() > 0) |
| return; |
| |
| // We need to report amount_sent of original buffer size, instead of |
| // amount sent to |socket|. |
| amount_sent = current_send_buffer_->size(); |
| DCHECK_GT(amount_sent, 0); |
| current_send_buffer_ = NULL; |
| if (!weak_ptr_factory_for_send_pending_.HasWeakPtrs()) { |
| base::MessageLoopForIO::current()->PostTask( |
| FROM_HERE, |
| base::Bind(&WebSocketJob::SendPending, |
| weak_ptr_factory_for_send_pending_.GetWeakPtr())); |
| } |
| delegate_->OnSentData(socket, amount_sent); |
| } |
| } |
| |
| void WebSocketJob::OnReceivedData( |
| SocketStream* socket, const char* data, int len) { |
| DCHECK_NE(INITIALIZED, state_); |
| if (state_ == CLOSED) |
| return; |
| if (state_ == CONNECTING) { |
| OnReceivedHandshakeResponse(socket, data, len); |
| return; |
| } |
| DCHECK(state_ == OPEN || state_ == CLOSING); |
| if (delegate_ && len > 0) |
| delegate_->OnReceivedData(socket, data, len); |
| } |
| |
| void WebSocketJob::OnClose(SocketStream* socket) { |
| state_ = CLOSED; |
| WebSocketThrottle::GetInstance()->RemoveFromQueue(this); |
| |
| scoped_refptr<WebSocketJob> protect(this); |
| weak_ptr_factory_.InvalidateWeakPtrs(); |
| |
| SocketStream::Delegate* delegate = delegate_; |
| delegate_ = NULL; |
| socket_ = NULL; |
| if (!callback_.is_null()) { |
| waiting_ = false; |
| callback_.Reset(); |
| Release(); // Balanced with OnStartOpenConnection(). |
| } |
| if (delegate) |
| delegate->OnClose(socket); |
| } |
| |
| void WebSocketJob::OnAuthRequired( |
| SocketStream* socket, AuthChallengeInfo* auth_info) { |
| if (delegate_) |
| delegate_->OnAuthRequired(socket, auth_info); |
| } |
| |
| void WebSocketJob::OnSSLCertificateError( |
| SocketStream* socket, const SSLInfo& ssl_info, bool fatal) { |
| if (delegate_) |
| delegate_->OnSSLCertificateError(socket, ssl_info, fatal); |
| } |
| |
| void WebSocketJob::OnError(const SocketStream* socket, int error) { |
| if (delegate_ && error != ERR_PROTOCOL_SWITCHED) |
| delegate_->OnError(socket, error); |
| } |
| |
| void WebSocketJob::OnCreatedSpdyStream(int result) { |
| DCHECK(spdy_websocket_stream_.get()); |
| DCHECK(socket_.get()); |
| DCHECK_NE(ERR_IO_PENDING, result); |
| |
| if (state_ == CLOSED) { |
| result = ERR_ABORTED; |
| } else if (result == OK) { |
| state_ = CONNECTING; |
| result = ERR_PROTOCOL_SWITCHED; |
| } else { |
| spdy_websocket_stream_.reset(); |
| } |
| |
| CompleteIO(result); |
| } |
| |
| void WebSocketJob::OnSentSpdyHeaders() { |
| DCHECK_NE(INITIALIZED, state_); |
| if (state_ != CONNECTING) |
| return; |
| size_t original_length = handshake_request_->original_length(); |
| handshake_request_.reset(); |
| if (delegate_) |
| delegate_->OnSentData(socket_.get(), original_length); |
| } |
| |
| void WebSocketJob::OnSpdyResponseHeadersUpdated( |
| const SpdyHeaderBlock& response_headers) { |
| DCHECK_NE(INITIALIZED, state_); |
| if (state_ != CONNECTING) |
| return; |
| // TODO(toyoshim): Fallback to non-spdy connection? |
| handshake_response_->ParseResponseHeaderBlock(response_headers, |
| challenge_, |
| spdy_protocol_version_); |
| |
| SaveCookiesAndNotifyHeadersComplete(); |
| } |
| |
| void WebSocketJob::OnSentSpdyData(size_t bytes_sent) { |
| DCHECK_NE(INITIALIZED, state_); |
| DCHECK_NE(CONNECTING, state_); |
| if (state_ == CLOSED) |
| return; |
| if (!spdy_websocket_stream_.get()) |
| return; |
| OnSentData(socket_.get(), static_cast<int>(bytes_sent)); |
| } |
| |
| void WebSocketJob::OnReceivedSpdyData(scoped_ptr<SpdyBuffer> buffer) { |
| DCHECK_NE(INITIALIZED, state_); |
| DCHECK_NE(CONNECTING, state_); |
| if (state_ == CLOSED) |
| return; |
| if (!spdy_websocket_stream_.get()) |
| return; |
| if (buffer) { |
| OnReceivedData( |
| socket_.get(), buffer->GetRemainingData(), buffer->GetRemainingSize()); |
| } else { |
| OnReceivedData(socket_.get(), NULL, 0); |
| } |
| } |
| |
| void WebSocketJob::OnCloseSpdyStream() { |
| spdy_websocket_stream_.reset(); |
| OnClose(socket_.get()); |
| } |
| |
| bool WebSocketJob::SendHandshakeRequest(const char* data, int len) { |
| DCHECK_EQ(state_, CONNECTING); |
| if (started_to_send_handshake_request_) |
| return false; |
| if (!handshake_request_->ParseRequest(data, len)) |
| return false; |
| |
| AddCookieHeaderAndSend(); |
| return true; |
| } |
| |
| void WebSocketJob::AddCookieHeaderAndSend() { |
| bool allow = true; |
| if (delegate_ && !delegate_->CanGetCookies(socket_.get(), GetURLForCookies())) |
| allow = false; |
| |
| if (socket_.get() && delegate_ && state_ == CONNECTING) { |
| handshake_request_->RemoveHeaders(kCookieHeaders, |
| arraysize(kCookieHeaders)); |
| if (allow && socket_->cookie_store()) { |
| // Add cookies, including HttpOnly cookies. |
| CookieOptions cookie_options; |
| cookie_options.set_include_httponly(); |
| socket_->cookie_store()->GetCookiesWithOptionsAsync( |
| GetURLForCookies(), cookie_options, |
| base::Bind(&WebSocketJob::LoadCookieCallback, |
| weak_ptr_factory_.GetWeakPtr())); |
| } else { |
| DoSendData(); |
| } |
| } |
| } |
| |
| void WebSocketJob::LoadCookieCallback(const std::string& cookie) { |
| if (!cookie.empty()) |
| // TODO(tyoshino): Sending cookie means that connection doesn't need |
| // PRIVACY_MODE_ENABLED as cookies may be server-bound and channel id |
| // wouldn't negatively affect privacy anyway. Need to restart connection |
| // or refactor to determine cookie status prior to connecting. |
| handshake_request_->AppendHeaderIfMissing("Cookie", cookie); |
| DoSendData(); |
| } |
| |
| void WebSocketJob::DoSendData() { |
| if (spdy_websocket_stream_.get()) { |
| scoped_ptr<SpdyHeaderBlock> headers(new SpdyHeaderBlock); |
| handshake_request_->GetRequestHeaderBlock( |
| socket_->url(), headers.get(), &challenge_, spdy_protocol_version_); |
| spdy_websocket_stream_->SendRequest(headers.Pass()); |
| } else { |
| const std::string& handshake_request = |
| handshake_request_->GetRawRequest(); |
| handshake_request_sent_ = 0; |
| socket_->net_log()->AddEvent( |
| NetLog::TYPE_WEB_SOCKET_SEND_REQUEST_HEADERS, |
| base::Bind(&NetLogWebSocketHandshakeCallback, &handshake_request)); |
| socket_->SendData(handshake_request.data(), |
| handshake_request.size()); |
| } |
| // Just buffered in |handshake_request_|. |
| started_to_send_handshake_request_ = true; |
| } |
| |
| void WebSocketJob::OnSentHandshakeRequest( |
| SocketStream* socket, int amount_sent) { |
| DCHECK_EQ(state_, CONNECTING); |
| handshake_request_sent_ += amount_sent; |
| DCHECK_LE(handshake_request_sent_, handshake_request_->raw_length()); |
| if (handshake_request_sent_ >= handshake_request_->raw_length()) { |
| // handshake request has been sent. |
| // notify original size of handshake request to delegate. |
| // Reset the handshake_request_ first in case this object is deleted by the |
| // delegate. |
| size_t original_length = handshake_request_->original_length(); |
| handshake_request_.reset(); |
| if (delegate_) |
| delegate_->OnSentData(socket, original_length); |
| } |
| } |
| |
| void WebSocketJob::OnReceivedHandshakeResponse( |
| SocketStream* socket, const char* data, int len) { |
| DCHECK_EQ(state_, CONNECTING); |
| if (handshake_response_->HasResponse()) { |
| // If we already has handshake response, received data should be frame |
| // data, not handshake message. |
| received_data_after_handshake_.insert( |
| received_data_after_handshake_.end(), data, data + len); |
| return; |
| } |
| |
| size_t response_length = handshake_response_->ParseRawResponse(data, len); |
| if (!handshake_response_->HasResponse()) { |
| // not yet. we need more data. |
| return; |
| } |
| // handshake message is completed. |
| std::string raw_response = handshake_response_->GetRawResponse(); |
| socket_->net_log()->AddEvent( |
| NetLog::TYPE_WEB_SOCKET_READ_RESPONSE_HEADERS, |
| base::Bind(&NetLogWebSocketHandshakeCallback, &raw_response)); |
| if (len - response_length > 0) { |
| // If we received extra data, it should be frame data. |
| DCHECK(received_data_after_handshake_.empty()); |
| received_data_after_handshake_.assign(data + response_length, data + len); |
| } |
| SaveCookiesAndNotifyHeadersComplete(); |
| } |
| |
| void WebSocketJob::SaveCookiesAndNotifyHeadersComplete() { |
| // handshake message is completed. |
| DCHECK(handshake_response_->HasResponse()); |
| |
| // Extract cookies from the handshake response into a temporary vector. |
| response_cookies_.clear(); |
| response_cookies_save_index_ = 0; |
| |
| handshake_response_->GetHeaders( |
| kSetCookieHeaders, arraysize(kSetCookieHeaders), &response_cookies_); |
| |
| // Now, loop over the response cookies, and attempt to persist each. |
| SaveNextCookie(); |
| } |
| |
| void WebSocketJob::NotifyHeadersComplete() { |
| // Remove cookie headers, with malformed headers preserved. |
| // Actual handshake should be done in Blink. |
| handshake_response_->RemoveHeaders( |
| kSetCookieHeaders, arraysize(kSetCookieHeaders)); |
| std::string handshake_response = handshake_response_->GetResponse(); |
| handshake_response_.reset(); |
| std::vector<char> received_data(handshake_response.begin(), |
| handshake_response.end()); |
| received_data.insert(received_data.end(), |
| received_data_after_handshake_.begin(), |
| received_data_after_handshake_.end()); |
| received_data_after_handshake_.clear(); |
| |
| state_ = OPEN; |
| |
| DCHECK(!received_data.empty()); |
| if (delegate_) |
| delegate_->OnReceivedData( |
| socket_.get(), &received_data.front(), received_data.size()); |
| |
| WebSocketThrottle::GetInstance()->RemoveFromQueue(this); |
| } |
| |
| void WebSocketJob::SaveNextCookie() { |
| if (!socket_.get() || !delegate_ || state_ != CONNECTING) |
| return; |
| |
| callback_pending_ = false; |
| save_next_cookie_running_ = true; |
| |
| if (socket_->cookie_store()) { |
| GURL url_for_cookies = GetURLForCookies(); |
| |
| CookieOptions options; |
| options.set_include_httponly(); |
| |
| // Loop as long as SetCookieWithOptionsAsync completes synchronously. Since |
| // CookieMonster's asynchronous operation APIs queue the callback to run it |
| // on the thread where the API was called, there won't be race. I.e. unless |
| // the callback is run synchronously, it won't be run in parallel with this |
| // method. |
| while (!callback_pending_ && |
| response_cookies_save_index_ < response_cookies_.size()) { |
| std::string cookie = response_cookies_[response_cookies_save_index_]; |
| response_cookies_save_index_++; |
| |
| if (!delegate_->CanSetCookie( |
| socket_.get(), url_for_cookies, cookie, &options)) |
| continue; |
| |
| callback_pending_ = true; |
| socket_->cookie_store()->SetCookieWithOptionsAsync( |
| url_for_cookies, cookie, options, |
| base::Bind(&WebSocketJob::OnCookieSaved, |
| weak_ptr_factory_.GetWeakPtr())); |
| } |
| } |
| |
| save_next_cookie_running_ = false; |
| |
| if (callback_pending_) |
| return; |
| |
| response_cookies_.clear(); |
| response_cookies_save_index_ = 0; |
| |
| NotifyHeadersComplete(); |
| } |
| |
| void WebSocketJob::OnCookieSaved(bool cookie_status) { |
| // Tell the caller of SetCookieWithOptionsAsync() that this completion |
| // callback is invoked. |
| // - If the caller checks callback_pending earlier than this callback, the |
| // caller exits to let this method continue iteration. |
| // - Otherwise, the caller continues iteration. |
| callback_pending_ = false; |
| |
| // Resume SaveNextCookie if the caller of SetCookieWithOptionsAsync() exited |
| // the loop. Otherwise, return. |
| if (save_next_cookie_running_) |
| return; |
| |
| SaveNextCookie(); |
| } |
| |
| GURL WebSocketJob::GetURLForCookies() const { |
| GURL url = socket_->url(); |
| std::string scheme = socket_->is_secure() ? "https" : "http"; |
| url::Replacements<char> replacements; |
| replacements.SetScheme(scheme.c_str(), url::Component(0, scheme.length())); |
| return url.ReplaceComponents(replacements); |
| } |
| |
| const AddressList& WebSocketJob::address_list() const { |
| return addresses_; |
| } |
| |
| int WebSocketJob::TrySpdyStream() { |
| if (!socket_.get()) |
| return ERR_FAILED; |
| |
| // Check if we have a SPDY session available. |
| HttpTransactionFactory* factory = |
| socket_->context()->http_transaction_factory(); |
| if (!factory) |
| return OK; |
| scoped_refptr<HttpNetworkSession> session = factory->GetSession(); |
| if (!session.get() || !session->params().enable_websocket_over_spdy) |
| return OK; |
| SpdySessionPool* spdy_pool = session->spdy_session_pool(); |
| PrivacyMode privacy_mode = socket_->privacy_mode(); |
| const SpdySessionKey key(HostPortPair::FromURL(socket_->url()), |
| socket_->proxy_server(), privacy_mode); |
| // Forbid wss downgrade to SPDY without SSL. |
| // TODO(toyoshim): Does it realize the same policy with HTTP? |
| base::WeakPtr<SpdySession> spdy_session = |
| spdy_pool->FindAvailableSession(key, *socket_->net_log()); |
| if (!spdy_session) |
| return OK; |
| |
| SSLInfo ssl_info; |
| bool was_npn_negotiated; |
| NextProto protocol_negotiated = kProtoUnknown; |
| bool use_ssl = spdy_session->GetSSLInfo( |
| &ssl_info, &was_npn_negotiated, &protocol_negotiated); |
| if (socket_->is_secure() && !use_ssl) |
| return OK; |
| |
| // Create SpdyWebSocketStream. |
| spdy_protocol_version_ = spdy_session->GetProtocolVersion(); |
| spdy_websocket_stream_.reset(new SpdyWebSocketStream(spdy_session, this)); |
| |
| int result = spdy_websocket_stream_->InitializeStream( |
| socket_->url(), MEDIUM, *socket_->net_log()); |
| if (result == OK) { |
| OnConnected(socket_.get(), kMaxPendingSendAllowed); |
| return ERR_PROTOCOL_SWITCHED; |
| } |
| if (result != ERR_IO_PENDING) { |
| spdy_websocket_stream_.reset(); |
| return OK; |
| } |
| |
| return ERR_IO_PENDING; |
| } |
| |
| void WebSocketJob::SetWaiting() { |
| waiting_ = true; |
| } |
| |
| bool WebSocketJob::IsWaiting() const { |
| return waiting_; |
| } |
| |
| void WebSocketJob::Wakeup() { |
| if (!waiting_) |
| return; |
| waiting_ = false; |
| DCHECK(!callback_.is_null()); |
| base::MessageLoopForIO::current()->PostTask( |
| FROM_HERE, |
| base::Bind(&WebSocketJob::RetryPendingIO, |
| weak_ptr_factory_.GetWeakPtr())); |
| } |
| |
| void WebSocketJob::RetryPendingIO() { |
| int result = TrySpdyStream(); |
| |
| // In the case of ERR_IO_PENDING, CompleteIO() will be called from |
| // OnCreatedSpdyStream(). |
| if (result != ERR_IO_PENDING) |
| CompleteIO(result); |
| } |
| |
| void WebSocketJob::CompleteIO(int result) { |
| // |callback_| may be null if OnClose() or DetachDelegate() was called. |
| if (!callback_.is_null()) { |
| CompletionCallback callback = callback_; |
| callback_.Reset(); |
| callback.Run(result); |
| Release(); // Balanced with OnStartOpenConnection(). |
| } |
| } |
| |
| bool WebSocketJob::SendDataInternal(const char* data, int length) { |
| if (spdy_websocket_stream_.get()) |
| return ERR_IO_PENDING == spdy_websocket_stream_->SendData(data, length); |
| if (socket_.get()) |
| return socket_->SendData(data, length); |
| return false; |
| } |
| |
| void WebSocketJob::CloseInternal() { |
| if (spdy_websocket_stream_.get()) |
| spdy_websocket_stream_->Close(); |
| if (socket_.get()) |
| socket_->Close(); |
| } |
| |
| void WebSocketJob::SendPending() { |
| if (current_send_buffer_.get()) |
| return; |
| |
| // Current buffer has been sent. Try next if any. |
| if (send_buffer_queue_.empty()) { |
| // No more data to send. |
| if (state_ == CLOSING) |
| CloseInternal(); |
| return; |
| } |
| |
| scoped_refptr<IOBufferWithSize> next_buffer = send_buffer_queue_.front(); |
| send_buffer_queue_.pop_front(); |
| current_send_buffer_ = |
| new DrainableIOBuffer(next_buffer.get(), next_buffer->size()); |
| SendDataInternal(current_send_buffer_->data(), |
| current_send_buffer_->BytesRemaining()); |
| } |
| |
| } // namespace net |