path: root/components/copresence/copresence_manager_impl.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/copresence/copresence_manager_impl.h"

#include <map>
#include <vector>

#include "base/bind.h"
#include "base/strings/stringprintf.h"
#include "base/time/time.h"
#include "base/timer/timer.h"
#include "components/copresence/copresence_state_impl.h"
#include "components/copresence/handlers/directive_handler_impl.h"
#include "components/copresence/handlers/gcm_handler_impl.h"
#include "components/copresence/proto/rpcs.pb.h"
#include "components/copresence/public/whispernet_client.h"
#include "components/copresence/rpc/rpc_handler.h"

using google::protobuf::RepeatedPtrField;

namespace {

const int kPollTimerIntervalMs = 3000;   // milliseconds.
const int kAudioCheckIntervalMs = 1000;  // milliseconds.

const int kQueuedMessageTimeout = 10;    // seconds.
const int kMaxQueuedMessages = 1000;

}  // namespace

namespace copresence {

bool SupportedTokenMedium(const TokenObservation& token) {
  for (const TokenSignals& signals : token.signals()) {
    if (signals.medium() == AUDIO_ULTRASOUND_PASSBAND ||
        signals.medium() == AUDIO_AUDIBLE_DTMF)
      return true;
  }
  return false;
}


// Public functions.

CopresenceManagerImpl::CopresenceManagerImpl(CopresenceDelegate* delegate)
    : delegate_(delegate),
      whispernet_init_callback_(base::Bind(
          &CopresenceManagerImpl::WhispernetInitComplete,
          // This callback gets cancelled when we are destroyed.
          base::Unretained(this))),
      init_failed_(false),
      state_(new CopresenceStateImpl),
      directive_handler_(new DirectiveHandlerImpl(
          // The directive handler and its descendants
          // will be destructed before the CopresenceState instance.
          base::Bind(&CopresenceStateImpl::UpdateDirectives,
                     base::Unretained(state_.get())))),
      poll_timer_(new base::RepeatingTimer<CopresenceManagerImpl>),
      audio_check_timer_(new base::RepeatingTimer<CopresenceManagerImpl>),
      queued_messages_by_token_(
          base::TimeDelta::FromSeconds(kQueuedMessageTimeout),
          kMaxQueuedMessages) {
  DCHECK(delegate_);
  DCHECK(delegate_->GetWhispernetClient());
  delegate_->GetWhispernetClient()->Initialize(
      whispernet_init_callback_.callback());

  MessagesCallback messages_callback = base::Bind(
      &CopresenceManagerImpl::DispatchMessages,
      // This will only be passed to objects that we own.
      base::Unretained(this));

  if (delegate->GetGCMDriver())
    gcm_handler_.reset(new GCMHandlerImpl(delegate->GetGCMDriver(),
                                          directive_handler_.get(),
                                          messages_callback));

  rpc_handler_.reset(new RpcHandler(delegate,
                                    state_.get(),
                                    directive_handler_.get(),
                                    gcm_handler_.get(),
                                    messages_callback));

  directive_handler_->Start(delegate_->GetWhispernetClient(),
                            base::Bind(&CopresenceManagerImpl::ReceivedTokens,
                                       base::Unretained(this)));
}

CopresenceManagerImpl::~CopresenceManagerImpl() {
  whispernet_init_callback_.Cancel();
}

CopresenceState* CopresenceManagerImpl::state() {
  return state_.get();
}

// Returns false if any operations were malformed.
void CopresenceManagerImpl::ExecuteReportRequest(
    const ReportRequest& request,
    const std::string& app_id,
    const std::string& auth_token,
    const StatusCallback& callback) {
  // If initialization has failed, reject all requests.
  if (init_failed_) {
    callback.Run(FAIL);
    return;
  }

  // We'll need to modify the ReportRequest, so we make our own copy to send.
  scoped_ptr<ReportRequest> request_copy(new ReportRequest(request));
  rpc_handler_->SendReportRequest(
      request_copy.Pass(), app_id, auth_token, callback);
}


// Private functions.

void CopresenceManagerImpl::WhispernetInitComplete(bool success) {
  if (success) {
    DVLOG(3) << "Whispernet initialized successfully.";
    poll_timer_->Start(FROM_HERE,
                       base::TimeDelta::FromMilliseconds(kPollTimerIntervalMs),
                       base::Bind(&CopresenceManagerImpl::PollForMessages,
                                  base::Unretained(this)));
    audio_check_timer_->Start(
        FROM_HERE, base::TimeDelta::FromMilliseconds(kAudioCheckIntervalMs),
        base::Bind(&CopresenceManagerImpl::AudioCheck, base::Unretained(this)));
  } else {
    LOG(ERROR) << "Whispernet initialization failed!";
    init_failed_ = true;
  }
}

void CopresenceManagerImpl::ReceivedTokens(
    const std::vector<AudioToken>& tokens) {
  rpc_handler_->ReportTokens(tokens);

  for (const AudioToken audio_token : tokens) {
    const std::string& token_id = audio_token.token;
    DVLOG(3) << "Heard token: " << token_id;

    // Update the CopresenceState.
    ReceivedToken token(
        token_id,
        audio_token.audible ? AUDIO_AUDIBLE_DTMF : AUDIO_ULTRASOUND_PASSBAND,
        base::Time::Now());
    state_->UpdateReceivedToken(token);

    // Deliver messages that were pre-sent on this token.
    if (queued_messages_by_token_.HasKey(token_id)) {
      // Not const because we have to remove the required tokens for delivery.
      // We're going to delete this whole vector at the end anyway.
      RepeatedPtrField<SubscribedMessage>* messages =
          queued_messages_by_token_.GetMutableValue(token_id);
      DCHECK_GT(messages->size(), 0)
          << "Empty entry in queued_messages_by_token_";

      // These messages still have their required tokens stored, and
      // DispatchMessages() will still check for them. If we don't remove
      // the tokens before delivery, we'll just end up re-queuing the message.
      for (SubscribedMessage& message : *messages)
        message.mutable_required_token()->Clear();

      DVLOG(3) << "Delivering " << messages->size()
               << " message(s) pre-sent on token " << token_id;
      DispatchMessages(*messages);

      // The messages have been delivered, so we don't need to keep them
      // in the queue. Note that the token will still be reported
      // to the server (above), so we'll keep getting the message.
      // But we can now drop our local copy of it.
      int erase_count = queued_messages_by_token_.Erase(token_id);
      DCHECK_GT(erase_count, 0);
    }
  }
}

void CopresenceManagerImpl::AudioCheck() {
  if (!directive_handler_->GetCurrentAudioToken(AUDIBLE).empty() &&
      !directive_handler_->IsAudioTokenHeard(AUDIBLE)) {
    delegate_->HandleStatusUpdate(AUDIO_FAIL);
  } else if (!directive_handler_->GetCurrentAudioToken(INAUDIBLE).empty() &&
             !directive_handler_->IsAudioTokenHeard(INAUDIBLE)) {
    delegate_->HandleStatusUpdate(AUDIO_FAIL);
  }
}

// Report our currently playing tokens to the server.
void CopresenceManagerImpl::PollForMessages() {
  const std::string& audible_token =
      directive_handler_->GetCurrentAudioToken(AUDIBLE);
  const std::string& inaudible_token =
      directive_handler_->GetCurrentAudioToken(INAUDIBLE);

  std::vector<AudioToken> tokens;
  if (!audible_token.empty())
    tokens.push_back(AudioToken(audible_token, true));
  if (!inaudible_token.empty())
    tokens.push_back(AudioToken(inaudible_token, false));

  if (!tokens.empty())
    rpc_handler_->ReportTokens(tokens);
}

void CopresenceManagerImpl::DispatchMessages(
    const RepeatedPtrField<SubscribedMessage>& messages) {
  if (messages.size() == 0)
    return;

  // Index the messages by subscription id.
  std::map<std::string, std::vector<Message>> messages_by_subscription;
  DVLOG(3) << "Processing " << messages.size() << " received message(s).";
  int immediate_message_count = 0;
  for (const SubscribedMessage& message : messages) {
    // If tokens are required for this message, queue it.
    // Otherwise stage it for delivery.
    if (message.required_token_size() > 0) {
      int supported_token_count = 0;
      for (const TokenObservation& token : message.required_token()) {
        if (SupportedTokenMedium(token)) {
          if (!queued_messages_by_token_.HasKey(token.token_id())) {
            queued_messages_by_token_.Add(
                token.token_id(), RepeatedPtrField<SubscribedMessage>());
          }
          RepeatedPtrField<SubscribedMessage>* queued_messages =
              queued_messages_by_token_.GetMutableValue(token.token_id());
          DCHECK(queued_messages);
          queued_messages->Add()->CopyFrom(message);
          supported_token_count++;
        }
      }

      if (supported_token_count > 0) {
        DVLOG(3) << "Queued message under " << supported_token_count
                 << "token(s).";
      } else {
        VLOG(2) << "Discarded message that requires one of "
                << message.required_token_size()
                << " token(s), all on unsupported mediums.";
      }
    } else {
      immediate_message_count++;
      for (const std::string& subscription_id : message.subscription_id()) {
        messages_by_subscription[subscription_id].push_back(
            message.published_message());
      }
    }
  }

  // Send the messages for each subscription.
  DVLOG(3) << "Dispatching " << immediate_message_count << "message(s) for "
           << messages_by_subscription.size() << " subscription(s).";
  for (const auto& map_entry : messages_by_subscription) {
    // TODO(ckehoe): Once we have the app ID from the server, we need to pass
    // it in here and get rid of the app id registry from the main API class.
    const std::string& subscription = map_entry.first;
    const std::vector<Message>& messages = map_entry.second;
    delegate_->HandleMessages(std::string(), subscription, messages);
  }
}

}  // namespace copresence