src/network/SqlNetworkHandler.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 <SqlNetworkHandler.h>
  21 #include <AbsSqlConnection.h>
  22 #include <SqlConnection.h>
  23 #include <SqlOdbcConnection.h>
  24 #include <AbsSqlStatement.h>
  25 #include <SqlStatement.h>
  26 #include <SqlOdbcStatement.h>
  27 #include <SqlLogStatement.h>
  28 #include <Parser.h>
  29
  30 List SqlNetworkHandler::stmtList;
  31 AbsSqlConnection* SqlNetworkHandler::conn;
  32 SqlApiImplType SqlNetworkHandler::type;
  33 int SqlNetworkHandler::stmtID;
  34
  35 void *SqlNetworkHandler::process(PacketHeader &header, char *buffer)
  36 {
  37     DbRetVal rv = OK;
  38     char *ptr = NULL;
  39     ResponsePacket *rpkt = NULL;
  40     switch(header.packetType)
  41     {
  42       //  case NW_PKT_PREPARE:
  43             //return processPrepare(header, buffer);
  44          //   break;
  45         //case NW_PKT_COMMIT:
  46             //return processCommit(header, buffer);
  47           //  break;
  48         case SQL_NW_PKT_CONNECT:
  49             return processSqlConnect(header, buffer);
  50             break;
  51         case SQL_NW_PKT_PREPARE:
  52             return processSqlPrepare(header, buffer);
  53             break;
  54         case SQL_NW_PKT_EXECUTE:
  55             return processSqlExecute(header, buffer);
  56             break;
  57         case SQL_NW_PKT_FETCH:
  58             return processSqlFetch(header, buffer);
  59             break;
  60         case SQL_NW_PKT_COMMIT:
  61             return processSqlCommit(header, buffer);
  62             break;
  63         case SQL_NW_PKT_ROLLBACK:
  64             return processSqlRollback(header, buffer);
  65             break;
  66         case SQL_NW_PKT_DISCONNECT:
  67             conn->rollback();
  68             rv = conn->disconnect();
  69             rpkt = new ResponsePacket();
  70             ptr = (char *) &rpkt->retVal;
  71             *ptr = 1;
  72             strcpy(rpkt->errorString, "Success");
  73             return rpkt;
  74         case SQL_NW_PKT_FREE:
  75             return processSqlFree(header, buffer);
  76             break;
  77     }
  78 }
  79
  80 void * SqlNetworkHandler::processSqlConnect(PacketHeader &header, char *buffer)
  81 {
  82     ResponsePacket *rpkt = new ResponsePacket();
  83     printDebug(DM_Network, "Processing CONNECT");
  84     SqlPacketConnect *pkt = new SqlPacketConnect();
  85     pkt->setBuffer(buffer);
  86     pkt->setBufferSize(header.packetLength);
  87     pkt->unmarshall();
  88     char *ptr = (char *) &rpkt->retVal;
  89     DbRetVal rv=conn->connect(pkt->userName, pkt->passWord);
  90     if (rv != OK) {
  91         *ptr = 0;
  92         strcpy(rpkt->errorString, "User Authentication Failure");
  93         return rpkt;
  94         printf("connection failure\n");
  95     }
  96     if (rv == OK) {
  97         *ptr = 1;
  98         rv = conn->beginTrans();
  99         return rpkt;
 100     }
 101 }
 102
 103 void* SqlNetworkHandler::processSqlPrepare(PacketHeader &header, char *buffer)
 104 {
 105     ResponsePacket *rpkt = new ResponsePacket();
 106     rpkt->isSelect = false;
 107     char *retval = (char *) &rpkt->retVal;
 108     SqlPacketPrepare *pkt = new SqlPacketPrepare();
 109     pkt->setBuffer(buffer);
 110     pkt->setBufferSize(header.packetLength);
 111     pkt->unmarshall();
 112     printDebug(DM_Network, "PREPARE %s\n", pkt->stmtString);
 113     AbsSqlStatement *sqlstmt = SqlFactory::createStatement(type);
 114     sqlstmt->setConnection(conn);
 115     NetworkStmt *nwStmt = new NetworkStmt();
 116     nwStmt->stmtID = ++stmtID;
 117     printDebug(DM_Network, "Statement string %s\n", pkt->stmtString);
 118     nwStmt->stmt = sqlstmt;
 119     DbRetVal rv = sqlstmt->prepare(pkt->stmtString);
 120     if (rv != OK)
 121     {
 122         printError(ErrSysInit, "statement prepare failed\n");
 123         *retval = 0;
 124         strcpy(rpkt->errorString, "Error:Statement prepare failed");
 125         return rpkt;
 126     }
 127     int param = sqlstmt->noOfParamFields();
 128     int proj = sqlstmt->noOfProjFields();
 129     BindSqlField *bindField = NULL;
 130     BindSqlProjectField *projField = NULL;
 131     //populate paramList
 132     FieldInfo * fInfo = new FieldInfo();
 133     for (int i = 0; i < param; i++) {
 134         bindField = new BindSqlField();
 135         sqlstmt->getParamFldInfo(i + 1, fInfo);
 136         strcpy(bindField->fName, fInfo->fldName);
 137         bindField->type = fInfo->type;
 138         bindField->length = fInfo->length;
 139         bindField->offset = fInfo->offset;
 140         strcpy(bindField->defaultValueBuf, fInfo->defaultValueBuf);
 141         bindField->isNull = fInfo->isNull;
 142         bindField->isPrimary = fInfo->isPrimary;
 143         bindField->isDefault = fInfo->isDefault;
 144         bindField->isUnique = fInfo->isUnique;
 145         bindField->value = AllDataType::alloc(bindField->type, bindField->length);
 146         nwStmt->paramList.append(bindField);
 147     }
 148     delete fInfo;
 149     FieldInfo *fldInfo = new FieldInfo();
 150     for (int i = 0; i < proj; i++) {
 151         projField = new BindSqlProjectField();
 152         sqlstmt->getProjFldInfo(i + 1, fldInfo);
 153         strcpy(projField->fName, fInfo->fldName);
 154         projField->type = fldInfo->type;
 155         projField->length = fldInfo->length;
 156         projField->offset = fldInfo->offset;
 157         strcpy(projField->defaultValueBuf, fldInfo->defaultValueBuf);
 158         projField->isNull = fldInfo->isNull;
 159         projField->isPrimary = fldInfo->isPrimary;
 160         projField->isDefault = fldInfo->isDefault;
 161         projField->isUnique = fldInfo->isUnique;
 162         projField->value = AllDataType::alloc(projField->type, projField->length);
 163         nwStmt->projList.append(projField);
 164     }
 165     delete fldInfo;
 166     stmtList.append(nwStmt);
 167     *retval = 1;
 168     if(sqlstmt->isSelect()) rpkt->isSelect = true;
 169     if (param) *(retval+2) = 1;
 170     if (proj) *(retval+3) = 1;
 171     rpkt->stmtID = nwStmt->stmtID;
 172     strcpy(rpkt->errorString, "Success");
 173     return rpkt;
 174 }
 175
 176 void * SqlNetworkHandler::processSqlExecute(PacketHeader &header, char *buffer)
 177 {
 178     ResponsePacket *rpkt = new ResponsePacket();
 179     char *retval = (char *) &rpkt->retVal;
 180     SqlPacketExecute *pkt = new SqlPacketExecute();
 181     pkt->setBuffer(buffer);
 182     pkt->setBufferSize(header.packetLength);
 183     pkt->setStatementList(stmtList);
 184     pkt->unmarshall();
 185     printDebug(DM_Network, "PREPARE %d\n", pkt->stmtID);
 186     rpkt->stmtID = pkt->stmtID;
 187     ListIterator stmtIter = stmtList.getIterator();
 188     NetworkStmt *stmt;
 189     while (stmtIter.hasElement())
 190     {
 191        stmt = (NetworkStmt*) stmtIter.nextElement();
 192        //TODO::Also check the srcNetworkID
 193        if (stmt->stmtID == pkt->stmtID ) break;
 194     }
 195     AbsSqlStatement *sqlstmt = stmt->stmt;
 196     int rows = 0;
 197     for (int i=0; i < pkt->noParams; i++) {
 198         BindSqlField *bindField = (BindSqlField *) stmt->paramList.get(i+1);
 199         setParamValues(sqlstmt, i+1, bindField->type, bindField->length, (char *)bindField->value);
 200     }
 201     //SqlStatement *st = (SqlStatement *)sqlstmt;
 202     if(sqlstmt->isSelect()) {
 203         int noProj = stmt->projList.size();
 204         for (int i=0; i < noProj; i++) {
 205             BindSqlProjectField *prjFld = (BindSqlProjectField *) stmt->projList.get(i+1);
 206             sqlstmt->bindField(i+1, prjFld->value);
 207         }
 208     }
 209     DbRetVal rv = sqlstmt->execute(rows);
 210     if (rv != OK) {
 211         *retval = 0;
 212         strcpy(rpkt->errorString, "Execute failed");
 213         return rpkt;
 214     }
 215     *retval = 1;
 216     rpkt->rows = rows;
 217     strcpy(rpkt->errorString, "Success");
 218     return rpkt;
 219 }
 220
 221 void * SqlNetworkHandler::processSqlFetch(PacketHeader &header, char *buffer)
 222 {
 223     ResponsePacket *rpkt = new ResponsePacket();
 224     char *retval = (char *) &rpkt->retVal;
 225     SqlPacketFetch *pkt = new SqlPacketFetch();
 226     pkt->setBuffer(buffer);
 227     pkt->unmarshall();
 228     rpkt->stmtID = pkt->stmtID;
 229     ListIterator stmtIter = stmtList.getIterator();
 230     NetworkStmt *stmt;
 231     while (stmtIter.hasElement())
 232     {
 233        stmt = (NetworkStmt*) stmtIter.nextElement();
 234        //TODO::Also check teh srcNetworkID
 235        if (stmt->stmtID == pkt->stmtID ) break;
 236     }
 237     AbsSqlStatement *sqlstmt = stmt->stmt;
 238     void *data=NULL;
 239     DbRetVal rv = OK;
 240     if ((data = sqlstmt->fetch(rv)) != NULL && rv == OK) {
 241         *retval = 1;
 242         strcpy(rpkt->errorString, "Success");
 243         return rpkt;
 244     }
 245     if (data == NULL && rv == OK) {
 246         sqlstmt->close();
 247         *retval = 1;
 248         *(retval + 1) = 1;
 249         strcpy(rpkt->errorString, "Success fetch completed");
 250         return rpkt;
 251     }
 252     else {
 253         *retval = 0;
 254         strcpy(rpkt->errorString, "fetch completed");
 255         return rpkt;
 256     }
 257 }
 258
 259 void * SqlNetworkHandler::processSqlFree(PacketHeader &header, char *buffer)
 260 {
 261     ResponsePacket *rpkt = new ResponsePacket();
 262     char *retval = (char *) &rpkt->retVal;
 263     SqlPacketFree *pkt = new SqlPacketFree();
 264     pkt->setBuffer(buffer);
 265     pkt->unmarshall();
 266     rpkt->stmtID = pkt->stmtID;
 267     ListIterator stmtIter = stmtList.getIterator();
 268     NetworkStmt *stmt;
 269     while (stmtIter.hasElement())
 270     {
 271        stmt = (NetworkStmt*) stmtIter.nextElement();
 272        //TODO::Also check teh srcNetworkID
 273        if (stmt->stmtID == pkt->stmtID ) break;
 274     }
 275     AbsSqlStatement *sqlstmt = stmt->stmt;
 276     sqlstmt->free();
 277     ListIterator itprm = stmt->paramList.getIterator();
 278     BindSqlField *fld = NULL;
 279     while((fld = (BindSqlField *) itprm.nextElement()) != NULL) delete fld;
 280     stmt->paramList.reset();
 281     ListIterator itprj = stmt->projList.getIterator();
 282     BindSqlProjectField *pfld = NULL;
 283     while((pfld = (BindSqlProjectField *) itprj.nextElement()) != NULL) delete pfld;
 284     stmt->projList.reset();
 285     delete stmt->stmt;
 286     stmtList.remove(stmt);
 287     delete stmt;
 288     *retval = 1;
 289     strcpy(rpkt->errorString, "Success");
 290     return rpkt;
 291 }
 292
 293
 294 void * SqlNetworkHandler::processSqlCommit(PacketHeader &header, char *buffer)
 295 {
 296     ResponsePacket *rpkt = new ResponsePacket();
 297     char *retval = (char *) &rpkt->retVal;
 298     DbRetVal rv = conn->commit();
 299     if (rv != OK) {
 300         *retval = 0;
 301         strcpy(rpkt->errorString, "Commit failure\n");
 302         return rpkt;
 303     }
 304     rv = conn->beginTrans();
 305     *retval = 1;
 306     strcpy(rpkt->errorString, "Success");
 307     return rpkt;
 308 }
 309
 310 void *SqlNetworkHandler::processSqlRollback(PacketHeader &header, char *buffer)
 311 {
 312     ResponsePacket *rpkt = new ResponsePacket();
 313     char *retval = (char *) &rpkt->retVal;
 314
 315     DbRetVal rv = conn->rollback();
 316     if (rv != OK) {
 317         *retval = 0;
 318         strcpy(rpkt->errorString, "Rollback failure\n");
 319         return rpkt;
 320     }
 321     rv = conn->beginTrans();
 322     *retval = 1;
 323     strcpy(rpkt->errorString, "Success");
 324     return rpkt;
 325 }
 326 /*
 327 void *SqlNetworkHandler::processCommit(PacketHeader &header, char *buffer)
 328 {
 329     printDebug(DM_Network, "Processing COMMIT");
 330     PacketCommit *pkt = new PacketCommit();
 331     pkt->setBuffer(buffer);
 332     pkt->setBufferSize(header.packetLength);
 333     pkt->unmarshall();
 334     List pktList;
 335     pkt->getExecPacketList(stmtList, pktList);
 336     DbRetVal rv = applyExecPackets(stmtList, pktList);
 337     int response = 1;
 338     if (rv != OK)
 339     {
 340          printError(ErrSysFatal, "Unable to apply the exec packets\n");
 341          response =0;
 342     }
 343     return response;
 344
 345 }
 346 void * SqlNetworkHandler::processFree(PacketHeader &header, char *buffer)
 347 {
 348     PacketFree *pkt = new PacketFree();
 349     pkt->setBuffer(buffer);
 350     pkt->setBufferSize(header.packetLength);
 351     pkt->unmarshall();
 352     //printf("FREE %d \n", pkt->stmtID);
 353     int response =1;
 354     //This wont work for two statement executed in same transaction using same SqlStatement object using free.
 355     //so do not delete now and put a flag 'readyfordelete' in NetworkStmt object and delete it during execute
 356 //
 357     ListIterator iter = stmtList.getIterator();
 358     NetworkStmt *stmt, *removeStmt = NULL;
 359     while (iter.hasElement())
 360     {
 361        stmt = (NetworkStmt*)iter.nextElement();
 362        if (stmt->srcNetworkID == header.srcNetworkID
 363                 && stmt->stmtID == pkt->stmtID)
 364        {
 365                   removeStmt = stmt;
 366                   break;
 367        }
 368     }
 369     if (removeStmt) stmtList.remove(removeStmt);
 370     else printf("Statement id %d not found in list \n",  pkt->stmtID);
 371
 372     return response;
 373 }
 374 void * SqlNetworkHandler::processPrepare(PacketHeader &header, char *buffer)
 375 {
 376     PacketPrepare *pkt = new PacketPrepare();
 377     pkt->setBuffer(buffer);
 378     pkt->setBufferSize(header.packetLength);
 379     pkt->unmarshall();
 380     printDebug(DM_Network, "PREPARE %d %s\n", pkt->stmtID, pkt->stmtString);
 381     //for (int i =0 ; i < pkt->noParams; i++)
 382         //printf("PREPARE type %d length %d \n", pkt->type[i], pkt->length[i]);
 383     int response =1;
 384     //TODO::add it to the SqlStatement list
 385     AbsSqlStatement *sqlstmt = SqlFactory::createStatement(type);
 386     sqlstmt->setConnection(conn);
 387     NetworkStmt *nwStmt = new NetworkStmt();
 388     printDebug(DM_Network, "Statement string %s\n", pkt->stmtString);
 389     nwStmt->srcNetworkID = header.srcNetworkID;
 390     nwStmt->stmtID =  pkt->stmtID;
 391     nwStmt->stmt = sqlstmt;
 392     DbRetVal rv = sqlstmt->prepare(pkt->stmtString);
 393     if (rv != OK)
 394     {
 395         printError(ErrSysInit, "statement prepare failed\n");
 396         response = 0;
 397         return response;
 398     }
 399     BindSqlField *bindField = NULL;
 400     //populate paramList
 401     for (int i = 0; i < pkt->noParams; i++)
 402     {
 403              bindField = new BindSqlField();
 404              bindField->type = (DataType) pkt->type[i];
 405              bindField->length = pkt->length[i];
 406              bindField->value = AllDataType::alloc(bindField->type,
 407                                                    bindField->length);
 408              nwStmt->paramList.append(bindField);
 409     }
 410     stmtList.append(nwStmt);
 411     return response;
 412
 413 }
 414
 415 DbRetVal SqlNetworkHandler::applyExecPackets(List sList, List pList)
 416 {
 417     ListIterator stmtIter = sList.getIterator();
 418     NetworkStmt *nwstmt;
 419     DbRetVal rv = conn->beginTrans();
 420     if (rv != OK) return rv;
 421     ListIterator pktIter = pList.getIterator();
 422     PacketExecute *pkt;
 423     int i = 0;
 424     BindSqlField *bindField;
 425     while (pktIter.hasElement())
 426     {
 427         pkt = (PacketExecute*) pktIter.nextElement();
 428         stmtIter.reset();
 429         bool found = false;
 430         while (stmtIter.hasElement())
 431         {
 432             nwstmt = (NetworkStmt*) stmtIter.nextElement();
 433             if (nwstmt->stmtID == pkt->stmtID) {found = true ; break;}
 434         }
 435         if (!found) {
 436            printf("stmt not found in list. Negleting unreplicated table...\n");
 437            continue;
 438         }
 439         ListIterator paramIter = nwstmt->paramList.getIterator();
 440         i = 0;
 441         while (paramIter.hasElement()) {
 442             bindField = (BindSqlField*) paramIter.nextElement();
 443             setParamValues(nwstmt->stmt, i+1,  bindField->type, bindField->length, pkt->paramValues[i]);
 444             i++;
 445         }
 446         int rows= 0;
 447         DbRetVal rv = nwstmt->stmt->execute(rows);
 448         if (rv != OK )
 449         {
 450             printError(ErrSysFatal, "sql execute failed with rv %d\n", rv);
 451             //TODO::log all things like SQL statements to a file
 452             SqlNetworkHandler::conn->rollback();
 453             printError(ErrPeerExecFailed, "Transaction Rolledback\n");
 454             return ErrPeerExecFailed;
 455         }
 456     }
 457     SqlNetworkHandler::conn->commit();
 458     return OK;
 459 }
 460 */
 461 void SqlNetworkHandler::setParamValues(AbsSqlStatement *stmt, int parampos, DataType type,
 462                     int length, char *value)
 463 {
 464     switch(type)
 465     {
 466         case typeInt:
 467             stmt->setIntParam(parampos, *(int*)value);
 468             break;
 469         case typeLong:
 470             stmt->setLongParam(parampos, *(long*)value);
 471             break;
 472         case typeLongLong:
 473             stmt->setLongLongParam(parampos, *(long long*)value);
 474             break;
 475         case typeShort:
 476             stmt->setShortParam(parampos, *(short*)value);
 477             break;
 478         case typeByteInt:
 479             stmt->setByteIntParam(parampos, *(char*)value);
 480             break;
 481         case typeDouble:
 482             stmt->setDoubleParam(parampos, *(double*)value);
 483             break;
 484         case typeFloat:
 485             stmt->setFloatParam(parampos, *(float*)value);
 486             break;
 487         case typeDate:
 488             stmt->setDateParam(parampos, *(Date*)value);
 489             break;
 490         case typeTime:
 491             stmt->setTimeParam(parampos, *(Time*)value);
 492             break;
 493         case typeTimeStamp:
 494             stmt->setTimeStampParam(parampos, *(TimeStamp*)value);
 495             break;
 496         case typeString:
 497             {
 498                 char *d =(char*)value;
 499                 d[length-1] = '\0';
 500                 stmt->setStringParam(parampos, (char*)value);
 501                 break;
 502             }
 503         case typeBinary:
 504             stmt->setBinaryParam(parampos, (char *) value);
 505             break;
 506     }
 507     return;
 508 }
 509