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

chrome/browser/media/router/media_router_mojo_impl.cc

Index: chrome/browser/media/router/media_router_mojo_impl.cc
diff --git a/chrome/browser/media/router/media_router_mojo_impl.cc b/chrome/browser/media/router/media_router_mojo_impl.cc
index 976b0eeb70d8ccfb8e3a220ef0bc170bc2c08eff..7c6841a445fce7c52a945b5fbb0f2768fdaab6c2 100644
--- a/chrome/browser/media/router/media_router_mojo_impl.cc
+++ b/chrome/browser/media/router/media_router_mojo_impl.cc
@@ -14,6 +14,7 @@
 #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) \
@@ -27,10 +28,12 @@ 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;
   std::string error;
   if (media_route.is_null()) {
     // An error occurred.
@@ -38,10 +41,11 @@ void RouteResponseReceived(
     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
@@ -54,26 +58,22 @@ void EventPageWakeComplete(bool success) {
 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>));
-      input->data.Swap(output->data.get());
+      output.reset(new content::PresentationSessionMessage(
+          content::PresentationMessageType::ARRAY_BUFFER));
+      input->data.Swap(&output->data);
       return output.Pass();
     }
   }
@@ -199,7 +199,7 @@ void MediaRouterMojoImpl::CreateRoute(
   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(
@@ -218,7 +218,7 @@ void MediaRouterMojoImpl::JoinRoute(
   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,
@@ -244,14 +244,6 @@ void MediaRouterMojoImpl::SendRouteMessage(
                         base::Unretained(this), route_id, message, 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());
 
@@ -287,7 +279,7 @@ 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;
@@ -337,6 +329,44 @@ void MediaRouterMojoImpl::UnregisterIssuesObserver(IssuesObserver* observer) {
   NOTIMPLEMENTED();
 }
 
+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

[haibinlu, 2015/07/30 23:59:11]

issue stop listening right away.

[imcheng, 2015/08/03 18:56:33]

Added a TODO here (see other comment). Also changed OnRouteMessagesReceived
logic to stop listening if empty array is received.

+  // stop the next time |OnRouteMessagesReceived| is invoked.
+}
+
 void MediaRouterMojoImpl::DoCreateRoute(
     const MediaSource::Id& source_id,
     const MediaSink::Id& sink_id,
@@ -349,7 +379,7 @@ void MediaRouterMojoImpl::DoCreateRoute(
                          << ", 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(
@@ -362,7 +392,7 @@ void MediaRouterMojoImpl::DoJoinRoute(
                          << ", 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) {
@@ -379,25 +409,43 @@ void MediaRouterMojoImpl::DoSendSessionMessage(
 }
 
 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);
+    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";
+
+  // First, 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) {