diff options
Diffstat (limited to 'src/components/media_manager/src/socket_streamer_adapter.cc')
-rw-r--r-- | src/components/media_manager/src/socket_streamer_adapter.cc | 312 |
1 files changed, 73 insertions, 239 deletions
diff --git a/src/components/media_manager/src/socket_streamer_adapter.cc b/src/components/media_manager/src/socket_streamer_adapter.cc index 45b1f63a17..2bb0fe10ec 100644 --- a/src/components/media_manager/src/socket_streamer_adapter.cc +++ b/src/components/media_manager/src/socket_streamer_adapter.cc @@ -33,282 +33,116 @@ #include <sys/socket.h> #include <sys/types.h> #include <sys/select.h> -#include <sys/time.h> #include <netinet/in.h> #include <arpa/inet.h> #include <netdb.h> #include <unistd.h> -#include <string.h> -#include <errno.h> -#include "config_profile/profile.h" -#include "media_manager/video/socket_video_streamer_adapter.h" #include "utils/logger.h" +#include "media_manager/socket_streamer_adapter.h" namespace media_manager { CREATE_LOGGERPTR_GLOBAL(logger, "SocketStreamerAdapter") -SocketStreamerAdapter::SocketStreamerAdapter() - : socket_fd_(0), - is_ready_(false), - streamer_(new Streamer(this)), - thread_(threads::CreateThread("SocketStreamer", streamer_)), - messages_() { -} - -SocketStreamerAdapter::~SocketStreamerAdapter() { +SocketStreamerAdapter::SocketStreamerAdapter(const std::string& ip, + const uint16_t port, + const std::string& header) + : StreamerAdapter(new SocketStreamer(this, ip, port, header)) {} + +SocketStreamerAdapter::~SocketStreamerAdapter() {} + +SocketStreamerAdapter::SocketStreamer::SocketStreamer( + SocketStreamerAdapter* const adapter, + const std::string& ip, + const uint16_t port, + const std::string& header) + : Streamer(adapter) + , ip_(ip) + , port_(port) + , header_(header) + , socket_fd_(0) + , send_socket_fd_(0) + , is_first_frame_(true) {} + +SocketStreamerAdapter::SocketStreamer::~SocketStreamer() {} + +bool SocketStreamerAdapter::SocketStreamer::Connect() { LOG4CXX_AUTO_TRACE(logger); - thread_->join(); - delete streamer_; - threads::DeleteThread(thread_); -} - -void SocketStreamerAdapter::StartActivity(int32_t application_key) { - LOG4CXX_TRACE(logger, "enter " << application_key); - - if (application_key == current_application_) { - LOG4CXX_INFO(logger, "Already running for app " << application_key); - } else { - is_ready_ = true; - current_application_ = application_key; - - messages_.Reset(); - - for (std::set<MediaListenerPtr>::iterator it = media_listeners_.begin(); - media_listeners_.end() != it; - ++it) { - (*it)->OnActivityStarted(application_key); - } - } - LOG4CXX_TRACE(logger, "exit"); -} - -void SocketStreamerAdapter::StopActivity(int32_t application_key) { - LOG4CXX_TRACE(logger, "enter " << application_key); - - if (application_key != current_application_) { - LOG4CXX_WARN(logger, "Streaming is not active for " << application_key); - } else { - is_ready_ = false; - current_application_ = 0; - - if (streamer_) { - streamer_->stop(); - messages_.Shutdown(); - } - - for (std::set<MediaListenerPtr>::iterator it = media_listeners_.begin(); - media_listeners_.end() != it; - ++it) { - (*it)->OnActivityEnded(application_key); - } - } - LOG4CXX_TRACE(logger, "exit"); -} - -bool SocketStreamerAdapter::is_app_performing_activity( - int32_t application_key) { - return (application_key == current_application_); -} - -void SocketStreamerAdapter::Init() { - LOG4CXX_DEBUG(logger, "Start sending thread"); - const size_t kStackSize = 16384; - thread_->start(threads::ThreadOptions(kStackSize)); -} - -void SocketStreamerAdapter::SendData( - int32_t application_key, - const ::protocol_handler::RawMessagePtr message) { - LOG4CXX_INFO(logger, "SendData(application_key = " << application_key << ")"); - - - if (application_key != current_application_) { - LOG4CXX_WARN(logger, "Currently working with other app " - << current_application_); - return; - } - - if (is_ready_) { - messages_.push(message); - } -} - -SocketStreamerAdapter::Streamer::Streamer( - SocketStreamerAdapter* const server) - : server_(server), - new_socket_fd_(0), - is_first_loop_(true), - is_client_connected_(false), - stop_flag_(false) { -} - -SocketStreamerAdapter::Streamer::~Streamer() { - stop(); -} - -void SocketStreamerAdapter::Streamer::threadMain() { - LOG4CXX_TRACE(logger,"enter " << this); - sync_primitives::AutoLock auto_lock(thread_lock); - start(); - - while (!stop_flag_) { - new_socket_fd_ = accept(server_->socket_fd_, NULL, NULL); - LOG4CXX_INFO(logger, "Client connectd " << new_socket_fd_); - if (0 > new_socket_fd_) { - LOG4CXX_ERROR(logger, "Socket is closed " << strerror(errno)); - sleep(1); - continue; - } - - is_client_connected_ = true; - is_first_loop_ = true; - while (is_client_connected_) { - while (!server_->messages_.empty()) { - ::protocol_handler::RawMessagePtr msg = server_->messages_.pop(); - if (!msg) { - LOG4CXX_ERROR(logger, "Null pointer message"); - continue; - } - - is_client_connected_ = send(msg); - static int32_t messages_for_session = 0; - ++messages_for_session; - - LOG4CXX_INFO(logger, "Handling map streaming message. This is " - << messages_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_, - messages_for_session); - } - } - - if (!is_ready()) { - LOG4CXX_INFO(logger, "Client disconnected."); - stop(); - break; - } - server_->messages_.wait(); - } - } - LOG4CXX_TRACE(logger,"exit " << this); -} - -void SocketStreamerAdapter::Streamer::exitThreadMain() { - LOG4CXX_TRACE(logger,"enter " << this); - stop_flag_ = true; - stop(); - server_->messages_.Shutdown(); - if (server_->socket_fd_ != -1) { - shutdown(server_->socket_fd_, SHUT_RDWR); - close(server_->socket_fd_); - } - LOG4CXX_TRACE(logger,"exit " << this); -} - -void SocketStreamerAdapter::Streamer::start() { - server_->socket_fd_ = socket(AF_INET, SOCK_STREAM, 0); - - if (0 >= server_->socket_fd_) { - LOG4CXX_ERROR_EXT(logger, "Server open error"); - return; + socket_fd_ = socket(AF_INET, SOCK_STREAM, 0); + if (0 >= socket_fd_) { + LOG4CXX_ERROR(logger, "Unable to create socket"); + return false; } int32_t optval = 1; - if (-1 == setsockopt(server_->socket_fd_, SOL_SOCKET, SO_REUSEADDR, - &optval, sizeof optval)) { - LOG4CXX_ERROR_EXT(logger, "Unable to set sockopt"); - return; + if (-1 == setsockopt( + socket_fd_, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof optval)) { + LOG4CXX_ERROR(logger, "Unable to set sockopt"); + return false; } - struct sockaddr_in serv_addr_ = { 0 }; - serv_addr_.sin_addr.s_addr = inet_addr(server_->ip_.c_str()); + struct sockaddr_in serv_addr_ = {0}; + serv_addr_.sin_addr.s_addr = inet_addr(ip_.c_str()); serv_addr_.sin_family = AF_INET; - serv_addr_.sin_port = htons(server_->port_); - - if (-1 == bind(server_->socket_fd_, + serv_addr_.sin_port = htons(port_); + if (-1 == bind(socket_fd_, reinterpret_cast<struct sockaddr*>(&serv_addr_), sizeof(serv_addr_))) { - LOG4CXX_ERROR_EXT(logger, "Unable to bind"); - return; + LOG4CXX_ERROR(logger, "Unable to bind"); + return false; } - LOG4CXX_INFO(logger, "SocketStreamerAdapter::listen for connections"); - if (-1 == listen(server_->socket_fd_, 5)) { - LOG4CXX_ERROR_EXT(logger, "Unable to listen"); - return; + if (-1 == listen(socket_fd_, 5)) { + LOG4CXX_ERROR(logger, "Unable to listen"); + return false; } -} -void SocketStreamerAdapter::Streamer::stop() { - LOG4CXX_TRACE(logger,"enter " << this); - if (0 == new_socket_fd_) { - LOG4CXX_ERROR(logger, "Client Socket does not exist: "); - } else if (-1 == shutdown(new_socket_fd_, SHUT_RDWR)) { - LOG4CXX_ERROR(logger, "Unable to shutdown socket " << strerror(errno)); - } else if (-1 == ::close(new_socket_fd_)) { - LOG4CXX_ERROR(logger, "Unable to close socket " << strerror(errno)); + send_socket_fd_ = accept(socket_fd_, NULL, NULL); + if (0 >= send_socket_fd_) { + LOG4CXX_ERROR(logger, "Unable to accept"); + return false; } - new_socket_fd_ = 0; - is_client_connected_ = false; - LOG4CXX_TRACE(logger,"exit" << this); + is_first_frame_ = true; + LOG4CXX_INFO(logger, "Client connected: " << send_socket_fd_); + return true; } -bool SocketStreamerAdapter::Streamer::is_ready() const { - bool result = true; - fd_set fds; - FD_ZERO(&fds); - FD_SET(new_socket_fd_, &fds); - struct timeval tv; - tv.tv_sec = 5; // set a 5 second timeout - tv.tv_usec = 0; - - int32_t retval = 0; - retval = select(new_socket_fd_ + 1, 0, &fds, 0, &tv); - - if (-1 == retval) { - LOG4CXX_ERROR_EXT(logger, "An error occurred"); - result = false; - } else if (0 == retval) { - LOG4CXX_ERROR_EXT(logger, "The timeout expired"); - result = false; +void SocketStreamerAdapter::SocketStreamer::Disconnect() { + LOG4CXX_AUTO_TRACE(logger); + if (0 < send_socket_fd_) { + close(send_socket_fd_); } - return result; -} - -bool SocketStreamerAdapter::Streamer::send( - const ::protocol_handler::RawMessagePtr msg) { - if (!is_ready()) { - LOG4CXX_ERROR_EXT(logger, " Socket is not ready"); - return false; + if (0 < socket_fd_) { + close(socket_fd_); } +} - if (is_first_loop_) { - is_first_loop_ = false; - char hdr[] = {"HTTP/1.1 200 OK\r\n " - "Connection: Keep-Alive\r\n" - "Keep-Alive: timeout=15, max=300\r\n" - "Server: SDL\r\n" - "Content-Type: video/mp4\r\n\r\n" - }; - - if (-1 == ::send(new_socket_fd_, hdr, strlen(hdr), MSG_NOSIGNAL)) { - LOG4CXX_ERROR_EXT(logger, " Unable to send"); +bool SocketStreamerAdapter::SocketStreamer::Send( + protocol_handler::RawMessagePtr msg) { + LOG4CXX_AUTO_TRACE(logger); + ssize_t ret; + if (is_first_frame_) { + ret = send(send_socket_fd_, header_.c_str(), header_.size(), MSG_NOSIGNAL); + if (static_cast<uint32_t>(ret) != header_.size()) { + LOG4CXX_ERROR(logger, "Unable to send data to socket"); return false; } + is_first_frame_ = false; } - if (-1 == ::send(new_socket_fd_, (*msg).data(), - (*msg).data_size(), MSG_NOSIGNAL)) { - LOG4CXX_ERROR_EXT(logger, " Unable to send"); + ret = send(send_socket_fd_, msg->data(), msg->data_size(), MSG_NOSIGNAL); + if (-1 == ret) { + LOG4CXX_ERROR(logger, "Unable to send data to socket"); return false; } - LOG4CXX_INFO(logger, "Streamer::sent " << (*msg).data_size()); + if (static_cast<uint32_t>(ret) != msg->data_size()) { + LOG4CXX_WARN(logger, + "Couldn't send all the data to socket " << send_socket_fd_); + } + + LOG4CXX_INFO(logger, "Streamer::sent " << msg->data_size()); return true; } |