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