remoting/host/heartbeat_sender.cc

// Copyright (c) 2012 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 "remoting/host/heartbeat_sender.h"

#include <math.h>

#include "base/bind.h"
#include "base/logging.h"
#include "base/message_loop_proxy.h"
#include "base/rand_util.h"
#include "base/string_number_conversions.h"
#include "base/time.h"
#include "remoting/base/constants.h"
#include "remoting/jingle_glue/iq_sender.h"
#include "remoting/jingle_glue/jingle_thread.h"
#include "remoting/jingle_glue/signal_strategy.h"
#include "third_party/libjingle/source/talk/xmllite/xmlelement.h"
#include "third_party/libjingle/source/talk/xmpp/constants.h"

using buzz::QName;
using buzz::XmlElement;

namespace remoting {

namespace {

const char kHeartbeatQueryTag[] = "heartbeat";
const char kHostIdAttr[] = "hostid";
const char kHeartbeatSignatureTag[] = "signature";
const char kSequenceIdAttr[] = "sequence-id";

const char kErrorTag[] = "error";
const char kNotFoundTag[] = "item-not-found";

const char kHeartbeatResultTag[] = "heartbeat-result";
const char kSetIntervalTag[] = "set-interval";
const char kExpectedSequenceIdTag[] = "expected-sequence-id";

const int64 kDefaultHeartbeatIntervalMs = 5 * 60 * 1000;  // 5 minutes.
const int64 kResendDelayMs = 10 * 1000;  // 10 seconds.

const int kExitCodeHostIdInvalid = 100;

}  // namespace

HeartbeatSender::HeartbeatSender(
    const std::string& host_id,
    SignalStrategy* signal_strategy,
    HostKeyPair* key_pair)
    : host_id_(host_id),
      signal_strategy_(signal_strategy),
      key_pair_(key_pair),
      interval_ms_(kDefaultHeartbeatIntervalMs),
      sequence_id_(0),
      sequence_id_was_set_(false),
      sequence_id_recent_set_num_(0) {
  DCHECK(signal_strategy_);
  DCHECK(key_pair_);

  signal_strategy_->AddListener(this);

  // Start heartbeats if the |signal_strategy_| is already connected.
  OnSignalStrategyStateChange(signal_strategy_->GetState());
}

HeartbeatSender::~HeartbeatSender() {
  signal_strategy_->RemoveListener(this);
}

void HeartbeatSender::OnSignalStrategyStateChange(SignalStrategy::State state) {
  if (state == SignalStrategy::CONNECTED) {
    iq_sender_.reset(new IqSender(signal_strategy_));
    SendStanza();
    timer_.Start(FROM_HERE, base::TimeDelta::FromMilliseconds(interval_ms_),
                 this, &HeartbeatSender::SendStanza);
  } else if (state == SignalStrategy::DISCONNECTED) {
    request_.reset();
    iq_sender_.reset();
    timer_.Stop();
    timer_resend_.Stop();
  }
}

void HeartbeatSender::SendStanza() {
  DoSendStanza();
  // Make sure we don't send another heartbeat before the heartbeat interval
  // has expired.
  timer_resend_.Stop();
}

void HeartbeatSender::ResendStanza() {
  DoSendStanza();
  // Make sure we don't send another heartbeat before the heartbeat interval
  // has expired.
  timer_.Reset();
}

void HeartbeatSender::DoSendStanza() {
  VLOG(1) << "Sending heartbeat stanza to " << kChromotingBotJid;
  request_ = iq_sender_->SendIq(
      buzz::STR_SET, kChromotingBotJid, CreateHeartbeatMessage(),
      base::Bind(&HeartbeatSender::ProcessResponse,
                 base::Unretained(this)));
  ++sequence_id_;
}

void HeartbeatSender::ProcessResponse(IqRequest* request,
                                      const XmlElement* response) {
  std::string type = response->Attr(buzz::QN_TYPE);
  if (type == buzz::STR_ERROR) {
    const XmlElement* error_element =
        response->FirstNamed(QName(buzz::NS_CLIENT, kErrorTag));
    if (error_element) {
      if (error_element->FirstNamed(QName(buzz::NS_STANZA, kNotFoundTag))) {
        // TODO(lambroslambrou): Trigger an application-defined callback to
        // shut down the host properly, instead of just exiting here
        // (http://crbug.com/112160).
        LOG(ERROR) << "Received error: Host ID invalid";
        exit(kExitCodeHostIdInvalid);
      }
    }

    LOG(ERROR) << "Received error in response to heartbeat: "
               << response->Str();
    return;
  }

  // This method must only be called for error or result stanzas.
  DCHECK_EQ(std::string(buzz::STR_RESULT), type);

  const XmlElement* result_element =
      response->FirstNamed(QName(kChromotingXmlNamespace, kHeartbeatResultTag));
  if (result_element) {
    const XmlElement* set_interval_element =
        result_element->FirstNamed(QName(kChromotingXmlNamespace,
                                         kSetIntervalTag));
    if (set_interval_element) {
      const std::string& interval_str = set_interval_element->BodyText();
      int interval;
      if (!base::StringToInt(interval_str, &interval) || interval <= 0) {
        LOG(ERROR) << "Received invalid set-interval: "
                   << set_interval_element->Str();
      } else {
        SetInterval(interval * base::Time::kMillisecondsPerSecond);
      }
    }

    bool did_set_sequence_id = false;
    const XmlElement* expected_sequence_id_element =
        result_element->FirstNamed(QName(kChromotingXmlNamespace,
                                         kExpectedSequenceIdTag));
    if (expected_sequence_id_element) {
      // The sequence ID sent in the previous heartbeat was not what the server
      // expected, so send another heartbeat with the expected sequence ID.
      const std::string& expected_sequence_id_str =
          expected_sequence_id_element->BodyText();
      int expected_sequence_id;
      if (!base::StringToInt(expected_sequence_id_str, &expected_sequence_id)) {
        LOG(ERROR) << "Received invalid " << kExpectedSequenceIdTag << ": " <<
            expected_sequence_id_element->Str();
      } else {
        SetSequenceId(expected_sequence_id);
        sequence_id_recent_set_num_++;
        did_set_sequence_id = true;
      }
    }
    if (!did_set_sequence_id) {
      sequence_id_recent_set_num_ = 0;
    }
  }
}

void HeartbeatSender::SetInterval(int interval) {
  if (interval != interval_ms_) {
    interval_ms_ = interval;

    // Restart the timer with the new interval.
    if (timer_.IsRunning()) {
      timer_.Stop();
      timer_.Start(FROM_HERE, base::TimeDelta::FromMilliseconds(interval_ms_),
                   this, &HeartbeatSender::SendStanza);
    }
  }
}

void HeartbeatSender::SetSequenceId(int sequence_id) {
  sequence_id_ = sequence_id;
  // Setting the sequence ID may be a symptom of a temporary server-side
  // problem, which would affect many hosts, so don't send a new heartbeat
  // immediately, as many hosts doing so may overload the server.
  // But the server will usually set the sequence ID when it receives the first
  // heartbeat from a host. In that case, we can send a new heartbeat
  // immediately, as that only happens once per host instance.
  if (!sequence_id_was_set_) {
    ResendStanza();
  } else {
    LOG(INFO) << "The heartbeat sequence ID has been set more than once: "
              << "the new value is " << sequence_id;
    double delay = pow(2.0, sequence_id_recent_set_num_) *
        (1 + base::RandDouble()) * kResendDelayMs;
    if (delay <= interval_ms_) {
      timer_resend_.Start(FROM_HERE, base::TimeDelta::FromMilliseconds(delay),
                          this, &HeartbeatSender::ResendStanza);
    }
  }
  sequence_id_was_set_ = true;
}

scoped_ptr<XmlElement> HeartbeatSender::CreateHeartbeatMessage() {
  scoped_ptr<XmlElement> query(new XmlElement(
      QName(kChromotingXmlNamespace, kHeartbeatQueryTag)));
  query->AddAttr(QName(kChromotingXmlNamespace, kHostIdAttr), host_id_);
  query->AddAttr(QName(kChromotingXmlNamespace, kSequenceIdAttr),
                 base::IntToString(sequence_id_));
  query->AddElement(CreateSignature().release());
  return query.Pass();
}

scoped_ptr<XmlElement> HeartbeatSender::CreateSignature() {
  scoped_ptr<XmlElement> signature_tag(new XmlElement(
      QName(kChromotingXmlNamespace, kHeartbeatSignatureTag)));

  std::string message = signal_strategy_->GetLocalJid() + ' ' +
      base::IntToString(sequence_id_);
  std::string signature(key_pair_->GetSignature(message));
  signature_tag->AddText(signature);

  return signature_tag.Pass();
}

}  // namespace remoting