src/network/TCPServer.cxx

   1 /***************************************************************************
   2  *   Copyright (C) 2007 by Prabakaran Thirumalai   *
   3  *   praba_tuty@yahoo.com   *
   4  *                                                                         *
   5  *   This program is free software; you can redistribute it and/or modify  *
   6  *   it under the terms of the GNU General Public License as published by  *
   7  *   the Free Software Foundation; either version 2 of the License, or     *
   8  *   (at your option) any later version.                                   *
   9  *                                                                         *
  10  *   This program is distributed in the hope that it will be useful,       *
  11  *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
  12  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
  13  *   GNU General Public License for more details.                          *
  14  *                                                                         *
  15  *   You should have received a copy of the GNU General Public License     *
  16  *   along with this program; if not, write to the                         *
  17  *   Free Software Foundation, Inc.,                                       *
  18  *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.             *
  19  ***************************************************************************/
  20 #include <CSql.h>
  21 #include <Network.h>
  22 #include <SqlNetworkHandler.h>
  23
  24 DbRetVal TCPServer::start()
  25 {
  26     DbRetVal rv = OK;
  27     if (port == 0 )
  28     {
  29         printError(ErrBadArg, "Set the port first before starting\n");
  30         return ErrBadArg;
  31     }
  32     struct sockaddr_in my_addr;
  33     if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
  34        printError(ErrOS, "Unable to create socket\n");
  35        return ErrOS;
  36     }
  37     my_addr.sin_family = AF_INET;
  38     my_addr.sin_port = htons(port);
  39     my_addr.sin_addr.s_addr = INADDR_ANY;
  40     memset(&(my_addr.sin_zero), '\0', 8);
  41     if (bind(sockfd, (struct sockaddr *)&my_addr,
  42                      sizeof(struct sockaddr)) == -1) {
  43         printError(ErrOS, "bind failed");
  44         return ErrOS;
  45     }
  46     if (listen(sockfd, 10) == -1) {
  47         printError(ErrOS, "listen failed");
  48         return ErrOS;
  49     }
  50     return rv;
  51
  52 }
  53 DbRetVal TCPServer::stop()
  54 {
  55     DbRetVal rv = OK;
  56     close (sockfd);
  57     return rv;
  58 }
  59 DbRetVal TCPServer::handleClient()
  60 {
  61 //   printf("PRABA::handling client \n");
  62    DbRetVal rv = OK;
  63    socklen_t addressLen = sizeof(struct sockaddr);
  64    clientfd = accept(sockfd, (struct sockaddr*) &clientAddress, &addressLen);
  65    int ret = os::fork();
  66    if (ret) {
  67        //parent
  68        os::signal(SIGCHLD, SIG_IGN);
  69        close(clientfd);
  70        return OK;
  71    }else if (ret == 0) {
  72        //child
  73        int response = 1;
  74        int ret = os::send(clientfd, &response, 4, 0);
  75        if (ret == -1)
  76        {
  77            printError(ErrOS, "Unable to communicate to peer\n");
  78            return ErrOS;
  79        }
  80 //       printf("sent connect ack packet to client\n");
  81        fd_set fdset;
  82        struct timeval timeout;
  83        SqlNetworkHandler handler;
  84        PacketHeader header;
  85        while(true) {
  86            FD_ZERO(&fdset);
  87            FD_SET(clientfd, &fdset);
  88            timeout.tv_sec = 5;
  89            timeout.tv_usec = 0;
  90            int ret = os::select(clientfd+1, &fdset, 0, 0, &timeout);
  91            if (ret > 0) {
  92 //                printf("something in fd\n");
  93                int numbytes = os::recv(clientfd,&header,sizeof(PacketHeader),0);
  94                if (numbytes == -1)
  95                {
  96                    printError(ErrOS, "Error reading from socket\n");
  97                    return ErrOS;
  98                }
  99 //               printf("HEADER says packet type is %d\n", header.packetType);
 100                char *buffer = NULL;
 101                if (header.packetType != SQL_NW_PKT_DISCONNECT &&
 102                    header.packetType != SQL_NW_PKT_COMMIT     &&
 103                    header.packetType != SQL_NW_PKT_ROLLBACK   &&
 104                    header.packetType != SQL_NW_PKT_SHOWTABLES )
 105                {
 106                    buffer = (char*) malloc(header.packetLength);
 107                    numbytes = os::recv(clientfd,buffer,header.packetLength,0);
 108                    if (numbytes == -1)
 109                    {
 110                        printError(ErrOS, "Error reading from socket\n");
 111                        return ErrOS;
 112                    }
 113                }
 114                ResponsePacket *rpkt = (ResponsePacket *) handler.process(header, buffer);
 115                numbytes = os::send(clientfd, rpkt, sizeof(ResponsePacket), 0);
 116                if (numbytes == -1)
 117                {
 118                    printError(ErrOS, "Error writing to socket\n");
 119                    return ErrOS;
 120                }
 121                char *ptr = (char *)&rpkt->retVal;
 122                if (*ptr==0) continue;
 123                if (*(ptr + 1)  == 1) continue; // for end of fetch
 124                NetworkStmt *stmt=NULL;
 125                int params =  *(ptr + 2);
 126                int proj = *(ptr + 3);
 127                if ((header.packetType == SQL_NW_PKT_PREPARE && params != 0) ||
 128                    (header.packetType == SQL_NW_PKT_PREPARE && proj != 0)) {
 129                    if (params) {
 130                        SqlPacketParamMetadata *prmpkt = new SqlPacketParamMetadata();
 131                        prmpkt->stmtID = rpkt->stmtID;
 132                        ListIterator stmtIter = SqlNetworkHandler::stmtList.getIterator();
 133                        while (stmtIter.hasElement()) {
 134                            stmt = (NetworkStmt*) stmtIter.nextElement();
 135                            if (stmt->stmtID == prmpkt->stmtID) break;
 136                        }
 137                        prmpkt->noParams = stmt->paramList.size();
 138                        rv = prmpkt->marshall();
 139                        if (rv != OK) {
 140                            printf("marshall failed\n");
 141                        }
 142                        rv = send(SQL_NW_PKT_PARAM_METADATA, prmpkt->getMarshalledBuffer(), prmpkt->getBufferSize());
 143                        if (rv != OK) {
 144                            printf("Error in sending the metadata to the client\n");
 145                            exit(1);
 146                        }
 147                    }
 148                    if (proj) {
 149                        //fill projection list and send it to client
 150                        SqlPacketProjMetadata *prjpkt = new SqlPacketProjMetadata();
 151                        prjpkt->stmtID = rpkt->stmtID;
 152                        ListIterator stmtIter = SqlNetworkHandler::stmtList.getIterator();
 153                        while (stmtIter.hasElement()) {
 154                            stmt = (NetworkStmt*) stmtIter.nextElement();
 155                            if (stmt->stmtID == prjpkt->stmtID) break;
 156                        }
 157                        prjpkt->noProjs = stmt->projList.size();
 158                        rv = prjpkt->marshall();
 159                        if (rv != OK) {
 160                            printf("marshall failed\n");
 161                        }
 162                        rv = send(SQL_NW_PKT_PROJ_METADATA, prjpkt->getMarshalledBuffer(), prjpkt->getBufferSize());
 163                        if (rv != OK) {
 164                            printf("Error in sending the metadata to the client\n");
 165                            exit(1);
 166                        }
 167                    }
 168                }
 169                if (header.packetType == SQL_NW_PKT_FETCH) {
 170                    SqlPacketResultSet *rspkt = new SqlPacketResultSet();
 171                    rspkt->stmtID = rpkt->stmtID;
 172                    ListIterator stmtIter = SqlNetworkHandler::stmtList.getIterator();
 173                    while (stmtIter.hasElement()) {
 174                        stmt = (NetworkStmt*) stmtIter.nextElement();
 175                        if (stmt->stmtID == rspkt->stmtID) break;
 176                    }
 177                    rspkt->noProjs = stmt->projList.size();
 178                    rspkt->setProjList(stmt->projList);
 179                    rspkt->marshall();
 180                    if (rv != OK) { printf("marshall failed\n"); }
 181                    rv = send(SQL_NW_PKT_RESULT_SET, rspkt->getMarshalledBuffer(), rspkt->getBufferSize());
 182                    if (rv != OK) {
 183                        printf("Error in sending the metadata to the client\n");
 184                        exit(1);
 185                    }
 186                }
 187                if (header.packetType == SQL_NW_PKT_SHOWTABLES) {
 188                    SqlPacketShowTables *shTblPkt = new SqlPacketShowTables();
 189                    shTblPkt->stmtID = rpkt->stmtID;
 190                    shTblPkt->numOfTables = rpkt->rows;
 191                    ListIterator stmtIter = SqlNetworkHandler::stmtList.getIterator();
 192                    while (stmtIter.hasElement()) {
 193                        stmt = (NetworkStmt*) stmtIter.nextElement();
 194                        if (stmt->stmtID == shTblPkt->stmtID) break;
 195                    }
 196                    rv = shTblPkt->marshall();
 197                    if (rv != OK) { printf("marshall failed\n"); }
 198                    rv = send(SQL_NW_PKT_SHOWTABLES, shTblPkt->getMarshalledBuffer(), shTblPkt->getBufferSize());
 199                    if (rv != OK) {
 200                        printf("Error in sending the metadata to the client\n");
 201                        exit(1);
 202                    }
 203                }
 204                if (header.packetType == SQL_NW_PKT_DISCONNECT) {
 205                    exit(0);
 206                }
 207            } //else printf("Nothing in fd %d\n", ret);
 208        }
 209    }else {
 210         printError(ErrOS, "Unable to fork new process");
 211         return ErrOS;
 212    }
 213    return OK;
 214 }
 215
 216 DbRetVal TCPServer::send(NetworkPacketType type, char *buf, int len)
 217 {
 218     DbRetVal rv = OK;
 219     void* totalBuffer = malloc(sizeof(PacketHeader)+ len);
 220     PacketHeader *hdr=  new PacketHeader();
 221     hdr->packetType = type;
 222     hdr->packetLength = len;
 223     hdr->srcNetworkID = 0;//networkid;
 224     hdr->version = 1;
 225     memcpy(((char*)totalBuffer) + sizeof(PacketHeader) , buf, len);
 226     int numbytes=0;
 227     if ((numbytes=os::send(clientfd, hdr, sizeof(PacketHeader), 0)) == -1) {
 228         printError(ErrOS, "Unable to send the packet\n");
 229         return ErrOS;
 230     }
 231 //    printf("Sent bytes %d\n", numbytes);
 232     if ((numbytes=os::send(clientfd, buf, len, 0)) == -1) {
 233         printError(ErrOS, "Unable to send the packet\n");
 234         return ErrOS;
 235     }
 236 //    printf("Sent bytes %d\n", numbytes);
 237     free(totalBuffer);
 238     return rv;
 239 }