src/lynq/lib/liblynq-shm/lynq_urc_socket.cpp

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
//#include <binder/Parcel.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <errno.h>
#include <stdbool.h>
#include <sys/types.h>
#include <unistd.h>
#include "lynq_urc_socket.h"
#include "log/log.h"
#include <list>

static int lynq_urc_socket_fd = 0;
static int registered_urc_socket_port = 0;
static int registered_urc_socket_fd = 0;

static pthread_mutex_t s_urc_broadcast_mtx = PTHREAD_MUTEX_INITIALIZER;
static std::list<struct sockaddr*> s_urc_broadcast_clients;

static pthread_mutex_t s_urc_register_mtx = PTHREAD_MUTEX_INITIALIZER;

#define LYNQ_URC_SERVER_LISTEN_PORT 8086
#define LYNQ_URC_SERVER_LISTEN_ADDRESS "127.0.0.1"

#define LYNQ_URC_CLIENT_PORT_START 7050
#define LYNQ_URC_CLIENT_PORT_END 7099
#define LYNQ_URC_CLIENT_LISTEN_ADDRESS "127.0.0.1"

static void * receive_client_heartbeat(void *parm);
static void find_all_client_to_notify_online();
int lynq_open_broadcast_urc_socket()
{
    int ret = 0;
    int on = 1;
    pthread_mutex_lock(&s_urc_broadcast_mtx);

    struct sockaddr_in addr_serv;
    //Creating a Socket object
    lynq_urc_socket_fd=socket(AF_INET,SOCK_DGRAM,0);
    memset(&addr_serv, 0, sizeof(addr_serv));
    addr_serv.sin_family =AF_INET;
    addr_serv.sin_port = htons(LYNQ_URC_SERVER_LISTEN_PORT);
    addr_serv.sin_addr.s_addr = inet_addr(LYNQ_URC_SERVER_LISTEN_ADDRESS);
    ret = setsockopt(lynq_urc_socket_fd,SOL_SOCKET,SO_REUSEADDR,&on,sizeof on);
    if (ret < 0){
        goto open_socket_exit;
    }
    ret = bind(lynq_urc_socket_fd,(struct sockaddr*)&addr_serv,sizeof(addr_serv));
    if (ret < 0){
        goto open_socket_exit;
    }
    //receive registion and display on at port
    pthread_t thid;
    if(pthread_create(&thid, NULL, receive_client_heartbeat, NULL) != 0) {
        RLOGE("thread creation failed\n");
        ret = -1;
        goto open_socket_exit;
    }

    find_all_client_to_notify_online();

open_socket_exit:
    pthread_mutex_unlock(&s_urc_broadcast_mtx);
    return ret == 0 ? lynq_urc_socket_fd: -1;
}

int lynq_broadcast_urc_msg(void * msg, int size)
{
    int all_ok = 1;
    pthread_mutex_lock(&s_urc_broadcast_mtx);
    for(auto it=s_urc_broadcast_clients.begin(); it != s_urc_broadcast_clients.end(); )
    {
        int len = sendto(lynq_urc_socket_fd, msg, size, 0, *it, sizeof(struct sockaddr_in));
        if (len < 0)
        {
            struct sockaddr_in *addr_cli = (struct sockaddr_in *) *it;
            printf("remove client %d now\n", ntohs(addr_cli->sin_port));
            it = s_urc_broadcast_clients.erase(it);
            free(addr_cli);
            all_ok = 0;
            continue;
        }
        it++;
    }
    pthread_mutex_unlock(&s_urc_broadcast_mtx);
    return all_ok;
}

int lynq_register_urc_socket()
{
    int on = 0;
    int ret = 0;
    int len_s;
    struct sockaddr_in addr_serv;
    //Creating a Socket object
    pthread_mutex_lock(&s_urc_register_mtx);
    registered_urc_socket_fd=socket(AF_INET,SOCK_DGRAM,0);
    memset(&addr_serv, 0, sizeof(addr_serv));
    addr_serv.sin_family =AF_INET;
    addr_serv.sin_addr.s_addr = inet_addr(LYNQ_URC_CLIENT_LISTEN_ADDRESS);
    if (registered_urc_socket_port > 0) {
        on = 1;
        ret = setsockopt(registered_urc_socket_fd,SOL_SOCKET,SO_REUSEADDR,&on,sizeof on);
        addr_serv.sin_port = htons(registered_urc_socket_port);
        ret = bind(registered_urc_socket_fd,(struct sockaddr*)&addr_serv,sizeof(addr_serv));
        if (ret < 0){
            goto register_socket_exit;
        }
    }
    else
    {
        for(registered_urc_socket_port = LYNQ_URC_CLIENT_PORT_START; registered_urc_socket_port < LYNQ_URC_CLIENT_PORT_END; registered_urc_socket_port ++)
        {
            addr_serv.sin_port = htons(registered_urc_socket_port);
            ret = bind(registered_urc_socket_fd,(struct sockaddr*)&addr_serv,sizeof(addr_serv));
            if (ret < 0){
                printf("bind socket %d fail\n", registered_urc_socket_port);
                continue;
            }
            break;
        }
    }

    addr_serv.sin_port = htons(LYNQ_URC_SERVER_LISTEN_PORT);
    len_s = sendto(registered_urc_socket_fd, &registered_urc_socket_port, sizeof(registered_urc_socket_port), 0, (struct sockaddr *)&addr_serv,sizeof(addr_serv));
    printf("send result %d %d\n", len_s, errno);
register_socket_exit:
    pthread_mutex_unlock(&s_urc_register_mtx);
    printf("init result %d\n", ret);
    return ret ==0 ? registered_urc_socket_fd : -1;
}

int lynq_recv_urc_msg(void * buffer, int size)
{
    int ret = 0;
    struct sockaddr_in addr_serv;
    socklen_t len_addr_serv = sizeof(addr_serv);
    memset(&addr_serv, 0, sizeof(addr_serv));
    printf("client %ul to recv now\n", pthread_self());
    pthread_mutex_lock(&s_urc_register_mtx);
    if (registered_urc_socket_fd <= 0) {
        pthread_mutex_unlock(&s_urc_register_mtx);
        return -1;
    }
    while(1)
    {
        memset(buffer, 0, 4);
        ret = recvfrom(registered_urc_socket_fd, buffer, size,0,(struct sockaddr *)&addr_serv,(socklen_t*)&len_addr_serv);
        if (ret > 0 && LYNQ_URC_SERVER_LISTEN_PORT == ntohs(addr_serv.sin_port))
        {
            if (ret == 4 && LYNQ_URC_SERVER_LISTEN_PORT == *((int*)buffer))
            {
                sendto(registered_urc_socket_fd, &registered_urc_socket_port, sizeof(registered_urc_socket_port), 0, (struct sockaddr *)&addr_serv,sizeof(addr_serv));
                printf("to send ack of headbeat\n");
                continue;
            }
        }
        else
        {
            printf("not a server packet\n");
            continue;
        }
        break;
    }
    pthread_mutex_unlock(&s_urc_register_mtx);
    printf("client %ul recv %d\n", pthread_self(), ret);
    return ret;
}

static void *receive_client_heartbeat(void *parm)
{
    RLOGE("receive_at thread start\n");
    int recv = 0;
    char buffer[1024] = {0};
    while(1)
    {
        RLOGE("receive third at cmd\n");

        socklen_t len = sizeof(struct sockaddr_in);
        struct sockaddr* cli = (struct sockaddr*)malloc(len);
        bzero(cli, len);
        printf("to recv now\n");
        recv = recvfrom(lynq_urc_socket_fd, buffer, 1024, 0, cli, &len);
        if(recv < 0)
        {
            RLOGE("recv fail\n");
            printf("recv fail\n");
            free(cli);
            continue;
        }
        printf("recv a packet %d \n", recv);
        struct sockaddr_in * register_socket = (struct sockaddr_in*)cli;

        int port = ntohs(register_socket->sin_port);
        if (recv == 4 && port == *((int*)buffer))
        {
            printf("port is %d\n", port);
            pthread_mutex_lock(&s_urc_broadcast_mtx);
            s_urc_broadcast_clients.push_back(cli);
            pthread_mutex_unlock(&s_urc_broadcast_mtx);
            RLOGE("recvfrom from client\n");
        }
    }
    return NULL;
}
static int exec_cmd(const char *str_cmd, char * str_cmd_ret, size_t max_len)
{
    FILE *fp;
    //printf("to exec cmd:%s\n", str_cmd);
    if((fp=popen(str_cmd,"r"))==NULL)
    {
        perror("popen error!");
        return -1;
    }
    if((fread(str_cmd_ret,max_len,1,fp))<0)
    {
        perror("fread fail!");
        fclose(fp);
        return -1;
    }
    fclose(fp);
    return 0;
}

static int lynq_split(char * str, int len, char delimiter, char * results[]) {
    int ret = 0;
    char * end = str + len - 1;
    results[ret++] = str;
    while(str < end) {
        if (*str == delimiter) {
            *str++ = '\0';
            results[ret++] = str;
            continue;
        }
        str++;
    }
    if (*str == delimiter) {
        *str = '\0';
    }

    return ret;
}

static void find_all_client_to_notify_online()
{
    char buffer[1024] = {0};
    char * split_lines[128]= {0};
    int count = 0;
    int port, len;
    int local_port = LYNQ_URC_SERVER_LISTEN_PORT;
    printf("find_all_client_to_notify_online\n");
    if (exec_cmd("netstat -an | grep -e \"127\\.0\\.0\\.1:70[5-9][0-9]\" | awk '{print $4}' | awk -F \":\" '{print $2}'", buffer, 1024) == 0)
    {
         count = lynq_split(buffer, 1024, '\n', split_lines);
         for(int i=0; i < count;i++)
         {
             port = atoi(split_lines[i]);
             printf("got %s port:%d\n", split_lines[i], port);
             if (port < LYNQ_URC_CLIENT_PORT_START || port > LYNQ_URC_CLIENT_PORT_END)
                 continue;

             struct sockaddr_in addr_serv;
             memset(&addr_serv, 0, sizeof(addr_serv));
             addr_serv.sin_family =AF_INET;
             addr_serv.sin_addr.s_addr = inet_addr(LYNQ_URC_CLIENT_LISTEN_ADDRESS);
             addr_serv.sin_port = htons(port);
             len = sendto(lynq_urc_socket_fd, &local_port, sizeof(local_port), 0, (struct sockaddr *)&addr_serv,sizeof(addr_serv));
             printf("send result %d %d\n", len, errno);
         }
    }
    else
    {
        printf("net stat fail\n");
    }
}