uaprof.c

移动流媒体同步服务器模块

/********************************************************************
Copyright(C), 2006, Wuhan Hongxu Information Technologies Co., Ltd.
Filename   : uaprof.c
Description: uaprof program.
Author     : LiWeiBing
Version    : Initial version
Date       : 2006年12月
History    :
<author>     <time>     <version >     <desc>
********************************************************************/
#include "uaprof.h"
#include "RdsUaprofMsg.h"
#include "msspUtilReadConf.h"
#include "pssUaprofParser.h"

int SendAdaptMsgToDataBase(char *contentid, pss_uaprof_t *uaprof, char *RspMsg);
int SendQueryMsgToDataBase(char *contentid, char *useragent, char *RspMsg);
int httpOnceQuery(char  *profile, pss_uaprof_t *uaprof);

UaprofConfSt gUaprofConf;
int gLogLevel = 1; /*可以配置,取值为1,2,3; 值越小打印的信息越多*/
FILE * gUaprofPrintFP = (FILE *)stdout;

int InitSocket(char *ip, int port)
{
    int iFd;
    int opt = SO_REUSEADDR;
    struct sockaddr_in my_addr;

    if( !ip || port <= 0 )
    {
        UAPROF_LOG(ERR, "InitSocket invalid input\n");
        return -1;
    }

    if ((iFd = socket(AF_INET, SOCK_STREAM, 0)) == -1)
    {
        UAPROF_LOG(ERR, "InitSocket socket\n");
        return -1;
    }

    my_addr.sin_family = AF_INET;
    my_addr.sin_port = htons(port);
    my_addr.sin_addr.s_addr = inet_addr(ip);

    setsockopt(iFd,SOL_SOCKET,SO_REUSEADDR,&opt,sizeof(opt));

    if (bind(iFd,(struct sockaddr *)&my_addr,sizeof(struct sockaddr))== -1)
    {
        UAPROF_LOG(ERR, "InitSocket bind error!,ip=%s, port=%d\n", ip, port);
        return -1;
    }

    if (listen(iFd, 10) == -1)
    {
        UAPROF_LOG(ERR, "InitSocket listen\n");
        return -1;
    }

    return(iFd);
}


void* uaprofStartRecvThread(void * arg)
{
    int                           ret;
    int                           exitValue = 1;
    int                           recvMsgLen = 0;
    int                           clientConnSocket;
    rds_uaprof_req_t              RdsReq;
    rds_uaprof_rsp_t              RdsRsp;
    uaprof_thread_arg_t           threadArg;
    uaprof_database_query_rsp_t   QueryRspMsg;
    uaprof_database_adapt_rsp_t   AdaptRspMsg;
    pss_uaprof_t                  uaprof;

    UAPROF_LOG(TRACE, "into uaprofStartRecvThread!\n");

    if (arg == NULL)
    {
        UAPROF_LOG(ERR, "INPUT CONFIGURE DATA POINTER IS NULL!!!\n");
        pthread_exit((void *)&exitValue);
        return (void *)NULL;
    }
    memcpy(&threadArg, arg, sizeof(uaprof_thread_arg_t));
    free(arg);

    clientConnSocket = threadArg.connfd;

    for(;;)
    {
        /*收RDS请求*/
        if((recvMsgLen=recv(clientConnSocket,&RdsReq,sizeof(rds_uaprof_req_t),0)) <= 0)
        {
            UAPROF_LOG(ERR, "recv() ERROR!!!errno=%d\n", errno);
            return((void *)0);
        }

        if(recvMsgLen != sizeof(rds_uaprof_req_t))
        {
            UAPROF_LOG(WARN, "uaprof recv invalid msg!\n");
        }

        UAPROF_LOG(TRACE, "uaprof recv rds msg, contentid=%s,ua=%s\n", RdsReq.msgInfo.ContentId, RdsReq.msgInfo.userAgent);

        memset(&RdsRsp, 0, sizeof(rds_uaprof_rsp_t));
        RdsRsp.msgHdr.srcSchedulerId = RdsReq.msgHdr.dstSchedulerId;
        RdsRsp.msgHdr.dstSchedulerId = RdsReq.msgHdr.srcSchedulerId;
        RdsRsp.msgHdr.srcId = RdsReq.msgHdr.dstId;
        RdsRsp.msgHdr.dstId = RdsReq.msgHdr.srcId;
        RdsRsp.msgHdr.msgType = RDS_UAPROF_RSP;
        RdsRsp.msgHdr.msgLength = sizeof(rds_uaprof_rsp_t);
        RdsRsp.msgInfo.resultCode = 0;

        /*向数据库发查询请求*/
        if(SendQueryMsgToDataBase(RdsReq.msgInfo.ContentId, RdsReq.msgInfo.userAgent, (char *)&QueryRspMsg) != 0)
        {
            UAPROF_LOG(ERR, "SendMsgToDataBase() ERROR!!!\n");
            RdsRsp.msgInfo.resultCode = 1;
            send(clientConnSocket, &RdsRsp, RdsRsp.msgHdr.msgLength, 0);
            continue;
        }
        UAPROF_LOG(TRACE, "uaprof send query to db ok!\n");

        /*根据数据库应答进行相应处理*/
        if(memcmp(QueryRspMsg.msgHead.rsv, "01", 2) == 0)
        {
            UAPROF_LOG(TRACE, "db have no recode!return 01!\n");
            /*数据库无记录，发起单次查询*/
            if(RdsReq.msgInfo.userProfile[0] != '\0' )
            {
                 ret = httpOnceQuery(RdsReq.msgInfo.userProfile,  &uaprof);
            }
            else
            {
                ret = httpOnceQuery(RdsReq.msgInfo.userAgent,  &uaprof);
            }

            if( ret != 0 )
            {
                RdsRsp.msgInfo.resultCode = 1;
            }
            else
            {
                if( 0 != SendAdaptMsgToDataBase(RdsReq.msgInfo.ContentId, &uaprof, (char *)&AdaptRspMsg) )
                {
                    RdsRsp.msgInfo.resultCode = 1;
                }
                else
                {
                    if(memcmp(AdaptRspMsg.msgHead.rsv, "00", 2) == 0)
                    {
                        strcpy(RdsRsp.msgInfo.filename, AdaptRspMsg.filename);
                        RdsRsp.msgInfo.content_type = AdaptRspMsg.content_type;
                    }
                }
            }
        }
        else if (memcmp(QueryRspMsg.msgHead.rsv, "00", 2) == 0)
        {
            UAPROF_LOG(TRACE, "filename=%s, type=%d\n", QueryRspMsg.filename,QueryRspMsg.content_type);
            memcpy(RdsRsp.msgInfo.filename, QueryRspMsg.filename, 8);
            RdsRsp.msgInfo.content_type = QueryRspMsg.content_type;
        }
        else
        {
            UAPROF_LOG(WARN, "SendQueryMsgToDataBase db rsponse error!\n");
            RdsRsp.msgInfo.resultCode = 1;
        }

        /*向RDS发应答*/
        send(clientConnSocket, &RdsRsp, RdsRsp.msgHdr.msgLength, 0);
        UAPROF_LOG(TRACE, "uaprof send rsp to rds ok!\n");
    }
    /*TRACE信息打印*/

    pthread_exit((void *)&exitValue);
    return((void *)0);
}


int SendQueryMsgToDataBase(char *contentid, char *useragent, char *RspMsg)
{
    int ret;
    int iConnFd;
    uaprof_database_query_req_t  QueryReqMsg;

    if((contentid == NULL) || (useragent == NULL))
    {
        UAPROF_LOG(ERR, "SendQueryMsgToDataBase fail! input is NULL!\n");
        return(-1);
    }

    /*connect to db*/
    iConnFd = ConnectToServer(gUaprofConf.caDataBaseIp, gUaprofConf.iDataBasePort);
    if( iConnFd < 0 )
    {
        UAPROF_LOG(ERR, "SendQueryMsgToDataBase connect to db fail!\n");
        return(-1);
    }

    memset(&QueryReqMsg, 0, sizeof(uaprof_database_query_req_t));
    QueryReqMsg.msgHead.moduleType = STREAM_DB_MODULE_TYPE_UAPROF;
    QueryReqMsg.msgHead.msgType = UAPROF_DATABASE_QUERY_REQ;
    QueryReqMsg.msgHead.msgLength = sizeof(uaprof_database_query_req_t);
    memcpy(QueryReqMsg.ContentId, contentid, MSSP_URL_CONTENTID_LEN);
    memcpy(QueryReqMsg.userAgent, useragent, 64);

    ret = send(iConnFd, &QueryReqMsg, QueryReqMsg.msgHead.msgLength, 0);
    if(ret < 0)
    {
        UAPROF_LOG(ERR, "SendQueryMsgToDataBase send fail!\n");
        close(iConnFd);
        return(-1);
    }

    ret = recv(iConnFd, RspMsg, sizeof(uaprof_database_query_rsp_t), 0);
    if(ret <= 0)
    {
        UAPROF_LOG(ERR, "SendQueryMsgToDataBase recv fail!errno=%d\n", errno);
        close(iConnFd);
        return(-1);
    }

    close(iConnFd);

    return(0);
}



int SendAdaptMsgToDataBase(char *contentid, pss_uaprof_t *uaprof, char *RspMsg)
{
    int ret;
    int iConnFd;
    uaprof_database_adapt_req_t  AdaptReqMsg;

    UAPROF_LOG(TRACE, "uaprof into SendAdaptMsgToDataBase!\n");

    if((contentid == NULL) || (uaprof == NULL))
    {
        UAPROF_LOG(ERR, "SendQueryMsgToDataBase fail! input is NULL!\n");
        return(-1);
    }

    /*connect to db*/
    iConnFd = ConnectToServer(gUaprofConf.caDataBaseIp, gUaprofConf.iDataBasePort);
    if( iConnFd < 0 )
    {
        UAPROF_LOG(ERR, "SendQueryMsgToDataBase connect to db fail!\n");
        return(-1);
    }

    memset(&AdaptReqMsg, 0, sizeof(uaprof_database_query_req_t));
    AdaptReqMsg.msgHead.moduleType = STREAM_DB_MODULE_TYPE_UAPROF;
    AdaptReqMsg.msgHead.msgType = UAPROF_DATABASE_ADAPT_REQ;
    AdaptReqMsg.msgHead.msgLength = sizeof(uaprof_database_adapt_req_t);
    memcpy(AdaptReqMsg.ContentId, contentid, MSSP_URL_CONTENTID_LEN);
    memcpy(&AdaptReqMsg.uaprof, uaprof, sizeof(pss_uaprof_t));

    ret = send(iConnFd, &AdaptReqMsg, AdaptReqMsg.msgHead.msgLength, 0);
    if(ret < 0)
    {
        UAPROF_LOG(ERR, "SendQueryMsgToDataBase send fail!\n");
        close(iConnFd);
        return(-1);
    }

    ret = recv(iConnFd, RspMsg, sizeof(uaprof_database_adapt_rsp_t), 0);
    if(ret <= 0)
    {
        UAPROF_LOG(ERR, "SendAdaptMsgToDataBase recv fail!\n");
        close(iConnFd);
        return(-1);
    }

    close(iConnFd);

    return(0);
}

int httpOnceQuery(char  *profile, pss_uaprof_t *uaprof)
{
    int sockfd;
    struct sockaddr_in my_addr;
    char request[256], reply[1460];
    int len, recvLen;
    char *pXml;
    int  First = 0;
    char caPort[6];
    char filename[512];
    FILE *pFile;

    UAPROF_LOG(TRACE, "uaprof into httpOnceQuery!\n");

    my_addr.sin_family = AF_INET;
    my_addr.sin_port = htons(gUaprofConf.iHttpPort);
    my_addr.sin_addr.s_addr = inet_addr(gUaprofConf.caUaprofIp);

    memset(caPort, 0, 6);
    sprintf(caPort, "%d", gUaprofConf.iHttpPort);
    memset(request, 0, sizeof(request));
    strcpy(request, "GET http://");
    strcat(request, gUaprofConf.caUaprofIp);
    strcat(request, ":");
    strcat(request, caPort);
    strcat(request, "/pss6Ua.xml HTTP/1.1\r\n");
    strcat(request, "x-uahost-get-type: once\r\n");
    strcat(request, "x-wap-profile:");
    strcat(request, profile);
    strcat(request, "\r\n");
    strcat(request, "\r\n");
    len = strlen(request);
    UAPROF_LOG(TRACE, "request len=%d\n", len);
    UAPROF_LOG(TRACE, "request=%s\n", request);


    if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) == -1)
    {
        UAPROF_LOG(ERR, "socket\n");
        return(1);
    }
    if (connect(sockfd,(struct sockaddr *)&my_addr,sizeof(my_addr))== -1)
    {
        UAPROF_LOG(ERR,"connect\n");
        close(sockfd);
        return(1);
    }
    send(sockfd, request, len, 0);

    /*接收HTTP应答*/
    for(;;)
    {
        memset(reply, 0, 1460);
        recvLen = recv(sockfd, reply, 1460, 0);
        if( recvLen <= 0 )
        {
            break;
        }
        if( First ==0 )
        {
            if(memcmp(reply, "HTTP/1.1 200", 12) == 0)
            {
                pXml = strstr(reply, "<?xml");
                if( pXml == NULL )
                {
                    UAPROF_LOG(WARN, "recv is not xml!\n");
                    break;
                }
                len = recvLen - ( pXml - reply);

                memset(filename, 0, 512);
                strcpy(filename, gUaprofConf.caLocalPath);
                strcat(filename, "/");
                strcat(filename, profile);
                pFile = fopen(filename, "wb");
                if( pFile == NULL )
                {
                    UAPROF_LOG(ERR, "fopen fail!file=%s\n", filename);
                    close(sockfd);
                    return(1);
                }

                fwrite(pXml, len, 1, pFile);
                First = 1;
            }
            else
            {
                close(sockfd);
                UAPROF_LOG(WARN, "http rsp is not 200!%.12s\n", reply);
                return(1);
            }
        }
        else
        {
            fwrite(reply, recvLen, 1, pFile);
        }
    }

    fclose(pFile);
    pFile = fopen(filename, "r");
    if( pFile == NULL )
    {
        UAPROF_LOG(ERR, "fopen fail!file=%s\n", filename);
        return(1);
    }

    if( parsePssUaProf(uaprof,  pFile) != UAPROF_PARSER_OK )
    {
        UAPROF_LOG(ERR, "parsePssUaProf fail!file=%s\n", filename);
    }
    else
    {
        /*将该uaprof信息写入数据库*/
        InsertUaprofDB(gUaprofConf.caDataBaseIp, gUaprofConf.iDataBasePort,  uaprof);
        UAPROF_LOG(TRACE, "InsertUaprofDB ok!\n");
    }
    fclose(pFile);

    close(sockfd);
    sockfd = -1;

    return(0);
}

int main()
{
    pthread_t  usrThread;
    pthread_attr_t attr;
    int listenfd,connfd;
    int sin_size;
    struct sockaddr_in client;
    uaprof_thread_arg_t *ThreadArg;
    fd_set  fdR;
    int ret;
#ifdef UAPROF_RUN_IN_BACK
    int aPid;
    FILE *fp;
#endif

    memset(&gUaprofConf, 0, sizeof(UaprofConfSt));
    if(MSSP_FAILURE == uaprofReadConf(&gUaprofConf))
    {
        UAPROF_LOG(ERR,"uaprofReadConf fail!\n");
        return(-1);
    }
    gLogLevel = gUaprofConf.iLogLevel;

    if( gUaprofConf.iPrintFile )
    {
        gUaprofPrintFP = fopen("uaprof_errlog", "w");
        if( gUaprofPrintFP == NULL )
        {
              gUaprofPrintFP = (FILE *)stdout;
              UAPROF_LOG(ERR, "fopen uaprof_errlog fail!errno=%d\n", errno);
        }

    }

    #ifdef UAPROF_RUN_IN_BACK
    if((aPid=fork())<0)
    {
        UAPROF_LOG(ERR, "fork failed:%s\n",strerror(errno));
        return -1;
    }
    else if(aPid!=0)/*in parent*/
    {
        exit(-1);
    }
        
    /*child goes on...*/
    setsid();
    /*chdir("/");*/

    #endif

    listenfd = InitSocket(gUaprofConf.caHostIp, gUaprofConf.iListenPort);
    if(listenfd < 0)
    {
        UAPROF_LOG(ERR,"InitSocket\n");
        return(-1);
    }


    FD_ZERO(&fdR);
    FD_SET(listenfd, &fdR);
    sin_size = sizeof(struct sockaddr);

    for(;;)
    {
        if(select(listenfd+1, &fdR, NULL, NULL, NULL) < 0)
        {
            close(listenfd);
            UAPROF_LOG(ERR, "select error\n");
            return(-1);
        }

        if(FD_ISSET(listenfd, &fdR))
        {
            /* if sockfd is father and server socket, u can now accept() */
            ThreadArg = (uaprof_thread_arg_t *)malloc(sizeof(uaprof_thread_arg_t));
            if(!ThreadArg)
            {
                UAPROF_LOG(WARN, "malloc ERROR!!!\n");
                continue;
            }
            if((connfd = accept(listenfd,(struct sockaddr *)&client,&sin_size))==-1)
            {
                UAPROF_LOG(ERR, "accept failed.\n");
                free(ThreadArg);
                continue;
            }

            ThreadArg->connfd = connfd;
            ThreadArg->client.sin_port = ntohs(client.sin_port);
            ThreadArg->client.sin_addr.s_addr = ntohl(client.sin_addr.s_addr);

            pthread_attr_init(&attr);
            pthread_attr_setscope(&attr,PTHREAD_SCOPE_SYSTEM);
            pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_DETACHED);
            ret = pthread_create(&usrThread,&attr,uaprofStartRecvThread,(void *)ThreadArg);
            if( ret )
            {
                UAPROF_LOG(ERR, "Create Thread Failed!errno=%d,ret=%d\n", errno,ret);
                continue;
            }
        }
        continue;
    }
}