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