Home | History | Annotate | Download | only in websockets 
      1 // Copyright (c) 2012 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 
      5 #include "net/websockets/websocket_job.h"
      6 
      7 #include <algorithm>
      8 
      9 #include "base/bind.h"
     10 #include "base/lazy_instance.h"
     11 #include "net/base/io_buffer.h"
     12 #include "net/base/net_errors.h"
     13 #include "net/base/net_log.h"
     14 #include "net/cookies/cookie_store.h"
     15 #include "net/http/http_network_session.h"
     16 #include "net/http/http_transaction_factory.h"
     17 #include "net/http/http_util.h"
     18 #include "net/spdy/spdy_session.h"
     19 #include "net/spdy/spdy_session_pool.h"
     20 #include "net/url_request/url_request_context.h"
     21 #include "net/websockets/websocket_handshake_handler.h"
     22 #include "net/websockets/websocket_net_log_params.h"
     23 #include "net/websockets/websocket_throttle.h"
     24 #include "url/gurl.h"
     25 
     26 static const int kMaxPendingSendAllowed = 32768;  // 32 kilobytes.
     27 
     28 namespace {
     29 
     30 // lower-case header names.
     31 const char* const kCookieHeaders[] = {
     32   "cookie", "cookie2"
     33 };
     34 const char* const kSetCookieHeaders[] = {
     35   "set-cookie", "set-cookie2"
     36 };
     37 
     38 net::SocketStreamJob* WebSocketJobFactory(
     39     const GURL& url, net::SocketStream::Delegate* delegate,
     40     net::URLRequestContext* context, net::CookieStore* cookie_store) {
     41   net::WebSocketJob* job = new net::WebSocketJob(delegate);
     42   job->InitSocketStream(new net::SocketStream(url, job, context, cookie_store));
     43   return job;
     44 }
     45 
     46 class WebSocketJobInitSingleton {
     47  private:
     48   friend struct base::DefaultLazyInstanceTraits<WebSocketJobInitSingleton>;
     49   WebSocketJobInitSingleton() {
     50     net::SocketStreamJob::RegisterProtocolFactory("ws", WebSocketJobFactory);
     51     net::SocketStreamJob::RegisterProtocolFactory("wss", WebSocketJobFactory);
     52   }
     53 };
     54 
     55 static base::LazyInstance<WebSocketJobInitSingleton> g_websocket_job_init =
     56     LAZY_INSTANCE_INITIALIZER;
     57 
     58 }  // anonymous namespace
     59 
     60 namespace net {
     61 
     62 // static
     63 void WebSocketJob::EnsureInit() {
     64   g_websocket_job_init.Get();
     65 }
     66 
     67 WebSocketJob::WebSocketJob(SocketStream::Delegate* delegate)
     68     : delegate_(delegate),
     69       state_(INITIALIZED),
     70       waiting_(false),
     71       handshake_request_(new WebSocketHandshakeRequestHandler),
     72       handshake_response_(new WebSocketHandshakeResponseHandler),
     73       started_to_send_handshake_request_(false),
     74       handshake_request_sent_(0),
     75       response_cookies_save_index_(0),
     76       spdy_protocol_version_(0),
     77       save_next_cookie_running_(false),
     78       callback_pending_(false),
     79       weak_ptr_factory_(this),
     80       weak_ptr_factory_for_send_pending_(this) {
     81 }
     82 
     83 WebSocketJob::~WebSocketJob() {
     84   DCHECK_EQ(CLOSED, state_);
     85   DCHECK(!delegate_);
     86   DCHECK(!socket_.get());
     87 }
     88 
     89 void WebSocketJob::Connect() {
     90   DCHECK(socket_.get());
     91   DCHECK_EQ(state_, INITIALIZED);
     92   state_ = CONNECTING;
     93   socket_->Connect();
     94 }
     95 
     96 bool WebSocketJob::SendData(const char* data, int len) {
     97   switch (state_) {
     98     case INITIALIZED:
     99       return false;
    100 
    101     case CONNECTING:
    102       return SendHandshakeRequest(data, len);
    103 
    104     case OPEN:
    105       {
    106         scoped_refptr<IOBufferWithSize> buffer = new IOBufferWithSize(len);
    107         memcpy(buffer->data(), data, len);
    108         if (current_send_buffer_.get() || !send_buffer_queue_.empty()) {
    109           send_buffer_queue_.push_back(buffer);
    110           return true;
    111         }
    112         current_send_buffer_ = new DrainableIOBuffer(buffer.get(), len);
    113         return SendDataInternal(current_send_buffer_->data(),
    114                                 current_send_buffer_->BytesRemaining());
    115       }
    116 
    117     case CLOSING:
    118     case CLOSED:
    119       return false;
    120   }
    121   return false;
    122 }
    123 
    124 void WebSocketJob::Close() {
    125   if (state_ == CLOSED)
    126     return;
    127 
    128   state_ = CLOSING;
    129   if (current_send_buffer_.get()) {
    130     // Will close in SendPending.
    131     return;
    132   }
    133   state_ = CLOSED;
    134   CloseInternal();
    135 }
    136 
    137 void WebSocketJob::RestartWithAuth(const AuthCredentials& credentials) {
    138   state_ = CONNECTING;
    139   socket_->RestartWithAuth(credentials);
    140 }
    141 
    142 void WebSocketJob::DetachDelegate() {
    143   state_ = CLOSED;
    144   WebSocketThrottle::GetInstance()->RemoveFromQueue(this);
    145 
    146   scoped_refptr<WebSocketJob> protect(this);
    147   weak_ptr_factory_.InvalidateWeakPtrs();
    148   weak_ptr_factory_for_send_pending_.InvalidateWeakPtrs();
    149 
    150   delegate_ = NULL;
    151   if (socket_.get())
    152     socket_->DetachDelegate();
    153   socket_ = NULL;
    154   if (!callback_.is_null()) {
    155     waiting_ = false;
    156     callback_.Reset();
    157     Release();  // Balanced with OnStartOpenConnection().
    158   }
    159 }
    160 
    161 int WebSocketJob::OnStartOpenConnection(
    162     SocketStream* socket, const CompletionCallback& callback) {
    163   DCHECK(callback_.is_null());
    164   state_ = CONNECTING;
    165 
    166   addresses_ = socket->address_list();
    167   if (!WebSocketThrottle::GetInstance()->PutInQueue(this)) {
    168     return ERR_WS_THROTTLE_QUEUE_TOO_LARGE;
    169   }
    170 
    171   if (delegate_) {
    172     int result = delegate_->OnStartOpenConnection(socket, callback);
    173     DCHECK_EQ(OK, result);
    174   }
    175   if (waiting_) {
    176     // PutInQueue() may set |waiting_| true for throttling. In this case,
    177     // Wakeup() will be called later.
    178     callback_ = callback;
    179     AddRef();  // Balanced when callback_ is cleared.
    180     return ERR_IO_PENDING;
    181   }
    182   return TrySpdyStream();
    183 }
    184 
    185 void WebSocketJob::OnConnected(
    186     SocketStream* socket, int max_pending_send_allowed) {
    187   if (state_ == CLOSED)
    188     return;
    189   DCHECK_EQ(CONNECTING, state_);
    190   if (delegate_)
    191     delegate_->OnConnected(socket, max_pending_send_allowed);
    192 }
    193 
    194 void WebSocketJob::OnSentData(SocketStream* socket, int amount_sent) {
    195   DCHECK_NE(INITIALIZED, state_);
    196   DCHECK_GT(amount_sent, 0);
    197   if (state_ == CLOSED)
    198     return;
    199   if (state_ == CONNECTING) {
    200     OnSentHandshakeRequest(socket, amount_sent);
    201     return;
    202   }
    203   if (delegate_) {
    204     DCHECK(state_ == OPEN || state_ == CLOSING);
    205     if (!current_send_buffer_.get()) {
    206       VLOG(1)
    207           << "OnSentData current_send_buffer=NULL amount_sent=" << amount_sent;
    208       return;
    209     }
    210     current_send_buffer_->DidConsume(amount_sent);
    211     if (current_send_buffer_->BytesRemaining() > 0)
    212       return;
    213 
    214     // We need to report amount_sent of original buffer size, instead of
    215     // amount sent to |socket|.
    216     amount_sent = current_send_buffer_->size();
    217     DCHECK_GT(amount_sent, 0);
    218     current_send_buffer_ = NULL;
    219     if (!weak_ptr_factory_for_send_pending_.HasWeakPtrs()) {
    220       base::MessageLoopForIO::current()->PostTask(
    221           FROM_HERE,
    222           base::Bind(&WebSocketJob::SendPending,
    223                      weak_ptr_factory_for_send_pending_.GetWeakPtr()));
    224     }
    225     delegate_->OnSentData(socket, amount_sent);
    226   }
    227 }
    228 
    229 void WebSocketJob::OnReceivedData(
    230     SocketStream* socket, const char* data, int len) {
    231   DCHECK_NE(INITIALIZED, state_);
    232   if (state_ == CLOSED)
    233     return;
    234   if (state_ == CONNECTING) {
    235     OnReceivedHandshakeResponse(socket, data, len);
    236     return;
    237   }
    238   DCHECK(state_ == OPEN || state_ == CLOSING);
    239   if (delegate_ && len > 0)
    240     delegate_->OnReceivedData(socket, data, len);
    241 }
    242 
    243 void WebSocketJob::OnClose(SocketStream* socket) {
    244   state_ = CLOSED;
    245   WebSocketThrottle::GetInstance()->RemoveFromQueue(this);
    246 
    247   scoped_refptr<WebSocketJob> protect(this);
    248   weak_ptr_factory_.InvalidateWeakPtrs();
    249 
    250   SocketStream::Delegate* delegate = delegate_;
    251   delegate_ = NULL;
    252   socket_ = NULL;
    253   if (!callback_.is_null()) {
    254     waiting_ = false;
    255     callback_.Reset();
    256     Release();  // Balanced with OnStartOpenConnection().
    257   }
    258   if (delegate)
    259     delegate->OnClose(socket);
    260 }
    261 
    262 void WebSocketJob::OnAuthRequired(
    263     SocketStream* socket, AuthChallengeInfo* auth_info) {
    264   if (delegate_)
    265     delegate_->OnAuthRequired(socket, auth_info);
    266 }
    267 
    268 void WebSocketJob::OnSSLCertificateError(
    269     SocketStream* socket, const SSLInfo& ssl_info, bool fatal) {
    270   if (delegate_)
    271     delegate_->OnSSLCertificateError(socket, ssl_info, fatal);
    272 }
    273 
    274 void WebSocketJob::OnError(const SocketStream* socket, int error) {
    275   if (delegate_ && error != ERR_PROTOCOL_SWITCHED)
    276     delegate_->OnError(socket, error);
    277 }
    278 
    279 void WebSocketJob::OnCreatedSpdyStream(int result) {
    280   DCHECK(spdy_websocket_stream_.get());
    281   DCHECK(socket_.get());
    282   DCHECK_NE(ERR_IO_PENDING, result);
    283 
    284   if (state_ == CLOSED) {
    285     result = ERR_ABORTED;
    286   } else if (result == OK) {
    287     state_ = CONNECTING;
    288     result = ERR_PROTOCOL_SWITCHED;
    289   } else {
    290     spdy_websocket_stream_.reset();
    291   }
    292 
    293   CompleteIO(result);
    294 }
    295 
    296 void WebSocketJob::OnSentSpdyHeaders() {
    297   DCHECK_NE(INITIALIZED, state_);
    298   if (state_ != CONNECTING)
    299     return;
    300   size_t original_length = handshake_request_->original_length();
    301   handshake_request_.reset();
    302   if (delegate_)
    303     delegate_->OnSentData(socket_.get(), original_length);
    304 }
    305 
    306 void WebSocketJob::OnSpdyResponseHeadersUpdated(
    307     const SpdyHeaderBlock& response_headers) {
    308   DCHECK_NE(INITIALIZED, state_);
    309   if (state_ != CONNECTING)
    310     return;
    311   // TODO(toyoshim): Fallback to non-spdy connection?
    312   handshake_response_->ParseResponseHeaderBlock(response_headers,
    313                                                 challenge_,
    314                                                 spdy_protocol_version_);
    315 
    316   SaveCookiesAndNotifyHeadersComplete();
    317 }
    318 
    319 void WebSocketJob::OnSentSpdyData(size_t bytes_sent) {
    320   DCHECK_NE(INITIALIZED, state_);
    321   DCHECK_NE(CONNECTING, state_);
    322   if (state_ == CLOSED)
    323     return;
    324   if (!spdy_websocket_stream_.get())
    325     return;
    326   OnSentData(socket_.get(), static_cast<int>(bytes_sent));
    327 }
    328 
    329 void WebSocketJob::OnReceivedSpdyData(scoped_ptr<SpdyBuffer> buffer) {
    330   DCHECK_NE(INITIALIZED, state_);
    331   DCHECK_NE(CONNECTING, state_);
    332   if (state_ == CLOSED)
    333     return;
    334   if (!spdy_websocket_stream_.get())
    335     return;
    336   if (buffer) {
    337     OnReceivedData(
    338         socket_.get(), buffer->GetRemainingData(), buffer->GetRemainingSize());
    339   } else {
    340     OnReceivedData(socket_.get(), NULL, 0);
    341   }
    342 }
    343 
    344 void WebSocketJob::OnCloseSpdyStream() {
    345   spdy_websocket_stream_.reset();
    346   OnClose(socket_.get());
    347 }
    348 
    349 bool WebSocketJob::SendHandshakeRequest(const char* data, int len) {
    350   DCHECK_EQ(state_, CONNECTING);
    351   if (started_to_send_handshake_request_)
    352     return false;
    353   if (!handshake_request_->ParseRequest(data, len))
    354     return false;
    355 
    356   AddCookieHeaderAndSend();
    357   return true;
    358 }
    359 
    360 void WebSocketJob::AddCookieHeaderAndSend() {
    361   bool allow = true;
    362   if (delegate_ && !delegate_->CanGetCookies(socket_.get(), GetURLForCookies()))
    363     allow = false;
    364 
    365   if (socket_.get() && delegate_ && state_ == CONNECTING) {
    366     handshake_request_->RemoveHeaders(kCookieHeaders,
    367                                       arraysize(kCookieHeaders));
    368     if (allow && socket_->cookie_store()) {
    369       // Add cookies, including HttpOnly cookies.
    370       CookieOptions cookie_options;
    371       cookie_options.set_include_httponly();
    372       socket_->cookie_store()->GetCookiesWithOptionsAsync(
    373           GetURLForCookies(), cookie_options,
    374           base::Bind(&WebSocketJob::LoadCookieCallback,
    375                      weak_ptr_factory_.GetWeakPtr()));
    376     } else {
    377       DoSendData();
    378     }
    379   }
    380 }
    381 
    382 void WebSocketJob::LoadCookieCallback(const std::string& cookie) {
    383   if (!cookie.empty())
    384     // TODO(tyoshino): Sending cookie means that connection doesn't need
    385     // PRIVACY_MODE_ENABLED as cookies may be server-bound and channel id
    386     // wouldn't negatively affect privacy anyway. Need to restart connection
    387     // or refactor to determine cookie status prior to connecting.
    388     handshake_request_->AppendHeaderIfMissing("Cookie", cookie);
    389   DoSendData();
    390 }
    391 
    392 void WebSocketJob::DoSendData() {
    393   if (spdy_websocket_stream_.get()) {
    394     scoped_ptr<SpdyHeaderBlock> headers(new SpdyHeaderBlock);
    395     handshake_request_->GetRequestHeaderBlock(
    396         socket_->url(), headers.get(), &challenge_, spdy_protocol_version_);
    397     spdy_websocket_stream_->SendRequest(headers.Pass());
    398   } else {
    399     const std::string& handshake_request =
    400         handshake_request_->GetRawRequest();
    401     handshake_request_sent_ = 0;
    402     socket_->net_log()->AddEvent(
    403         NetLog::TYPE_WEB_SOCKET_SEND_REQUEST_HEADERS,
    404         base::Bind(&NetLogWebSocketHandshakeCallback, &handshake_request));
    405     socket_->SendData(handshake_request.data(),
    406                       handshake_request.size());
    407   }
    408   // Just buffered in |handshake_request_|.
    409   started_to_send_handshake_request_ = true;
    410 }
    411 
    412 void WebSocketJob::OnSentHandshakeRequest(
    413     SocketStream* socket, int amount_sent) {
    414   DCHECK_EQ(state_, CONNECTING);
    415   handshake_request_sent_ += amount_sent;
    416   DCHECK_LE(handshake_request_sent_, handshake_request_->raw_length());
    417   if (handshake_request_sent_ >= handshake_request_->raw_length()) {
    418     // handshake request has been sent.
    419     // notify original size of handshake request to delegate.
    420     // Reset the handshake_request_ first in case this object is deleted by the
    421     // delegate.
    422     size_t original_length = handshake_request_->original_length();
    423     handshake_request_.reset();
    424     if (delegate_)
    425       delegate_->OnSentData(socket, original_length);
    426   }
    427 }
    428 
    429 void WebSocketJob::OnReceivedHandshakeResponse(
    430     SocketStream* socket, const char* data, int len) {
    431   DCHECK_EQ(state_, CONNECTING);
    432   if (handshake_response_->HasResponse()) {
    433     // If we already has handshake response, received data should be frame
    434     // data, not handshake message.
    435     received_data_after_handshake_.insert(
    436         received_data_after_handshake_.end(), data, data + len);
    437     return;
    438   }
    439 
    440   size_t response_length = handshake_response_->ParseRawResponse(data, len);
    441   if (!handshake_response_->HasResponse()) {
    442     // not yet. we need more data.
    443     return;
    444   }
    445   // handshake message is completed.
    446   std::string raw_response = handshake_response_->GetRawResponse();
    447   socket_->net_log()->AddEvent(
    448       NetLog::TYPE_WEB_SOCKET_READ_RESPONSE_HEADERS,
    449       base::Bind(&NetLogWebSocketHandshakeCallback, &raw_response));
    450   if (len - response_length > 0) {
    451     // If we received extra data, it should be frame data.
    452     DCHECK(received_data_after_handshake_.empty());
    453     received_data_after_handshake_.assign(data + response_length, data + len);
    454   }
    455   SaveCookiesAndNotifyHeadersComplete();
    456 }
    457 
    458 void WebSocketJob::SaveCookiesAndNotifyHeadersComplete() {
    459   // handshake message is completed.
    460   DCHECK(handshake_response_->HasResponse());
    461 
    462   // Extract cookies from the handshake response into a temporary vector.
    463   response_cookies_.clear();
    464   response_cookies_save_index_ = 0;
    465 
    466   handshake_response_->GetHeaders(
    467       kSetCookieHeaders, arraysize(kSetCookieHeaders), &response_cookies_);
    468 
    469   // Now, loop over the response cookies, and attempt to persist each.
    470   SaveNextCookie();
    471 }
    472 
    473 void WebSocketJob::NotifyHeadersComplete() {
    474   // Remove cookie headers, with malformed headers preserved.
    475   // Actual handshake should be done in Blink.
    476   handshake_response_->RemoveHeaders(
    477       kSetCookieHeaders, arraysize(kSetCookieHeaders));
    478   std::string handshake_response = handshake_response_->GetResponse();
    479   handshake_response_.reset();
    480   std::vector<char> received_data(handshake_response.begin(),
    481                                   handshake_response.end());
    482   received_data.insert(received_data.end(),
    483                        received_data_after_handshake_.begin(),
    484                        received_data_after_handshake_.end());
    485   received_data_after_handshake_.clear();
    486 
    487   state_ = OPEN;
    488 
    489   DCHECK(!received_data.empty());
    490   if (delegate_)
    491     delegate_->OnReceivedData(
    492         socket_.get(), &received_data.front(), received_data.size());
    493 
    494   WebSocketThrottle::GetInstance()->RemoveFromQueue(this);
    495 }
    496 
    497 void WebSocketJob::SaveNextCookie() {
    498   if (!socket_.get() || !delegate_ || state_ != CONNECTING)
    499     return;
    500 
    501   callback_pending_ = false;
    502   save_next_cookie_running_ = true;
    503 
    504   if (socket_->cookie_store()) {
    505     GURL url_for_cookies = GetURLForCookies();
    506 
    507     CookieOptions options;
    508     options.set_include_httponly();
    509 
    510     // Loop as long as SetCookieWithOptionsAsync completes synchronously. Since
    511     // CookieMonster's asynchronous operation APIs queue the callback to run it
    512     // on the thread where the API was called, there won't be race. I.e. unless
    513     // the callback is run synchronously, it won't be run in parallel with this
    514     // method.
    515     while (!callback_pending_ &&
    516            response_cookies_save_index_ < response_cookies_.size()) {
    517       std::string cookie = response_cookies_[response_cookies_save_index_];
    518       response_cookies_save_index_++;
    519 
    520       if (!delegate_->CanSetCookie(
    521               socket_.get(), url_for_cookies, cookie, &options))
    522         continue;
    523 
    524       callback_pending_ = true;
    525       socket_->cookie_store()->SetCookieWithOptionsAsync(
    526           url_for_cookies, cookie, options,
    527           base::Bind(&WebSocketJob::OnCookieSaved,
    528                      weak_ptr_factory_.GetWeakPtr()));
    529     }
    530   }
    531 
    532   save_next_cookie_running_ = false;
    533 
    534   if (callback_pending_)
    535     return;
    536 
    537   response_cookies_.clear();
    538   response_cookies_save_index_ = 0;
    539 
    540   NotifyHeadersComplete();
    541 }
    542 
    543 void WebSocketJob::OnCookieSaved(bool cookie_status) {
    544   // Tell the caller of SetCookieWithOptionsAsync() that this completion
    545   // callback is invoked.
    546   // - If the caller checks callback_pending earlier than this callback, the
    547   //   caller exits to let this method continue iteration.
    548   // - Otherwise, the caller continues iteration.
    549   callback_pending_ = false;
    550 
    551   // Resume SaveNextCookie if the caller of SetCookieWithOptionsAsync() exited
    552   // the loop. Otherwise, return.
    553   if (save_next_cookie_running_)
    554     return;
    555 
    556   SaveNextCookie();
    557 }
    558 
    559 GURL WebSocketJob::GetURLForCookies() const {
    560   GURL url = socket_->url();
    561   std::string scheme = socket_->is_secure() ? "https" : "http";
    562   url::Replacements<char> replacements;
    563   replacements.SetScheme(scheme.c_str(), url::Component(0, scheme.length()));
    564   return url.ReplaceComponents(replacements);
    565 }
    566 
    567 const AddressList& WebSocketJob::address_list() const {
    568   return addresses_;
    569 }
    570 
    571 int WebSocketJob::TrySpdyStream() {
    572   if (!socket_.get())
    573     return ERR_FAILED;
    574 
    575   // Check if we have a SPDY session available.
    576   HttpTransactionFactory* factory =
    577       socket_->context()->http_transaction_factory();
    578   if (!factory)
    579     return OK;
    580   scoped_refptr<HttpNetworkSession> session = factory->GetSession();
    581   if (!session.get() || !session->params().enable_websocket_over_spdy)
    582     return OK;
    583   SpdySessionPool* spdy_pool = session->spdy_session_pool();
    584   PrivacyMode privacy_mode = socket_->privacy_mode();
    585   const SpdySessionKey key(HostPortPair::FromURL(socket_->url()),
    586                            socket_->proxy_server(), privacy_mode);
    587   // Forbid wss downgrade to SPDY without SSL.
    588   // TODO(toyoshim): Does it realize the same policy with HTTP?
    589   base::WeakPtr<SpdySession> spdy_session =
    590       spdy_pool->FindAvailableSession(key, *socket_->net_log());
    591   if (!spdy_session)
    592     return OK;
    593 
    594   SSLInfo ssl_info;
    595   bool was_npn_negotiated;
    596   NextProto protocol_negotiated = kProtoUnknown;
    597   bool use_ssl = spdy_session->GetSSLInfo(
    598       &ssl_info, &was_npn_negotiated, &protocol_negotiated);
    599   if (socket_->is_secure() && !use_ssl)
    600     return OK;
    601 
    602   // Create SpdyWebSocketStream.
    603   spdy_protocol_version_ = spdy_session->GetProtocolVersion();
    604   spdy_websocket_stream_.reset(new SpdyWebSocketStream(spdy_session, this));
    605 
    606   int result = spdy_websocket_stream_->InitializeStream(
    607       socket_->url(), MEDIUM, *socket_->net_log());
    608   if (result == OK) {
    609     OnConnected(socket_.get(), kMaxPendingSendAllowed);
    610     return ERR_PROTOCOL_SWITCHED;
    611   }
    612   if (result != ERR_IO_PENDING) {
    613     spdy_websocket_stream_.reset();
    614     return OK;
    615   }
    616 
    617   return ERR_IO_PENDING;
    618 }
    619 
    620 void WebSocketJob::SetWaiting() {
    621   waiting_ = true;
    622 }
    623 
    624 bool WebSocketJob::IsWaiting() const {
    625   return waiting_;
    626 }
    627 
    628 void WebSocketJob::Wakeup() {
    629   if (!waiting_)
    630     return;
    631   waiting_ = false;
    632   DCHECK(!callback_.is_null());
    633   base::MessageLoopForIO::current()->PostTask(
    634       FROM_HERE,
    635       base::Bind(&WebSocketJob::RetryPendingIO,
    636                  weak_ptr_factory_.GetWeakPtr()));
    637 }
    638 
    639 void WebSocketJob::RetryPendingIO() {
    640   int result = TrySpdyStream();
    641 
    642   // In the case of ERR_IO_PENDING, CompleteIO() will be called from
    643   // OnCreatedSpdyStream().
    644   if (result != ERR_IO_PENDING)
    645     CompleteIO(result);
    646 }
    647 
    648 void WebSocketJob::CompleteIO(int result) {
    649   // |callback_| may be null if OnClose() or DetachDelegate() was called.
    650   if (!callback_.is_null()) {
    651     CompletionCallback callback = callback_;
    652     callback_.Reset();
    653     callback.Run(result);
    654     Release();  // Balanced with OnStartOpenConnection().
    655   }
    656 }
    657 
    658 bool WebSocketJob::SendDataInternal(const char* data, int length) {
    659   if (spdy_websocket_stream_.get())
    660     return ERR_IO_PENDING == spdy_websocket_stream_->SendData(data, length);
    661   if (socket_.get())
    662     return socket_->SendData(data, length);
    663   return false;
    664 }
    665 
    666 void WebSocketJob::CloseInternal() {
    667   if (spdy_websocket_stream_.get())
    668     spdy_websocket_stream_->Close();
    669   if (socket_.get())
    670     socket_->Close();
    671 }
    672 
    673 void WebSocketJob::SendPending() {
    674   if (current_send_buffer_.get())
    675     return;
    676 
    677   // Current buffer has been sent. Try next if any.
    678   if (send_buffer_queue_.empty()) {
    679     // No more data to send.
    680     if (state_ == CLOSING)
    681       CloseInternal();
    682     return;
    683   }
    684 
    685   scoped_refptr<IOBufferWithSize> next_buffer = send_buffer_queue_.front();
    686   send_buffer_queue_.pop_front();
    687   current_send_buffer_ =
    688       new DrainableIOBuffer(next_buffer.get(), next_buffer->size());
    689   SendDataInternal(current_send_buffer_->data(),
    690                    current_send_buffer_->BytesRemaining());
    691 }
    692 
    693 }  // namespace net
    694