Adding support for pre-sent messages

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 <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"
 
 namespace {
 
 const int kPollTimerIntervalMs = 3000;   // milliseconds.
 const int kAudioCheckIntervalMs = 1000;  // milliseconds.
 
 }  // 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>) {
   DCHECK(delegate_);
   DCHECK(delegate_->GetWhispernetClient());
   delegate_->GetWhispernetClient()->Initialize(
       whispernet_init_callback_.callback());
 
+  MessagesCallback messages_callback = base::Bind(
+      &CopresenceManagerImpl::DispatchMessages<
+          google::protobuf::RepeatedPtrField<SubscribedMessage>>,
+      base::Unretained(this));

[rkc, 2015/01/07 21:31:43]

Add a one liner about why unretained is safe here.

[Charlie, 2015/01/07 22:06:07]

Done.

+
   if (delegate->GetGCMDriver())
     gcm_handler_.reset(new GCMHandlerImpl(delegate->GetGCMDriver(),
-                                          directive_handler_.get()));
+                                          directive_handler_.get(),
+                                          messages_callback));
 
   rpc_handler_.reset(new RpcHandler(delegate,
                                     state_.get(),
                                     directive_handler_.get(),
-                                    gcm_handler_.get()));
+                                    gcm_handler_.get(),
+                                    messages_callback));
 }
 
 CopresenceManagerImpl::~CopresenceManagerImpl() {
   whispernet_init_callback_.Cancel();
 }
 
 CopresenceState* CopresenceManagerImpl::state() {
   return state_.get();
 }
 
@@ skipping 37 matching lines @@
   } else {
     LOG(ERROR) << "Whispernet initialization failed!";
     init_failed_ = true;
   }
 }
 
 void CopresenceManagerImpl::ReceivedTokens(
     const std::vector<AudioToken>& tokens) {
   rpc_handler_->ReportTokens(tokens);
 
-  // Update the CopresenceState.
   for (const AudioToken audio_token : tokens) {
-    DVLOG(3) << "Heard token: " << audio_token.token;
+    const std::string& token_id = audio_token.token;
+    DVLOG(3) << "Heard token: " << token_id;
+
+    // Update the CopresenceState.
     ReceivedToken token(
-        audio_token.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_.count(token_id) > 0) {
+      // Not const because we have to remove the required tokens for delivery.
+      // We're going to delete this whole vector at the end anyway.
+      std::vector<SubscribedMessage>& messages =
+          queued_messages_by_token_[token_id];
+      DCHECK(!messages.empty()) << "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);
+
+      int erase_count = queued_messages_by_token_.erase(token_id);

[rkc, 2015/01/07 21:31:43]

Is this the behavior we want? Currently for non-present messages, an app will
get the same message every time the machine hears a token. This will make it so
that the app will get the message just once, after which it won't get it on
subsequent times that the token is heard, which is completely different
behavior.

[Charlie, 2015/01/07 22:06:07]

The token will still be reported to the server and the message delivered
normally. Added a clarifying comment.

+      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() {
-  // Report our currently playing tokens.
   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::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);
+template<typename IterableSubscribedMessages>
+void CopresenceManagerImpl::DispatchMessages(
+    const IterableSubscribedMessages& 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)) {
+          queued_messages_by_token_[token.token_id()].push_back(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