Fix misplaced OnDataVideoStreaming (#3154)

* Fix misplaced OnDataVideoStreaming

OnDataVideoStreaming notification was sent for each application
independently. Now one would be sent from application_manager with
respect to service status

* Fix OnDataStreamingNotification

* rename parameters

* fixup! rename parameters

6 files changed, 82 insertions, 13 deletions

diff --git a/src/components/application_manager/include/application_manager/application_manager_impl.h b/src/components/application_manager/include/application_manager/application_manager_impl.h
index 438b62c797..ee4e3a05a0 100644
--- a/src/components/application_manager/include/application_manager/application_manager_impl.h
+++ b/src/components/application_manager/include/application_manager/application_manager_impl.h
@@ -117,6 +117,9 @@ class ApplicationManagerImpl;
 
 enum VRTTSSessionChanging { kVRSessionChanging = 0, kTTSSessionChanging };
 
+typedef std::map<protocol_handler::ServiceType, std::set<uint32_t> >
+    ServiceStreamingStatusMap;
+
 struct CommandParametersPermissions;
 typedef std::map<std::string, hmi_apis::Common_TransportType::eType>
     DeviceTypes;
@@ -214,6 +217,11 @@ class ApplicationManagerImpl
                          mobile_apis::HMILevel::eType from,
                          mobile_apis::HMILevel::eType to) OVERRIDE;
 
+  void ProcessOnDataStreamingNotification(
+      const protocol_handler::ServiceType service_type,
+      const uint32_t app_id,
+      const bool streaming_data_available) FINAL;
+
   void SendDriverDistractionState(ApplicationSharedPtr application);
 
   void SendGetIconUrlNotifications(const uint32_t connection_key,
@@ -1597,6 +1605,9 @@ class ApplicationManagerImpl
   std::unique_ptr<rpc_service::RPCService> rpc_service_;
   std::unique_ptr<rpc_handler::RPCHandler> rpc_handler_;
 
+  ServiceStreamingStatusMap streaming_application_services_;
+  sync_primitives::Lock streaming_services_lock_;
+
 #ifdef BUILD_TESTS
  public:
   /**
diff --git a/src/components/application_manager/src/application_impl.cc b/src/components/application_manager/src/application_impl.cc
index f66c87b47b..f44a63d5d6 100644
--- a/src/components/application_manager/src/application_impl.cc
+++ b/src/components/application_manager/src/application_impl.cc
@@ -631,7 +631,8 @@ void ApplicationImpl::SuspendStreaming(
     sync_primitives::AutoLock lock(audio_streaming_suspended_lock_);
     audio_streaming_suspended_ = true;
   }
-  MessageHelper::SendOnDataStreaming(service_type, false, application_manager_);
+  application_manager_.ProcessOnDataStreamingNotification(
+      service_type, app_id(), false);
 }
 
 void ApplicationImpl::WakeUpStreaming(
@@ -647,8 +648,8 @@ void ApplicationImpl::WakeUpStreaming(
     sync_primitives::AutoLock lock(video_streaming_suspended_lock_);
     if (video_streaming_suspended_) {
       application_manager_.OnAppStreaming(app_id(), service_type, true);
-      MessageHelper::SendOnDataStreaming(
-          ServiceType::kMobileNav, true, application_manager_);
+      application_manager_.ProcessOnDataStreamingNotification(
+          service_type, app_id(), true);
       video_streaming_suspended_ = false;
     }
     video_stream_suspend_timer_.Start(
@@ -658,8 +659,8 @@ void ApplicationImpl::WakeUpStreaming(
     sync_primitives::AutoLock lock(audio_streaming_suspended_lock_);
     if (audio_streaming_suspended_) {
       application_manager_.OnAppStreaming(app_id(), service_type, true);
-      MessageHelper::SendOnDataStreaming(
-          ServiceType::kAudio, true, application_manager_);
+      application_manager_.ProcessOnDataStreamingNotification(
+          service_type, app_id(), true);
       audio_streaming_suspended_ = false;
     }
     audio_stream_suspend_timer_.Start(
diff --git a/src/components/application_manager/src/application_manager_impl.cc b/src/components/application_manager/src/application_manager_impl.cc
index d51d4c9d06..974f125def 100644
--- a/src/components/application_manager/src/application_manager_impl.cc
+++ b/src/components/application_manager/src/application_manager_impl.cc
@@ -3479,6 +3479,46 @@ void ApplicationManagerImpl::ProcessPostponedMessages(const uint32_t app_id) {
   std::for_each(messages.begin(), messages.end(), push_allowed_messages);
 }
 
+void ApplicationManagerImpl::ProcessOnDataStreamingNotification(
+    const protocol_handler::ServiceType service_type,
+    const uint32_t app_id,
+    const bool streaming_data_available) {
+  LOG4CXX_AUTO_TRACE(logger_);
+
+  bool should_send_notification = false;
+
+  {
+    sync_primitives::AutoLock lock(streaming_services_lock_);
+    auto& active_services = streaming_application_services_[service_type];
+    should_send_notification = active_services.empty();
+    if (streaming_data_available) {
+      active_services.insert(app_id);
+      LOG4CXX_DEBUG(logger_,
+                    "Streaming session with id "
+                        << app_id << " for service "
+                        << static_cast<uint32_t>(service_type)
+                        << " was added. Currently streaming sessions count: "
+                        << active_services.size());
+    } else {
+      active_services.erase(app_id);
+      should_send_notification =
+          !should_send_notification && active_services.empty();
+
+      LOG4CXX_DEBUG(logger_,
+                    "Streaming session with id "
+                        << app_id << " for service "
+                        << static_cast<uint32_t>(service_type)
+                        << " was removed. Currently streaming sessions count: "
+                        << active_services.size());
+    }
+  }
+
+  if (should_send_notification) {
+    MessageHelper::SendOnDataStreaming(
+        service_type, streaming_data_available, *this);
+  }
+}
+
 void ApplicationManagerImpl::ProcessApp(const uint32_t app_id,
                                         const mobile_apis::HMILevel::eType from,
                                         const mobile_apis::HMILevel::eType to) {
diff --git a/src/components/application_manager/test/application_impl_test.cc b/src/components/application_manager/test/application_impl_test.cc
index 7b45d387f7..11578f19d6 100644
--- a/src/components/application_manager/test/application_impl_test.cc
+++ b/src/components/application_manager/test/application_impl_test.cc
@@ -764,16 +764,16 @@ TEST_F(ApplicationImplTest, SuspendNaviStreaming) {
   protocol_handler::ServiceType type =
       protocol_handler::ServiceType::kMobileNav;
   EXPECT_CALL(mock_application_manager_, OnAppStreaming(app_id, type, false));
-  EXPECT_CALL(*MockMessageHelper::message_helper_mock(),
-              SendOnDataStreaming(type, false, _));
+  EXPECT_CALL(mock_application_manager_,
+              ProcessOnDataStreamingNotification(type, app_id, false));
   app_impl->SuspendStreaming(type);
 }
 
 TEST_F(ApplicationImplTest, SuspendAudioStreaming) {
   protocol_handler::ServiceType type = protocol_handler::ServiceType::kAudio;
   EXPECT_CALL(mock_application_manager_, OnAppStreaming(app_id, type, false));
-  EXPECT_CALL(*MockMessageHelper::message_helper_mock(),
-              SendOnDataStreaming(type, false, _));
+  EXPECT_CALL(mock_application_manager_,
+              ProcessOnDataStreamingNotification(type, app_id, false));
   app_impl->SuspendStreaming(type);
 }
 
@@ -812,8 +812,8 @@ TEST_F(ApplicationImplTest, StopStreaming_StreamingApproved) {
   app_impl->set_video_streaming_approved(true);
 
   EXPECT_CALL(mock_application_manager_, OnAppStreaming(app_id, type, false));
-  EXPECT_CALL(*MockMessageHelper::message_helper_mock(),
-              SendOnDataStreaming(type, false, _));
+  EXPECT_CALL(mock_application_manager_,
+              ProcessOnDataStreamingNotification(type, app_id, false));
   EXPECT_CALL(*MockMessageHelper::message_helper_mock(),
               SendNaviStopStream(app_id, _));
 
@@ -824,8 +824,8 @@ TEST_F(ApplicationImplTest, StopStreaming_StreamingApproved) {
   app_impl->set_audio_streaming_approved(true);
   type = protocol_handler::ServiceType::kAudio;
   EXPECT_CALL(mock_application_manager_, OnAppStreaming(app_id, type, false));
-  EXPECT_CALL(*MockMessageHelper::message_helper_mock(),
-              SendOnDataStreaming(type, false, _));
+  EXPECT_CALL(mock_application_manager_,
+              ProcessOnDataStreamingNotification(type, app_id, false));
   EXPECT_CALL(*MockMessageHelper::message_helper_mock(),
               SendAudioStopStream(app_id, _));
 
diff --git a/src/components/include/application_manager/application_manager.h b/src/components/include/application_manager/application_manager.h
index e96101e7f1..6d5c7607cc 100644
--- a/src/components/include/application_manager/application_manager.h
+++ b/src/components/include/application_manager/application_manager.h
@@ -299,6 +299,19 @@ class ApplicationManager {
                                  mobile_apis::HMILevel::eType to) = 0;
 
   /**
+   * @brief Updates streaming service status for specified session and notifies
+   * HMI via notification if required
+   * @param service_type Id of service which status should be updated
+   * @param app_id Id of session which status should be updated
+   * @param streaming_data_available Availability of streaming data for
+   * specified session
+   */
+  virtual void ProcessOnDataStreamingNotification(
+      const protocol_handler::ServiceType service_type,
+      const uint32_t app_id,
+      const bool streaming_data_available) = 0;
+
+  /**
    * @brief Checks if driver distraction state is valid, creates message
    * which is sent to the application if allowed, otherwise it is added
    * to a list of postponed messages.
diff --git a/src/components/include/test/application_manager/mock_application_manager.h b/src/components/include/test/application_manager/mock_application_manager.h
index 2e0269cefd..92677b12bb 100644
--- a/src/components/include/test/application_manager/mock_application_manager.h
+++ b/src/components/include/test/application_manager/mock_application_manager.h
@@ -143,6 +143,10 @@ class MockApplicationManager : public application_manager::ApplicationManager {
                void(uint32_t app_id,
                     mobile_apis::HMILevel::eType from,
                     mobile_apis::HMILevel::eType to));
+  MOCK_METHOD3(ProcessOnDataStreamingNotification,
+               void(const protocol_handler::ServiceType service_type,
+                    const uint32_t app_id,
+                    const bool streaming_data_available));
   MOCK_METHOD1(
       SendHMIStatusNotification,
       void(const std::shared_ptr<application_manager::Application> app));