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 // handshake message is completed. 362 handshake_response_->set_protocol_version( 363 handshake_request_->protocol_version()); 364 AddCookieHeaderAndSend(); 365 return true; 366 } 367 368 void WebSocketJob::AddCookieHeaderAndSend() { 369 bool allow = true; 370 if (delegate_ && !delegate_->CanGetCookies(socket_.get(), GetURLForCookies())) 371 allow = false; 372 373 if (socket_.get() && delegate_ && state_ == CONNECTING) { 374 handshake_request_->RemoveHeaders(kCookieHeaders, 375 arraysize(kCookieHeaders)); 376 if (allow && socket_->context()->cookie_store()) { 377 // Add cookies, including HttpOnly cookies. 378 CookieOptions cookie_options; 379 cookie_options.set_include_httponly(); 380 socket_->context()->cookie_store()->GetCookiesWithOptionsAsync( 381 GetURLForCookies(), cookie_options, 382 base::Bind(&WebSocketJob::LoadCookieCallback, 383 weak_ptr_factory_.GetWeakPtr())); 384 } else { 385 DoSendData(); 386 } 387 } 388 } 389 390 void WebSocketJob::LoadCookieCallback(const std::string& cookie) { 391 if (!cookie.empty()) 392 // TODO(tyoshino): Sending cookie means that connection doesn't need 393 // kPrivacyModeEnabled as cookies may be server-bound and channel id 394 // wouldn't negatively affect privacy anyway. Need to restart connection 395 // or refactor to determine cookie status prior to connecting. 396 handshake_request_->AppendHeaderIfMissing("Cookie", cookie); 397 DoSendData(); 398 } 399 400 void WebSocketJob::DoSendData() { 401 if (spdy_websocket_stream_.get()) { 402 scoped_ptr<SpdyHeaderBlock> headers(new SpdyHeaderBlock); 403 handshake_request_->GetRequestHeaderBlock( 404 socket_->url(), headers.get(), &challenge_, spdy_protocol_version_); 405 spdy_websocket_stream_->SendRequest(headers.Pass()); 406 } else { 407 const std::string& handshake_request = 408 handshake_request_->GetRawRequest(); 409 handshake_request_sent_ = 0; 410 socket_->net_log()->AddEvent( 411 NetLog::TYPE_WEB_SOCKET_SEND_REQUEST_HEADERS, 412 base::Bind(&NetLogWebSocketHandshakeCallback, &handshake_request)); 413 socket_->SendData(handshake_request.data(), 414 handshake_request.size()); 415 } 416 // Just buffered in |handshake_request_|. 417 started_to_send_handshake_request_ = true; 418 } 419 420 void WebSocketJob::OnSentHandshakeRequest( 421 SocketStream* socket, int amount_sent) { 422 DCHECK_EQ(state_, CONNECTING); 423 handshake_request_sent_ += amount_sent; 424 DCHECK_LE(handshake_request_sent_, handshake_request_->raw_length()); 425 if (handshake_request_sent_ >= handshake_request_->raw_length()) { 426 // handshake request has been sent. 427 // notify original size of handshake request to delegate. 428 if (delegate_) 429 delegate_->OnSentData( 430 socket, 431 handshake_request_->original_length()); 432 handshake_request_.reset(); 433 } 434 } 435 436 void WebSocketJob::OnReceivedHandshakeResponse( 437 SocketStream* socket, const char* data, int len) { 438 DCHECK_EQ(state_, CONNECTING); 439 if (handshake_response_->HasResponse()) { 440 // If we already has handshake response, received data should be frame 441 // data, not handshake message. 442 received_data_after_handshake_.insert( 443 received_data_after_handshake_.end(), data, data + len); 444 return; 445 } 446 447 size_t response_length = handshake_response_->ParseRawResponse(data, len); 448 if (!handshake_response_->HasResponse()) { 449 // not yet. we need more data. 450 return; 451 } 452 // handshake message is completed. 453 std::string raw_response = handshake_response_->GetRawResponse(); 454 socket_->net_log()->AddEvent( 455 NetLog::TYPE_WEB_SOCKET_READ_RESPONSE_HEADERS, 456 base::Bind(&NetLogWebSocketHandshakeCallback, &raw_response)); 457 if (len - response_length > 0) { 458 // If we received extra data, it should be frame data. 459 DCHECK(received_data_after_handshake_.empty()); 460 received_data_after_handshake_.assign(data + response_length, data + len); 461 } 462 SaveCookiesAndNotifyHeadersComplete(); 463 } 464 465 void WebSocketJob::SaveCookiesAndNotifyHeadersComplete() { 466 // handshake message is completed. 467 DCHECK(handshake_response_->HasResponse()); 468 469 // Extract cookies from the handshake response into a temporary vector. 470 response_cookies_.clear(); 471 response_cookies_save_index_ = 0; 472 473 handshake_response_->GetHeaders( 474 kSetCookieHeaders, arraysize(kSetCookieHeaders), &response_cookies_); 475 476 // Now, loop over the response cookies, and attempt to persist each. 477 SaveNextCookie(); 478 } 479 480 void WebSocketJob::NotifyHeadersComplete() { 481 // Remove cookie headers, with malformed headers preserved. 482 // Actual handshake should be done in Blink. 483 handshake_response_->RemoveHeaders( 484 kSetCookieHeaders, arraysize(kSetCookieHeaders)); 485 std::string handshake_response = handshake_response_->GetResponse(); 486 handshake_response_.reset(); 487 std::vector<char> received_data(handshake_response.begin(), 488 handshake_response.end()); 489 received_data.insert(received_data.end(), 490 received_data_after_handshake_.begin(), 491 received_data_after_handshake_.end()); 492 received_data_after_handshake_.clear(); 493 494 state_ = OPEN; 495 496 DCHECK(!received_data.empty()); 497 if (delegate_) 498 delegate_->OnReceivedData( 499 socket_.get(), &received_data.front(), received_data.size()); 500 501 WebSocketThrottle::GetInstance()->RemoveFromQueue(this); 502 } 503 504 void WebSocketJob::SaveNextCookie() { 505 if (!socket_.get() || !delegate_ || state_ != CONNECTING) 506 return; 507 508 callback_pending_ = false; 509 save_next_cookie_running_ = true; 510 511 if (socket_->context()->cookie_store()) { 512 GURL url_for_cookies = GetURLForCookies(); 513 514 CookieOptions options; 515 options.set_include_httponly(); 516 517 // Loop as long as SetCookieWithOptionsAsync completes synchronously. Since 518 // CookieMonster's asynchronous operation APIs queue the callback to run it 519 // on the thread where the API was called, there won't be race. I.e. unless 520 // the callback is run synchronously, it won't be run in parallel with this 521 // method. 522 while (!callback_pending_ && 523 response_cookies_save_index_ < response_cookies_.size()) { 524 std::string cookie = response_cookies_[response_cookies_save_index_]; 525 response_cookies_save_index_++; 526 527 if (!delegate_->CanSetCookie( 528 socket_.get(), url_for_cookies, cookie, &options)) 529 continue; 530 531 callback_pending_ = true; 532 socket_->context()->cookie_store()->SetCookieWithOptionsAsync( 533 url_for_cookies, cookie, options, 534 base::Bind(&WebSocketJob::OnCookieSaved, 535 weak_ptr_factory_.GetWeakPtr())); 536 } 537 } 538 539 save_next_cookie_running_ = false; 540 541 if (callback_pending_) 542 return; 543 544 response_cookies_.clear(); 545 response_cookies_save_index_ = 0; 546 547 NotifyHeadersComplete(); 548 } 549 550 void WebSocketJob::OnCookieSaved(bool cookie_status) { 551 // Tell the caller of SetCookieWithOptionsAsync() that this completion 552 // callback is invoked. 553 // - If the caller checks callback_pending earlier than this callback, the 554 // caller exits to let this method continue iteration. 555 // - Otherwise, the caller continues iteration. 556 callback_pending_ = false; 557 558 // Resume SaveNextCookie if the caller of SetCookieWithOptionsAsync() exited 559 // the loop. Otherwise, return. 560 if (save_next_cookie_running_) 561 return; 562 563 SaveNextCookie(); 564 } 565 566 GURL WebSocketJob::GetURLForCookies() const { 567 GURL url = socket_->url(); 568 std::string scheme = socket_->is_secure() ? "https" : "http"; 569 url_canon::Replacements<char> replacements; 570 replacements.SetScheme(scheme.c_str(), 571 url_parse::Component(0, scheme.length())); 572 return url.ReplaceComponents(replacements); 573 } 574 575 const AddressList& WebSocketJob::address_list() const { 576 return addresses_; 577 } 578 579 int WebSocketJob::TrySpdyStream() { 580 if (!socket_.get()) 581 return ERR_FAILED; 582 583 if (!websocket_over_spdy_enabled_) 584 return OK; 585 586 // Check if we have a SPDY session available. 587 HttpTransactionFactory* factory = 588 socket_->context()->http_transaction_factory(); 589 if (!factory) 590 return OK; 591 scoped_refptr<HttpNetworkSession> session = factory->GetSession(); 592 if (!session.get()) 593 return OK; 594 SpdySessionPool* spdy_pool = session->spdy_session_pool(); 595 PrivacyMode privacy_mode = socket_->privacy_mode(); 596 const SpdySessionKey key(HostPortPair::FromURL(socket_->url()), 597 socket_->proxy_server(), privacy_mode); 598 // Forbid wss downgrade to SPDY without SSL. 599 // TODO(toyoshim): Does it realize the same policy with HTTP? 600 base::WeakPtr<SpdySession> spdy_session = 601 spdy_pool->FindAvailableSession(key, *socket_->net_log()); 602 if (!spdy_session) 603 return OK; 604 605 SSLInfo ssl_info; 606 bool was_npn_negotiated; 607 NextProto protocol_negotiated = kProtoUnknown; 608 bool use_ssl = spdy_session->GetSSLInfo( 609 &ssl_info, &was_npn_negotiated, &protocol_negotiated); 610 if (socket_->is_secure() && !use_ssl) 611 return OK; 612 613 // Create SpdyWebSocketStream. 614 spdy_protocol_version_ = spdy_session->GetProtocolVersion(); 615 spdy_websocket_stream_.reset(new SpdyWebSocketStream(spdy_session, this)); 616 617 int result = spdy_websocket_stream_->InitializeStream( 618 socket_->url(), MEDIUM, *socket_->net_log()); 619 if (result == OK) { 620 OnConnected(socket_.get(), kMaxPendingSendAllowed); 621 return ERR_PROTOCOL_SWITCHED; 622 } 623 if (result != ERR_IO_PENDING) { 624 spdy_websocket_stream_.reset(); 625 return OK; 626 } 627 628 return ERR_IO_PENDING; 629 } 630 631 void WebSocketJob::SetWaiting() { 632 waiting_ = true; 633 } 634 635 bool WebSocketJob::IsWaiting() const { 636 return waiting_; 637 } 638 639 void WebSocketJob::Wakeup() { 640 if (!waiting_) 641 return; 642 waiting_ = false; 643 DCHECK(!callback_.is_null()); 644 base::MessageLoopForIO::current()->PostTask( 645 FROM_HERE, 646 base::Bind(&WebSocketJob::RetryPendingIO, 647 weak_ptr_factory_.GetWeakPtr())); 648 } 649 650 void WebSocketJob::RetryPendingIO() { 651 int result = TrySpdyStream(); 652 653 // In the case of ERR_IO_PENDING, CompleteIO() will be called from 654 // OnCreatedSpdyStream(). 655 if (result != ERR_IO_PENDING) 656 CompleteIO(result); 657 } 658 659 void WebSocketJob::CompleteIO(int result) { 660 // |callback_| may be null if OnClose() or DetachDelegate() was called. 661 if (!callback_.is_null()) { 662 CompletionCallback callback = callback_; 663 callback_.Reset(); 664 callback.Run(result); 665 Release(); // Balanced with OnStartOpenConnection(). 666 } 667 } 668 669 bool WebSocketJob::SendDataInternal(const char* data, int length) { 670 if (spdy_websocket_stream_.get()) 671 return ERR_IO_PENDING == spdy_websocket_stream_->SendData(data, length); 672 if (socket_.get()) 673 return socket_->SendData(data, length); 674 return false; 675 } 676 677 void WebSocketJob::CloseInternal() { 678 if (spdy_websocket_stream_.get()) 679 spdy_websocket_stream_->Close(); 680 if (socket_.get()) 681 socket_->Close(); 682 } 683 684 void WebSocketJob::SendPending() { 685 if (current_send_buffer_.get()) 686 return; 687 688 // Current buffer has been sent. Try next if any. 689 if (send_buffer_queue_.empty()) { 690 // No more data to send. 691 if (state_ == CLOSING) 692 CloseInternal(); 693 return; 694 } 695 696 scoped_refptr<IOBufferWithSize> next_buffer = send_buffer_queue_.front(); 697 send_buffer_queue_.pop_front(); 698 current_send_buffer_ = 699 new DrainableIOBuffer(next_buffer.get(), next_buffer->size()); 700 SendDataInternal(current_send_buffer_->data(), 701 current_send_buffer_->BytesRemaining()); 702 } 703 704 } // namespace net 705