[cast_channel] Implement CastSocketService::OpenSocket()

components/cast_channel/cast_socket.cc

 // Copyright 2014 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 "components/cast_channel/cast_socket.h"
 
 #include <stdlib.h>
 #include <string.h>
 
 #include <utility>
@@ skipping 118 matching lines @@
   DCHECK(net_log_);
   net_log_source_.type = net::NetLogSourceType::SOCKET;
   net_log_source_.id = net_log_->NextID();
 }
 
 CastSocketImpl::~CastSocketImpl() {
   // Ensure that resources are freed but do not run pending callbacks that
   // would result in re-entrancy.
   CloseInternal();
 
-  if (!connect_callback_.is_null())
-    base::ResetAndReturn(&connect_callback_).Run(ChannelError::UNKNOWN);
+  for (auto& connect_callback : connect_callbacks_)
+    connect_callback.Run(channel_id_, ChannelError::UNKNOWN);
+  connect_callbacks_.clear();
 }
 
 ReadyState CastSocketImpl::ready_state() const {
   return ready_state_;
 }
 
 ChannelError CastSocketImpl::error_state() const {
   return error_state_;
 }
 
@@ skipping 82 matching lines @@
     }
   }
   return result.success();
 }
 
 void CastSocketImpl::SetTransportForTesting(
     std::unique_ptr<CastTransport> transport) {
   transport_ = std::move(transport);
 }
 
-void CastSocketImpl::Connect(base::Callback<void(ChannelError)> callback) {
+void CastSocketImpl::Connect(const OnOpenCallback& callback) {
+  switch (ready_state_) {
+    case ReadyState::NONE:
+      connect_callbacks_.push_back(callback);
+      Connect();
+      break;
+    case ReadyState::CONNECTING:
+      connect_callbacks_.push_back(callback);
+      break;
+    case ReadyState::OPEN:
+      callback.Run(channel_id_, ChannelError::NONE);
+      break;
+    case ReadyState::CLOSED:
+      callback.Run(channel_id_, ChannelError::CONNECT_ERROR);
+      break;
+    default:
+      NOTREACHED() << "Unknown ReadyState: "
+                   << ReadyStateToString(ready_state_);
+  }
+}
+
+void CastSocketImpl::Connect() {
   DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
   VLOG_WITH_CONNECTION(1) << "Connect readyState = "
                           << ReadyStateToString(ready_state_);
+  DCHECK_EQ(ReadyState::NONE, ready_state_);
   DCHECK_EQ(ConnectionState::START_CONNECT, connect_state_);
 
   delegate_ = base::MakeUnique<CastSocketMessageDelegate>(this);
 
-  if (ready_state_ != ReadyState::NONE) {
-    callback.Run(ChannelError::CONNECT_ERROR);
-    return;
-  }
-
-  connect_callback_ = callback;
   SetReadyState(ReadyState::CONNECTING);
   SetConnectState(ConnectionState::TCP_CONNECT);
 
   // Set up connection timeout.
   if (connect_timeout_.InMicroseconds() > 0) {
     DCHECK(connect_timeout_callback_.IsCancelled());
     connect_timeout_callback_.Reset(
         base::Bind(&CastSocketImpl::OnConnectTimeout, base::Unretained(this)));
     GetTimer()->Start(FROM_HERE, connect_timeout_,
                       connect_timeout_callback_.callback());
@@ skipping 258 matching lines @@
   }
   VLOG_WITH_CONNECTION(1) << "Auth challenge verification succeeded";
 
   SetConnectState(ConnectionState::FINISHED);
   return net::OK;
 }
 
 void CastSocketImpl::DoConnectCallback() {
   VLOG(1) << "DoConnectCallback (error_state = "
           << ChannelErrorToString(error_state_) << ")";
-  if (connect_callback_.is_null()) {
+  if (connect_callbacks_.empty()) {
     DLOG(FATAL) << "Connection callback invoked multiple times.";
     return;
   }
 
   if (error_state_ == ChannelError::NONE) {
     SetReadyState(ReadyState::OPEN);
     if (keep_alive()) {
       auto* keep_alive_delegate =
           new KeepAliveDelegate(this, logger_, std::move(delegate_),
                                 ping_interval_, liveness_timeout_);
       delegate_.reset(keep_alive_delegate);
     }
     transport_->SetReadDelegate(std::move(delegate_));
   } else {
     CloseInternal();
   }
 
-  base::ResetAndReturn(&connect_callback_).Run(error_state_);
+  for (auto& connect_callback : connect_callbacks_)
+    connect_callback.Run(channel_id_, error_state_);
+  connect_callbacks_.clear();
 }
 
 void CastSocketImpl::Close(const net::CompletionCallback& callback) {
   DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
   CloseInternal();
   // Run this callback last.  It may delete the socket.
   callback.Run(net::OK);
 }
 
 void CastSocketImpl::CloseInternal() {
@@ skipping 62 matching lines @@
 void CastSocketImpl::CastSocketMessageDelegate::OnMessage(
     const CastMessage& message) {
   for (auto& observer : socket_->observers_)
     observer.OnMessage(*socket_, message);
 }
 
 void CastSocketImpl::CastSocketMessageDelegate::Start() {}
 
 }  // namespace cast_channel
 #undef VLOG_WITH_CONNECTION