src/net/websockets/websocket_job.cc

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