socket_adaptor.cpp

/*
 * If not stated otherwise in this file or this component's Licenses.txt file the
 * following copyright and licenses apply:
 *
 * Copyright 2016 RDK Management
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
*/
#include "socket_adaptor.h"
#include "basic_types.h"
#include <signal.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <stdio.h>
#include <errno.h>
#include "safec_lib.h"
 
#define _GNU_SOURCE
#include <fcntl.h>
#include <unistd.h>
 
static const int PIPE_READ_FD = 0;
static const int PIPE_WRITE_FD = 1;
static const unsigned int MAX_CONNECTIONS = 1;
 
static bool g_one_time_init_complete = false;
 
socket_adaptor::socket_adaptor() : m_listen_fd(-1), m_write_fd(-1), m_num_connections(0), m_callback(nullptr)
{
        INFO("Enter\n")
        if(!g_one_time_init_complete)
        {
                /*SIGPIPE must be ignored or process will exit when client closes connection*/
                struct sigaction sig_settings;
                sig_settings.sa_handler = SIG_IGN;
                sig_settings.sa_flags = 0;  //CID:81611 - Initialize uninit
                sigemptyset(&sig_settings.sa_mask); //CID:81611 - Initialize uninit
                sigaction(SIGPIPE, &sig_settings, NULL);
        m_callback_data = nullptr ; //CID:80575 - Intialize a nullptr
                g_one_time_init_complete = true;
        }
        REPORT_IF_UNEQUAL(0, pipe2(m_control_pipe, O_NONBLOCK));
}
 
socket_adaptor::~socket_adaptor()
{
        INFO("Enter\n")
        stop_listening();
        close(m_control_pipe[PIPE_WRITE_FD]);
        close(m_control_pipe[PIPE_READ_FD]);
}
 
int socket_adaptor::write_data(const char * buffer, const unsigned int size)
{
        int ret = write(m_write_fd, buffer, size);
        if(0 > ret)
        {
                WARN("Write error! Closing socket. errno: 0x%x\n", errno);
                perror("socket_adaptor::data_callback() ");
 
                close(m_write_fd);
                lock();
                if(0 < m_write_fd)
                {
                        m_write_fd = -1;
                        m_num_connections--;
                }
                unlock();
        }
        else if(ret != size)
        {
                WARN("Incomplete buffer write!\n");
        }
        return ret;
}
 
std::string& socket_adaptor::get_path()
{
        return m_path;
}
 
int socket_adaptor::start_listening(const std::string &path)
{
        int ret = 0;
        m_path = path;
        errno_t rc = -1;
        /*Open new UNIX socket to transfer data*/
        m_listen_fd = socket(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK, 0); //TODO: Does it really need to be non-blocking?
        if(0 < m_listen_fd)
        {
                DEBUG("Socket created.\n");
 
                for(int retries = 0; retries <  2; retries++) //If first attempt fails due to path in use, unlink it and try again.
                {
                        struct sockaddr_un bind_path;
                        bind_path.sun_family = AF_UNIX;
                        rc = strcpy_s(bind_path.sun_path, sizeof(bind_path.sun_path), m_path.c_str());
                        if(rc != EOK)
                        {
                                ERR_CHK(rc);
                        }
                        INFO("Binding to path %s\n", bind_path.sun_path);
                        ret = bind(m_listen_fd, (const struct sockaddr *) &bind_path, sizeof(bind_path));
                        if(-1 == ret)
                        {
                                if(EADDRINUSE == errno)
                                {
                                        WARN("path is already in use. Using brute force.\n");
                                        unlink(m_path.c_str());
                                }
                                else
                                {
                                        ERROR("Failed to bind to path. Error is %d\n", errno);
                                        perror("bind error");
                                        close(m_listen_fd);
                                        m_listen_fd = -1;
                                        break;
                                }
                        }
                        else
                        {
                                INFO("Bound successfully to path.\n");
                                REPORT_IF_UNEQUAL(0, listen(m_listen_fd, 3));
                                m_thread = std::thread(&socket_adaptor::worker_thread, this);
                                break;
                        }
                }
        }
        else
        {
                ERROR("Could not open socket.\n");
                ret = -1;
        }
        return ret;
}
 
void socket_adaptor::stop_listening()
{
        INFO("Enter\n");
        lock();
 
        /*Shut down worker thread that listens to incoming connections.*/
        if(m_thread.joinable())
        {
                int message = EXIT;
                int ret = write(m_control_pipe[PIPE_WRITE_FD], &message, sizeof(message));
                if(ret != sizeof(message))
                {
                        ERROR("Couldn't trigger worker thread shutdown.\n");
                }
                else
                {
                        INFO("Waiting for worker thread to join.\n");
                        unlock();
                        m_thread.join();
                        lock();
                        INFO("Worker thread has joined.\n");
                }
        }
        if(!m_path.empty())
        {
                if(0 < m_write_fd)
                {
                        close(m_write_fd);
                        m_write_fd = -1;
                        m_num_connections--;
                        INFO("Closed write fd. Total active connections now is %d\n", m_num_connections);
                }
                INFO("Removing named socket %s.\n", m_path.c_str());
                unlink(m_path.c_str());
                m_path.clear();
        }
        unlock();
        INFO("Exit\n");
}
 
void socket_adaptor::process_new_connection()
{
        INFO("Enter\n");
        lock();
        if(0 < m_write_fd)
        {
                WARN("Trashing old write socket in favour of new one.\n");
                close(m_write_fd);
                m_write_fd = -1;
                m_num_connections--;
        }       
        
        DEBUG("Setting up a new connection.\n");
        struct sockaddr_un incoming_addr;
        int addrlen = sizeof(incoming_addr);
        m_write_fd = accept(m_listen_fd, (sockaddr *) &incoming_addr, (socklen_t *)&addrlen);
 
        if(0 > m_write_fd)
        {
                ERROR("Error accepting connection.\n");
        }
        else
        {
                m_num_connections++;
                INFO("Connected to new client.\n");
        }
        unlock();
 
        if(m_callback)
        {
                m_callback(m_callback_data);
        }
        return;
}
 
void socket_adaptor::worker_thread()
{
        INFO("Enter\n");
        int control_fd = m_control_pipe[PIPE_READ_FD];
        int max_fd = (m_listen_fd > control_fd ? m_listen_fd : control_fd);
 
 
        bool check_fds = true;
 
        while(check_fds)
        {
                fd_set poll_fd_set;
                FD_ZERO(&poll_fd_set);
                FD_SET(m_listen_fd, &poll_fd_set);
                FD_SET(control_fd, &poll_fd_set);
 
                int ret = select((max_fd + 1), &poll_fd_set, NULL, NULL, NULL);
                DEBUG("Unblocking now. ret is 0x%x\n", ret);
                if(0 == ret)
                {
                        ERROR("select() returned 0.\n");
                }
                else if(0 < ret)
                {
                        //Some activity was detected. Process event further.
                        if(0 != FD_ISSET(control_fd, &poll_fd_set))
                        {
                                control_code_t message;
                                REPORT_IF_UNEQUAL((sizeof(message)), read(m_control_pipe[PIPE_READ_FD], &message, sizeof(message)));
                                if(EXIT == message)
                                {
                                        INFO("Exiting monitor thread.\n");
                                        break;
                                }
                                else
                                {
                                        process_control_message(message);
                                }
                        }
                        if(0 != FD_ISSET(m_listen_fd, &poll_fd_set))
                        {
                                process_new_connection();
                        }
                }
                else
                {
                        ERROR("Error polling monitor FD!\n");
                        break;
                }
        }
 
        INFO("Exit\n");
}
 
void socket_adaptor::terminate_current_connection()
{
        lock();
        if(0 < m_write_fd)
        {
                close(m_write_fd);
                m_write_fd = -1;
                m_num_connections--;
                INFO("Terminated current connection.\n");
        }
        unlock();
}
 
 
void socket_adaptor::lock()
{
        m_mutex.lock();
}
 
void socket_adaptor::unlock()
{
        m_mutex.unlock();
}
 
void socket_adaptor::process_control_message(control_code_t message)
{
        if(NEW_CALLBACK == message)
        {
                lock();
                auto num_connections = m_num_connections;
                auto callback = m_callback;
                auto callback_data = m_callback_data;
                unlock();
 
                if(callback && (0 != num_connections))
                {
                        callback(callback_data);
                }
        }
}
 
 
unsigned int socket_adaptor::get_active_connections() 
{
        lock();
        unsigned int num_conn = m_num_connections;
        unlock();
        return m_num_connections;
}
 
void socket_adaptor::register_data_ready_callback(socket_adaptor_cb_t callback, void *data)
{
        lock();
        m_callback = callback;
        m_callback_data = data;
        unlock();
 
        if(nullptr != callback)
        {
                control_code_t message = NEW_CALLBACK;
                int ret = write(m_control_pipe[PIPE_WRITE_FD], &message, sizeof(message));
        }
}