extensions/browser/api/cast_channel/keep_alive_delegate.cc

   1 // Copyright 2015 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 "base/json/json_reader.h"
   6 #include "base/json/json_writer.h"
   7 #include "extensions/browser/api/cast_channel/cast_message_util.h"
   8 #include "extensions/browser/api/cast_channel/cast_socket.h"
   9 #include "extensions/browser/api/cast_channel/keep_alive_delegate.h"
  10 #include "extensions/common/api/cast_channel/cast_channel.pb.h"
  11 #include "extensions/common/api/cast_channel/logging.pb.h"
  12 #include "net/base/net_errors.h"
  13
  14 namespace extensions {
  15 namespace core_api {
  16 namespace cast_channel {
  17 namespace {
  18
  19 const char kHeartbeatNamespace[] = "urn:x-cast:com.google.cast.tp.heartbeat";
  20 const char kPingSenderId[] = "chrome";
  21 const char kPingReceiverId[] = "receiver-0";
  22 const char kTypeNodeId[] = "type";
  23
  24 // Determines if the JSON-encoded payload is equivalent to
  25 // { "type": |chk_type| }
  26 bool NestedPayloadTypeEquals(const std::string& chk_type,
  27                              const CastMessage& message) {
  28   MessageInfo message_info;
  29   CastMessageToMessageInfo(message, &message_info);
  30   std::string type_json;
  31   if (!message_info.data->GetAsString(&type_json)) {
  32     return false;
  33   }
  34   scoped_ptr<base::Value> type_value(base::JSONReader::Read(type_json));
  35   if (!type_value.get()) {
  36     return false;
  37   }
  38
  39   base::DictionaryValue* type_dict;
  40   if (!type_value->GetAsDictionary(&type_dict)) {
  41     return false;
  42   }
  43
  44   std::string type_string;
  45   return (type_dict->HasKey(kTypeNodeId) &&
  46           type_dict->GetString(kTypeNodeId, &type_string) &&
  47           type_string == chk_type);
  48 }
  49
  50 }  // namespace
  51
  52 // static
  53 const char KeepAliveDelegate::kHeartbeatPingType[] = "PING";
  54
  55 // static
  56 const char KeepAliveDelegate::kHeartbeatPongType[] = "PONG";
  57
  58 // static
  59 CastMessage KeepAliveDelegate::CreateKeepAliveMessage(
  60     const char* message_type) {
  61   CastMessage output;
  62   output.set_protocol_version(CastMessage::CASTV2_1_0);
  63   output.set_source_id(kPingSenderId);
  64   output.set_destination_id(kPingReceiverId);
  65   output.set_namespace_(kHeartbeatNamespace);
  66   base::DictionaryValue type_dict;
  67   type_dict.SetString(kTypeNodeId, message_type);
  68   if (!base::JSONWriter::Write(&type_dict, output.mutable_payload_utf8())) {
  69     LOG(ERROR) << "Failed to serialize dictionary.";
  70     return output;
  71   }
  72   output.set_payload_type(
  73       CastMessage::PayloadType::CastMessage_PayloadType_STRING);
  74   return output;
  75 }
  76
  77 KeepAliveDelegate::KeepAliveDelegate(
  78     CastSocket* socket,
  79     scoped_ptr<CastTransport::Delegate> inner_delegate,
  80     base::TimeDelta ping_interval,
  81     base::TimeDelta liveness_timeout)
  82     : started_(false),
  83       socket_(socket),
  84       inner_delegate_(inner_delegate.Pass()),
  85       liveness_timeout_(liveness_timeout),
  86       ping_interval_(ping_interval) {
  87   DCHECK(ping_interval_ < liveness_timeout_);
  88   DCHECK(inner_delegate_);
  89   DCHECK(socket_);
  90   ping_message_ = CreateKeepAliveMessage(kHeartbeatPingType);
  91   pong_message_ = CreateKeepAliveMessage(kHeartbeatPongType);
  92 }
  93
  94 KeepAliveDelegate::~KeepAliveDelegate() {
  95 }
  96
  97 void KeepAliveDelegate::SetTimersForTest(
  98     scoped_ptr<base::Timer> injected_ping_timer,
  99     scoped_ptr<base::Timer> injected_liveness_timer) {
 100   ping_timer_ = injected_ping_timer.Pass();
 101   liveness_timer_ = injected_liveness_timer.Pass();
 102 }
 103
 104 void KeepAliveDelegate::Start() {
 105   DCHECK(thread_checker_.CalledOnValidThread());
 106   DCHECK(!started_);
 107
 108   VLOG(1) << "Starting keep-alive timers.";
 109   VLOG(1) << "Ping timeout: " << ping_interval_;
 110   VLOG(1) << "Liveness timeout: " << liveness_timeout_;
 111
 112   // Use injected mock timers, if provided.
 113   if (!ping_timer_) {
 114     ping_timer_.reset(new base::Timer(true, false));
 115   }
 116   if (!liveness_timer_) {
 117     liveness_timer_.reset(new base::Timer(true, false));
 118   }
 119
 120   ping_timer_->Start(
 121       FROM_HERE, ping_interval_,
 122       base::Bind(&KeepAliveDelegate::SendKeepAliveMessage,
 123                  base::Unretained(this), ping_message_, kHeartbeatPingType));
 124   liveness_timer_->Start(
 125       FROM_HERE, liveness_timeout_,
 126       base::Bind(&KeepAliveDelegate::LivenessTimeout, base::Unretained(this)));
 127
 128   started_ = true;
 129   inner_delegate_->Start();
 130 }
 131
 132 void KeepAliveDelegate::ResetTimers() {
 133   DCHECK(started_);
 134   ping_timer_->Reset();
 135   liveness_timer_->Reset();
 136 }
 137
 138 void KeepAliveDelegate::SendKeepAliveMessage(const CastMessage& message,
 139                                              const char* message_type) const {
 140   DCHECK(thread_checker_.CalledOnValidThread());
 141   VLOG(1) << "Sending " << message_type;
 142   socket_->transport()->SendMessage(
 143       message, base::Bind(&KeepAliveDelegate::SendKeepAliveMessageComplete,
 144                           base::Unretained(this), message_type));
 145 }
 146
 147 void KeepAliveDelegate::SendKeepAliveMessageComplete(const char* message_type,
 148                                                      int rv) const {
 149   VLOG(2) << "Sending " << message_type << " complete, rv=" << rv;
 150   if (rv != net::OK) {
 151     // An error occurred while sending the ping response.
 152     // Close the connection.
 153     VLOG(1) << "Error sending " << message_type;
 154     inner_delegate_->OnError(cast_channel::CHANNEL_ERROR_SOCKET_ERROR,
 155                              LastErrors());
 156   }
 157 }
 158
 159 void KeepAliveDelegate::LivenessTimeout() const {
 160   VLOG(1) << "Ping timeout";
 161   inner_delegate_->OnError(cast_channel::CHANNEL_ERROR_PING_TIMEOUT,
 162                            LastErrors());
 163 }
 164
 165 // CastTransport::Delegate interface.
 166 void KeepAliveDelegate::OnError(ChannelError error_state,
 167                                 const LastErrors& last_errors) {
 168   DCHECK(started_);
 169   DCHECK(thread_checker_.CalledOnValidThread());
 170   VLOG(2) << "KeepAlive::OnError";
 171   inner_delegate_->OnError(error_state, last_errors);
 172 }
 173
 174 void KeepAliveDelegate::OnMessage(const CastMessage& message) {
 175   DCHECK(started_);
 176   DCHECK(thread_checker_.CalledOnValidThread());
 177   VLOG(2) << "KeepAlive::OnMessage : " << message.payload_utf8();
 178
 179   ResetTimers();
 180
 181   if (NestedPayloadTypeEquals(kHeartbeatPingType, message)) {
 182     VLOG(1) << "Received PING.";
 183     SendKeepAliveMessage(pong_message_, kHeartbeatPongType);
 184   } else if (NestedPayloadTypeEquals(kHeartbeatPongType, message)) {
 185     VLOG(1) << "Received PONG.";
 186   } else {
 187     // PING and PONG messages are intentionally suppressed from layers above.
 188     inner_delegate_->OnMessage(message);
 189   }
 190 }
 191
 192 }  // namespace cast_channel
 193 }  // namespace core_api
 194 }  // namespace extensions