[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;
   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);
+  // TODO(imcheng): Queue a task to stop listening for messages by asking
+  // the extension to invoke the oustanding Mojo callback with empty list. We
+  // don't want the Mojo callback to exist indefinitely on the extension side
+  // and there is currently no way to cancel the callback from this side.
+}
+
 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(route_ids_listening_for_messages_, route_id)) {
+    route_ids_listening_for_messages_.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>());
-  for (size_t i = 0; i < messages.size(); ++i) {
-    session_messages->push_back(
-        ConvertToPresentationSessionMessage(messages[i].Pass()).Pass());
+  DVLOG(1) << "OnRouteMessagesReceived";
+
+  // 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) {
+    route_ids_listening_for_messages_.erase(route_id);
+    return;
   }
-  message_cb.Run(session_messages.Pass());
+
+  // Empty |messages| means we told the extension that we were no longer
+  // listening for messages on that route. But since now we have observers
+  // again, we should keep listening.
+  if (messages.storage().empty()) {
+    DVLOG(2) << "Received empty messages for " << route_id;
+  } else {
+    ScopedVector<content::PresentationSessionMessage> session_messages;
+    session_messages.reserve(messages.size());
+    for (size_t i = 0; i < messages.size(); ++i) {
+      session_messages.push_back(
+          ConvertToPresentationSessionMessage(messages[i].Pass()).Pass());
+    }
+
+    // TODO(imcheng): If there is only 1 observer, we should be able to pass
+    // the messages to avoid additional copies. (crbug.com/517234)
+    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