[Presentation API] Change ListenForSessionMessages API to client-style.

chrome/browser/media/router/media_router_mojo_impl.cc

 // Copyright 2015 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 "chrome/browser/media/router/media_router_mojo_impl.h"
 
 #include "base/bind.h"
 #include "base/guid.h"
 #include "base/logging.h"
 #include "base/memory/scoped_vector.h"
 #include "base/observer_list.h"
 #include "base/strings/stringprintf.h"
 #include "chrome/browser/media/router/issues_observer.h"
 #include "chrome/browser/media/router/media_router_factory.h"
 #include "chrome/browser/media/router/media_router_type_converters.h"
 #include "chrome/browser/media/router/media_routes_observer.h"
 #include "chrome/browser/media/router/media_sinks_observer.h"
+#include "chrome/browser/media/router/presentation_session_messages_observer.h"
 #include "extensions/browser/process_manager.h"
 
 #define DVLOG_WITH_INSTANCE(level) \
   DVLOG(level) << "MR #" << instance_id_ << ": "
 
 #define DLOG_WITH_INSTANCE(level) DLOG(level) << "MR #" << instance_id_ << ": "
 
 namespace media_router {
 namespace {
 
 // Converts the callback result of calling Mojo CreateRoute()/JoinRoute()
 // into a local callback.
 void RouteResponseReceived(
+    const std::string& presentation_id,
     const std::vector<MediaRouteResponseCallback>& callbacks,
     interfaces::MediaRoutePtr media_route,
     const mojo::String& error_text) {
   scoped_ptr<MediaRoute> route;
+  std::string actual_presentation_id;

[mark a. foltz, 2015/08/04 23:47:02]

Is this necessary?  You could use a ternary below.

[imcheng, 2015/08/05 21:38:35]

I wanted to avoid duplicating the check for error condition in callback.Run()
below.

   std::string error;
   if (media_route.is_null()) {
     // An error occurred.
     DCHECK(!error_text.is_null());
     error = !error_text.get().empty() ? error_text.get() : "Unknown error.";
   } else {
     route = media_route.To<scoped_ptr<MediaRoute>>();
+    actual_presentation_id = presentation_id;
   }
 
   for (const MediaRouteResponseCallback& callback : callbacks)
-    callback.Run(route.get(), error);
+    callback.Run(route.get(), actual_presentation_id, error);
 }
 
 // TODO(imcheng): We should handle failure in this case. One way is to invoke
 // all pending requests with failure. (crbug.com/490787)
 void EventPageWakeComplete(bool success) {
   if (!success)
     LOG(ERROR) << "An error encountered while waking the event page.";
 }
 
 scoped_ptr<content::PresentationSessionMessage>
 ConvertToPresentationSessionMessage(interfaces::RouteMessagePtr input) {
   DCHECK(!input.is_null());
-  const auto& id_and_url = GetPresentationIdAndUrl(input->route_id);
-  const std::string& presentation_id = id_and_url.first;
-  const std::string& presentation_url = id_and_url.second;
   scoped_ptr<content::PresentationSessionMessage> output;
   switch (input->type) {
     case interfaces::RouteMessage::Type::TYPE_TEXT: {
       DCHECK(!input->message.is_null());
       DCHECK(input->data.is_null());
-      output = content::PresentationSessionMessage::CreateStringMessage(
-          presentation_url, presentation_id, make_scoped_ptr(new std::string));
-      input->message.Swap(output->message.get());
+      output.reset(new content::PresentationSessionMessage(
+          content::PresentationMessageType::TEXT));
+      input->message.Swap(&output->message);
       return output.Pass();
     }
     case interfaces::RouteMessage::Type::TYPE_BINARY: {
       DCHECK(!input->data.is_null());
       DCHECK(input->message.is_null());
-      output = content::PresentationSessionMessage::CreateArrayBufferMessage(
-          presentation_url, presentation_id,
-          make_scoped_ptr(new std::vector<uint8_t>));
+      output.reset(new content::PresentationSessionMessage(
+          content::PresentationMessageType::ARRAY_BUFFER));
+      output->data.reset(new std::vector<uint8_t>);
       input->data.Swap(output->data.get());
       return output.Pass();
     }
   }
 
   NOTREACHED() << "Invalid route message type " << input->type;
   return output.Pass();
 }
 
 }  // namespace
@@ skipping 107 matching lines @@
     const MediaSource::Id& source_id,
     const MediaSink::Id& sink_id,
     const GURL& origin,
     int tab_id,
     const std::vector<MediaRouteResponseCallback>& callbacks) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   if (!origin.is_valid()) {
     DVLOG_WITH_INSTANCE(1) << "Invalid origin: " << origin;
     for (const MediaRouteResponseCallback& callback : callbacks)
-      callback.Run(nullptr, "Invalid origin");
+      callback.Run(nullptr, "", "Invalid origin");
     return;
   }
   RunOrDefer(base::Bind(
       &MediaRouterMojoImpl::DoCreateRoute, base::Unretained(this), source_id,
       sink_id, origin.is_empty() ? "" : origin.spec(), tab_id, callbacks));
 }
 
 void MediaRouterMojoImpl::JoinRoute(
     const MediaSource::Id& source_id,
     const std::string& presentation_id,
     const GURL& origin,
     int tab_id,
     const std::vector<MediaRouteResponseCallback>& callbacks) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   if (!origin.is_valid()) {
     DVLOG_WITH_INSTANCE(1) << "Invalid origin: " << origin;
     for (const MediaRouteResponseCallback& callback : callbacks)
-      callback.Run(nullptr, "Invalid origin");
+      callback.Run(nullptr, "", "Invalid origin");
     return;
   }
   RunOrDefer(base::Bind(&MediaRouterMojoImpl::DoJoinRoute,
                         base::Unretained(this), source_id, presentation_id,
                         origin.is_empty() ? "" : origin.spec(), tab_id,
                         callbacks));
 }
 
 void MediaRouterMojoImpl::CloseRoute(const MediaRoute::Id& route_id) {
   DCHECK(thread_checker_.CalledOnValidThread());
@@ skipping 16 matching lines @@
     const MediaRoute::Id& route_id,
     scoped_ptr<std::vector<uint8>> data,
     const SendRouteMessageCallback& callback) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   RunOrDefer(base::Bind(&MediaRouterMojoImpl::DoSendSessionBinaryMessage,
                         base::Unretained(this), route_id,
                         base::Passed(data.Pass()), callback));
 }
 
-void MediaRouterMojoImpl::ListenForRouteMessages(
-    const std::vector<MediaRoute::Id>& route_ids,
-    const PresentationSessionMessageCallback& message_cb) {
-  DCHECK(thread_checker_.CalledOnValidThread());
-  RunOrDefer(base::Bind(&MediaRouterMojoImpl::DoListenForRouteMessages,
-                        base::Unretained(this), route_ids, message_cb));
-}
-
 void MediaRouterMojoImpl::ClearIssue(const Issue::Id& issue_id) {
   DCHECK(thread_checker_.CalledOnValidThread());
   issue_manager_.ClearIssue(issue_id);
   RunOrDefer(base::Bind(&MediaRouterMojoImpl::DoClearIssue,
                         base::Unretained(this), issue_id));
 }
 
 void MediaRouterMojoImpl::RegisterMediaSinksObserver(
     MediaSinksObserver* observer) {
   DCHECK(thread_checker_.CalledOnValidThread());
@@ skipping 15 matching lines @@
   // TODO(imcheng): Implement caching. (crbug.com/492451)
   observer_list->AddObserver(observer);
   RunOrDefer(base::Bind(&MediaRouterMojoImpl::DoStartObservingMediaSinks,
                         base::Unretained(this), source_id));
 }
 
 void MediaRouterMojoImpl::UnregisterMediaSinksObserver(
     MediaSinksObserver* observer) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
-  const std::string& source_id = observer->source().id();
+  const MediaSource::Id& source_id = observer->source().id();
   auto* observer_list = sinks_observers_.get(source_id);
   if (!observer_list || !observer_list->HasObserver(observer)) {
     return;
   }
 
   // If we are removing the final observer for the source, then stop
   // observing sinks for it.
   // might_have_observers() is reliable here on the assumption that this call
   // is not inside the ObserverList iteration.
   observer_list->RemoveObserver(observer);
@@ skipping 29 matching lines @@
 void MediaRouterMojoImpl::RegisterIssuesObserver(IssuesObserver* observer) {
   DCHECK(thread_checker_.CalledOnValidThread());
   issue_manager_.RegisterObserver(observer);
 }
 
 void MediaRouterMojoImpl::UnregisterIssuesObserver(IssuesObserver* observer) {
   DCHECK(thread_checker_.CalledOnValidThread());
   issue_manager_.UnregisterObserver(observer);
 }
 
+void MediaRouterMojoImpl::RegisterPresentationSessionMessagesObserver(
+    PresentationSessionMessagesObserver* observer) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(observer);
+  const MediaRoute::Id& route_id = observer->route_id();
+  auto* observer_list = messages_observers_.get(route_id);
+  if (!observer_list) {
+    observer_list = new base::ObserverList<PresentationSessionMessagesObserver>;
+    messages_observers_.add(route_id, make_scoped_ptr(observer_list));
+  } else {
+    DCHECK(!observer_list->HasObserver(observer));
+  }
+
+  bool should_listen = !observer_list->might_have_observers();
+  observer_list->AddObserver(observer);
+  if (should_listen) {
+    RunOrDefer(base::Bind(&MediaRouterMojoImpl::DoListenForRouteMessages,
+                          base::Unretained(this), route_id));
+  }
+}
+
+void MediaRouterMojoImpl::UnregisterPresentationSessionMessagesObserver(
+    PresentationSessionMessagesObserver* observer) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK(observer);
+
+  const MediaRoute::Id& route_id = observer->route_id();
+  auto* observer_list = messages_observers_.get(route_id);
+  if (!observer_list || !observer_list->HasObserver(observer))
+    return;
+
+  observer_list->RemoveObserver(observer);
+  if (!observer_list->might_have_observers())
+    messages_observers_.erase(route_id);
+  // Note that if we are still listening for messages for that route, we will
+  // stop the next time |OnRouteMessagesReceived| is invoked.
+  // TODO(imcheng): Queue a task to stop listening for messages (by asking

[mark a. foltz, 2015/08/04 23:47:02]

It seems like the scenario when an observer is unregistered is when the route is
closed or the PresentationSession goes away (not based on other actions by MRP).

[imcheng, 2015/08/05 21:38:35]

Yes. The issue here is mojo callbacks cannot be dropped, so if we no longer want
to listen for messages and there is an oustanding mojo callback for it, we have
to induce invocation of callback with empty list on exension side. When MR
receives empty list of messages, it will no longer ask the extension for next
batch of messages (i.e. create another mojo callback)

+  // the extension to invoke |OnRouteMessagesReceived| with empty list) here.
+}
+
 void MediaRouterMojoImpl::DoCreateRoute(
     const MediaSource::Id& source_id,
     const MediaSink::Id& sink_id,
     const std::string& origin,
     int tab_id,
     const std::vector<MediaRouteResponseCallback>& callbacks) {
   std::string presentation_id("mr_");
   presentation_id += base::GenerateGUID();
   DVLOG_WITH_INSTANCE(1) << "DoCreateRoute " << source_id << "=>" << sink_id
                          << ", presentation ID: " << presentation_id;
   media_route_provider_->CreateRoute(
       source_id, sink_id, presentation_id, origin, tab_id,
-      base::Bind(&RouteResponseReceived, callbacks));
+      base::Bind(&RouteResponseReceived, presentation_id, callbacks));
 }
 
 void MediaRouterMojoImpl::DoJoinRoute(
     const MediaSource::Id& source_id,
     const std::string& presentation_id,
     const std::string& origin,
     int tab_id,
     const std::vector<MediaRouteResponseCallback>& callbacks) {
   DVLOG_WITH_INSTANCE(1) << "DoJoinRoute " << source_id
                          << ", presentation ID: " << presentation_id;
   media_route_provider_->JoinRoute(
       source_id, presentation_id, origin, tab_id,
-      base::Bind(&RouteResponseReceived, callbacks));
+      base::Bind(&RouteResponseReceived, presentation_id, callbacks));
 }
 
 void MediaRouterMojoImpl::DoCloseRoute(const MediaRoute::Id& route_id) {
   DVLOG_WITH_INSTANCE(1) << "DoCloseRoute " << route_id;
   media_route_provider_->CloseRoute(route_id);
 }
 
 void MediaRouterMojoImpl::DoSendSessionMessage(
     const MediaRoute::Id& route_id,
     const std::string& message,
     const SendRouteMessageCallback& callback) {
   DVLOG_WITH_INSTANCE(1) << "SendRouteMessage " << route_id;
   media_route_provider_->SendRouteMessage(route_id, message, callback);
 }
 
 void MediaRouterMojoImpl::DoSendSessionBinaryMessage(
     const MediaRoute::Id& route_id,
     scoped_ptr<std::vector<uint8>> data,
     const SendRouteMessageCallback& callback) {
   DVLOG_WITH_INSTANCE(1) << "SendRouteBinaryMessage " << route_id;
   mojo::Array<uint8> mojo_array;
   mojo_array.Swap(data.get());
   media_route_provider_->SendRouteBinaryMessage(route_id, mojo_array.Pass(),
                                                 callback);
 }
 
 void MediaRouterMojoImpl::DoListenForRouteMessages(
-    const std::vector<MediaRoute::Id>& route_ids,
-    const PresentationSessionMessageCallback& message_cb) {
+    const MediaRoute::Id& route_id) {
   DVLOG_WITH_INSTANCE(1) << "ListenForRouteMessages";
-  media_route_provider_->ListenForRouteMessages(
-      mojo::Array<mojo::String>::From(route_ids),
-      base::Bind(&MediaRouterMojoImpl::OnRouteMessageReceived,
-                 base::Unretained(this), message_cb));
+  if (!ContainsValue(listening_messages_for_routes_, route_id)) {
+    listening_messages_for_routes_.insert(route_id);

[mark a. foltz, 2015/08/04 23:47:02]

It seems like listening_messages_for_routes_ is the same as the keys of
messages_observers_.

[imcheng, 2015/08/05 21:38:35]

They are most of the time. However adding the observer and issuing the command
to extension to start listening doesn't always occur at the same time.
|listening_messages_for_routes_| is more like "the set of route ids for which we
have an oustanding mojo callback for".

+    media_route_provider_->ListenForRouteMessages(
+        route_id, base::Bind(&MediaRouterMojoImpl::OnRouteMessagesReceived,
+                             base::Unretained(this), route_id));
+  }
 }
 
-void MediaRouterMojoImpl::OnRouteMessageReceived(
-    const PresentationSessionMessageCallback& message_cb,
+void MediaRouterMojoImpl::OnRouteMessagesReceived(
+    const MediaRoute::Id& route_id,
     mojo::Array<interfaces::RouteMessagePtr> messages) {
-  scoped_ptr<ScopedVector<content::PresentationSessionMessage>>
-      session_messages(new ScopedVector<content::PresentationSessionMessage>());
+  DVLOG(1) << "OnRouteMessagesReceived";
+
+  // Empty |messages| means there are no more messages for that route, so we
+  // can stop listening.

[mark a. foltz, 2015/08/04 23:47:02]

See my comment above, why do we need an extra protocol to manage the
registration of the observer?

What if we receive the empty message list and there are still observers?  It
seems like the next batch of messages will still go through to them...

[imcheng, 2015/08/05 21:38:35]

Okay, I reworked this a little bit. When the last observer is gone, we should
still tell the extension to invoke the callback with empty. However, when the
callback is invoked, we should only check if there are any observers to decide
if we should listen again. We only check for emptiness when we notify the
observers.

+  if (messages.storage().empty()) {
+    DVLOG(2) << "Received empty messages for " << route_id;
+    listening_messages_for_routes_.erase(route_id);
+    return;
+  }
+
+  // Check if there are any observers remaining. If not, the messages
+  // can be discarded and we can stop listening for the next batch of messages.
+  auto* observer_list = messages_observers_.get(route_id);
+  if (!observer_list) {
+    listening_messages_for_routes_.erase(route_id);
+    return;
+  }
+
+  ScopedVector<content::PresentationSessionMessage> session_messages;
+  session_messages.reserve(messages.size());
   for (size_t i = 0; i < messages.size(); ++i) {
-    session_messages->push_back(
+    session_messages.push_back(
         ConvertToPresentationSessionMessage(messages[i].Pass()).Pass());
   }
-  message_cb.Run(session_messages.Pass());
+
+  FOR_EACH_OBSERVER(PresentationSessionMessagesObserver, *observer_list,
+                    OnMessagesReceived(session_messages));
+
+  // Listen for more messages.
+  media_route_provider_->ListenForRouteMessages(
+      route_id, base::Bind(&MediaRouterMojoImpl::OnRouteMessagesReceived,
+                           base::Unretained(this), route_id));
 }
 
 void MediaRouterMojoImpl::DoClearIssue(const Issue::Id& issue_id) {
   DVLOG_WITH_INSTANCE(1) << "DoClearIssue " << issue_id;
   media_route_provider_->ClearIssue(issue_id);
 }
 
 void MediaRouterMojoImpl::DoStartObservingMediaSinks(
     const MediaSource::Id& source_id) {
   DVLOG_WITH_INSTANCE(1) << "DoStartObservingMediaSinks: " << source_id;
@@ skipping 60 matching lines @@
     return;
   }
 
   for (const auto& next_request : pending_requests_)
     next_request.Run();
 
   pending_requests_.clear();
 }
 
 }  // namespace media_router