1 files changed, 55 insertions, 157 deletions

diff --git a/src/components/media_manager/src/pipe_streamer_adapter.cc b/src/components/media_manager/src/pipe_streamer_adapter.cc
index 5990428c68..a61280f057 100644
--- a/src/components/media_manager/src/pipe_streamer_adapter.cc
+++ b/src/components/media_manager/src/pipe_streamer_adapter.cc
@@ -41,190 +41,88 @@
 
 namespace media_manager {
 
-CREATE_LOGGERPTR_GLOBAL(logger_, "PipeStreamerAdapter")
+CREATE_LOGGERPTR_GLOBAL(logger, "PipeStreamerAdapter")
 
-PipeStreamerAdapter::PipeStreamerAdapter()
-  : is_ready_(false),
-    thread_(threads::CreateThread("PipeStreamer", new Streamer(this))),
-    messages_() {
-  LOG4CXX_AUTO_TRACE(logger_);
+PipeStreamerAdapter::PipeStreamerAdapter(
+    const std::string& named_pipe_path)
+  : StreamerAdapter(new PipeStreamer(this, named_pipe_path)) {
 }
 
 PipeStreamerAdapter::~PipeStreamerAdapter() {
-  LOG4CXX_AUTO_TRACE(logger_);
-
-  if ((0 != current_application_ ) && (is_ready_)) {
-    StopActivity(current_application_);
-  }
-
-  thread_->join();
-  delete thread_->delegate();
-  threads::DeleteThread(thread_);
 }
 
-void PipeStreamerAdapter::SendData(
-  int32_t application_key,
-  const ::protocol_handler::RawMessagePtr message) {
-  LOG4CXX_AUTO_TRACE(logger_);
-
-  if (application_key != current_application_) {
-    LOG4CXX_WARN(logger_, "Wrong application " << application_key);
+PipeStreamerAdapter::PipeStreamer::PipeStreamer(
+    PipeStreamerAdapter* const adapter,
+    const std::string& named_pipe_path)
+  : Streamer(adapter),
+    named_pipe_path_(named_pipe_path),
+    pipe_fd_(0) {
+  if (!file_system::CreateDirectoryRecursively(
+      profile::Profile::instance()->app_storage_folder())) {
+    LOG4CXX_ERROR(logger, "Cannot create app storage folder "
+                  << profile::Profile::instance()->app_storage_folder() );
     return;
   }
 
-  if (is_ready_) {
-    messages_.push(message);
+  if ((mkfifo(named_pipe_path_.c_str(),
+              S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH) < 0)
+      && (errno != EEXIST)) {
+    LOG4CXX_ERROR(logger, "Cannot create pipe " << named_pipe_path_);
+  } else {
+    LOG4CXX_INFO(logger, "Pipe " << named_pipe_path_
+                 << " was successfuly created");
   }
 }
 
-void PipeStreamerAdapter::StartActivity(int32_t application_key) {
-  LOG4CXX_AUTO_TRACE(logger_);
-
-  if (application_key == current_application_) {
-    LOG4CXX_WARN(logger_, "Already started activity for " << application_key);
-    return;
+PipeStreamerAdapter::PipeStreamer::~PipeStreamer() {
+  if (0 == unlink(named_pipe_path_.c_str()) ) {
+    LOG4CXX_INFO(logger, "Pipe " << named_pipe_path_ << " was removed");
+  } else {
+    LOG4CXX_ERROR(logger, "Error removing pipe " << named_pipe_path_);
   }
-
-  current_application_ = application_key;
-  is_ready_ = true;
-
-  for (std::set<MediaListenerPtr>::iterator it = media_listeners_.begin();
-       media_listeners_.end() != it;
-       ++it) {
-    (*it)->OnActivityStarted(application_key);
-  }
-
-  LOG4CXX_DEBUG(logger_, "Pipe was opened for writing " << named_pipe_path_);
 }
 
-void PipeStreamerAdapter::StopActivity(int32_t application_key) {
-  LOG4CXX_AUTO_TRACE(logger_);
-
-  if (application_key != current_application_) {
-    LOG4CXX_WARN(logger_, "Not performing activity for " << application_key);
-    return;
-  }
+bool PipeStreamerAdapter::PipeStreamer::Connect() {
+  LOG4CXX_AUTO_TRACE(logger);
 
-  is_ready_ = false;
-  current_application_ = 0;
-
-  messages_.Reset();
-
-  for (std::set<MediaListenerPtr>::iterator it = media_listeners_.begin();
-       media_listeners_.end() != it;
-       ++it) {
-    (*it)->OnActivityEnded(application_key);
-  }
-}
-
-bool PipeStreamerAdapter::is_app_performing_activity( int32_t application_key) {
-  return (application_key == current_application_);
-}
-
-void PipeStreamerAdapter::Init() {
-  LOG4CXX_AUTO_TRACE(logger_);
-  if (thread_->is_running()) {
-    thread_->stop();
-    thread_->join();
+  pipe_fd_ = open(named_pipe_path_.c_str(), O_RDWR, 0);
+  if (-1 == pipe_fd_) {
+    LOG4CXX_ERROR(logger, "Cannot open pipe for writing "
+                  << named_pipe_path_);
+    return false;
   }
-  LOG4CXX_DEBUG(logger_, "Start sending thread");
-  const size_t kStackSize = 16384;
-  thread_->start(threads::ThreadOptions(kStackSize));
-}
 
-PipeStreamerAdapter::Streamer::Streamer(
-  PipeStreamerAdapter* server)
-  : server_(server),
-    pipe_fd_(0),
-    stop_flag_(false) {
+  LOG4CXX_INFO(logger, "Pipe " << named_pipe_path_
+                << " was successfuly opened for writing");
+  return true;
 }
 
-PipeStreamerAdapter::Streamer::~Streamer() {
-  server_ = NULL;
-}
-
-void PipeStreamerAdapter::Streamer::threadMain() {
-  LOG4CXX_AUTO_TRACE(logger_);
-
-  open();
-
-  while (!stop_flag_) {
-    while (!server_->messages_.empty()) {
-      ::protocol_handler::RawMessagePtr msg = server_->messages_.pop();
-      if (!msg) {
-        LOG4CXX_ERROR(logger_, "Null pointer message");
-        continue;
-      }
-
-      ssize_t ret = write(pipe_fd_, msg.get()->data(),
-                          msg.get()->data_size());
-
-      if (ret == -1) {
-        LOG4CXX_ERROR(logger_, "Failed writing data to pipe "
-                      << server_->named_pipe_path_);
-
-        std::set<MediaListenerPtr>::iterator it =
-            server_->media_listeners_.begin();
-        for (;server_->media_listeners_.end() != it; ++it) {
-          (*it)->OnErrorReceived(server_->current_application_, -1);
-        }
-      } else if (static_cast<uint32_t>(ret) != msg.get()->data_size()) {
-        LOG4CXX_WARN(logger_, "Couldn't write all the data to pipe "
-                     << server_->named_pipe_path_);
-      }
-
-      static int32_t messsages_for_session = 0;
-      ++messsages_for_session;
-
-      LOG4CXX_DEBUG(logger_, "Handling map streaming message. This is "
-                   << messsages_for_session << " the message for "
-                   << server_->current_application_);
-      std::set<MediaListenerPtr>::iterator it =
-          server_->media_listeners_.begin();
-      for (; server_->media_listeners_.end() != it; ++it) {
-        (*it)->OnDataReceived(server_->current_application_,
-                              messsages_for_session);
-      }
-    }
-    server_->messages_.wait();
+void PipeStreamerAdapter::PipeStreamer::Disconnect() {
+  LOG4CXX_AUTO_TRACE(logger);
+  if (0 == close(pipe_fd_)) {
+    LOG4CXX_INFO(logger, "Pipe " << named_pipe_path_ << " was closed");
+  } else {
+    LOG4CXX_ERROR(logger, "Error closing pipe " << named_pipe_path_);
   }
-  close();
-}
-
-void PipeStreamerAdapter::Streamer::exitThreadMain() {
-  LOG4CXX_AUTO_TRACE(logger_);
-  stop_flag_ = true;
-  server_->messages_.Shutdown();
 }
 
-void PipeStreamerAdapter::Streamer::open() {
-  LOG4CXX_AUTO_TRACE(logger_);
-
-  DCHECK(file_system::CreateDirectoryRecursively(
-      profile::Profile::instance()->app_storage_folder()));
-
-  if ((mkfifo(server_->named_pipe_path_.c_str(),
-              S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH) < 0)
-      && (errno != EEXIST)) {
-    LOG4CXX_ERROR(logger_, "Cannot create pipe " << server_->named_pipe_path_);
-    return;
+bool PipeStreamerAdapter::PipeStreamer::Send(
+    protocol_handler::RawMessagePtr msg) {
+  LOG4CXX_AUTO_TRACE(logger);
+  ssize_t ret = write(pipe_fd_, msg->data(), msg->data_size());
+  if (-1 == ret) {
+    LOG4CXX_ERROR(logger, "Failed writing data to pipe "
+                  << named_pipe_path_);
+    return false;
   }
 
-  pipe_fd_ = ::open(server_->named_pipe_path_.c_str(), O_RDWR, 0);
-  if (-1 == pipe_fd_) {
-    LOG4CXX_ERROR(logger_, "Cannot open pipe for writing "
-                  << server_->named_pipe_path_);
-    return;
+  if (static_cast<uint32_t>(ret) != msg->data_size()) {
+    LOG4CXX_WARN(logger, "Couldn't write all the data to pipe "
+                 << named_pipe_path_);
   }
 
-  LOG4CXX_DEBUG(logger_, "Pipe " << server_->named_pipe_path_
-                << " was successfully created");
-}
-
-void PipeStreamerAdapter::Streamer::close() {
-  LOG4CXX_AUTO_TRACE(logger_);
-  ::close(pipe_fd_);
-  unlink(server_->named_pipe_path_.c_str());
+  LOG4CXX_INFO(logger, "Streamer::sent " << msg->data_size());
+  return true;
 }
 
 }  // namespace media_manager