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
  28 #include <SqlLogStatement.h>
  29
  30 List SqlNetworkHandler::stmtList;
  31 AbsSqlConnection* SqlNetworkHandler::conn;
  32 SqlApiImplType SqlNetworkHandler::type;
  33
  34 int SqlNetworkHandler::process(PacketHeader &header, char *buffer)
  35 {
  36     switch(header.packetType)
  37     {
  38         case NW_PKT_PREPARE:
  39             return processPrepare(header, buffer);
  40             break;
  41         case NW_PKT_COMMIT:
  42             return processCommit(header, buffer);
  43             break;
  44     }
  45 }
  46
  47 int SqlNetworkHandler::processCommit(PacketHeader &header, char *buffer)
  48 {
  49     printDebug(DM_Network, "Processing COMMIT");
  50     PacketCommit *pkt = new PacketCommit();
  51     pkt->setBuffer(buffer);
  52     pkt->setBufferSize(header.packetLength);
  53     pkt->unmarshall();
  54     List pktList;
  55     pkt->getExecPacketList(stmtList, pktList);
  56     DbRetVal rv = applyExecPackets(stmtList, pktList);
  57     int response = 1;
  58     if (rv != OK)
  59     {
  60          printError(ErrSysFatal, "Unable to apply the exec packets\n");
  61          response =0;
  62     }
  63     return response;
  64
  65 }
  66 int SqlNetworkHandler::processFree(PacketHeader &header, char *buffer)
  67 {
  68     PacketFree *pkt = new PacketFree();
  69     pkt->setBuffer(buffer);
  70     pkt->setBufferSize(header.packetLength);
  71     pkt->unmarshall();
  72     //printf("FREE %d \n", pkt->stmtID);
  73     int response =1;
  74     //This wont work for two statement executed in same transaction using same SqlStatement object using free.
  75     //so do not delete now and put a flag 'readyfordelete' in NetworkStmt object and delete it during execute
  76     /*
  77     ListIterator iter = stmtList.getIterator();
  78     NetworkStmt *stmt, *removeStmt = NULL;
  79     while (iter.hasElement())
  80     {
  81        stmt = (NetworkStmt*)iter.nextElement();
  82        if (stmt->srcNetworkID == header.srcNetworkID
  83                 && stmt->stmtID == pkt->stmtID)
  84        {
  85                   removeStmt = stmt;
  86                   break;
  87        }
  88     }
  89     if (removeStmt) stmtList.remove(removeStmt);
  90     else printf("Statement id %d not found in list \n",  pkt->stmtID);
  91     */
  92     return response;
  93 }
  94 int SqlNetworkHandler::processPrepare(PacketHeader &header, char *buffer)
  95 {
  96     PacketPrepare *pkt = new PacketPrepare();
  97     pkt->setBuffer(buffer);
  98     pkt->setBufferSize(header.packetLength);
  99     pkt->unmarshall();
 100     printDebug(DM_Network, "PREPARE %d %s\n", pkt->stmtID, pkt->stmtString);
 101     //for (int i =0 ; i < pkt->noParams; i++)
 102         //printf("PREPARE type %d length %d \n", pkt->type[i], pkt->length[i]);
 103     int response =1;
 104     //TODO::add it to the SqlStatement list
 105     AbsSqlStatement *sqlstmt = SqlFactory::createStatement(type);
 106     sqlstmt->setConnection(conn);
 107     NetworkStmt *nwStmt = new NetworkStmt();
 108     printDebug(DM_Network, "Statement string %s\n", pkt->stmtString);
 109     nwStmt->srcNetworkID = header.srcNetworkID;
 110     nwStmt->stmtID =  pkt->stmtID;
 111     nwStmt->stmt = sqlstmt;
 112     DbRetVal rv = sqlstmt->prepare(pkt->stmtString);
 113     if (rv != OK)
 114     {
 115         printError(ErrSysInit, "statement prepare failed\n");
 116         response = 0;
 117         return response;
 118     }
 119     BindSqlField *bindField = NULL;
 120     //populate paramList
 121     for (int i = 0; i < pkt->noParams; i++)
 122     {
 123              bindField = new BindSqlField();
 124              bindField->type = (DataType) pkt->type[i];
 125              bindField->length = pkt->length[i];
 126              bindField->value = AllDataType::alloc(bindField->type,
 127                                                    bindField->length);
 128              nwStmt->paramList.append(bindField);
 129     }
 130     stmtList.append(nwStmt);
 131     return response;
 132
 133 }
 134
 135 DbRetVal SqlNetworkHandler::applyExecPackets(List sList, List pList)
 136 {
 137     ListIterator stmtIter = sList.getIterator();
 138     NetworkStmt *nwstmt;
 139     DbRetVal rv = conn->beginTrans();
 140     if (rv != OK) return rv;
 141     ListIterator pktIter = pList.getIterator();
 142     PacketExecute *pkt;
 143     int i = 0;
 144     BindSqlField *bindField;
 145     while (pktIter.hasElement())
 146     {
 147         pkt = (PacketExecute*) pktIter.nextElement();
 148         stmtIter.reset();
 149         bool found = false;
 150         while (stmtIter.hasElement())
 151         {
 152             nwstmt = (NetworkStmt*) stmtIter.nextElement();
 153             if (nwstmt->stmtID == pkt->stmtID) {found = true ; break;}
 154         }
 155         if (!found) {
 156            printf("stmt not found in list. Negleting unreplicated table...\n");
 157            continue;
 158         }
 159         ListIterator paramIter = nwstmt->paramList.getIterator();
 160         i = 0;
 161         while (paramIter.hasElement()) {
 162             bindField = (BindSqlField*) paramIter.nextElement();
 163             setParamValues(nwstmt->stmt, i+1,  bindField->type, bindField->length, pkt->paramValues[i]);
 164             i++;
 165         }
 166         int rows= 0;
 167         DbRetVal rv = nwstmt->stmt->execute(rows);
 168         if (rv != OK )
 169         {
 170             printError(ErrSysFatal, "sql execute failed with rv %d\n", rv);
 171             //TODO::log all things like SQL statements to a file
 172             SqlNetworkHandler::conn->rollback();
 173             printError(ErrPeerExecFailed, "Transaction Rolledback\n");
 174             return ErrPeerExecFailed;
 175         }
 176     }
 177     SqlNetworkHandler::conn->commit();
 178     return OK;
 179 }
 180
 181 void SqlNetworkHandler::setParamValues(AbsSqlStatement *stmt, int parampos, DataType type,
 182                     int length, char *value)
 183 {
 184     switch(type)
 185     {
 186         case typeInt:
 187             stmt->setIntParam(parampos, *(int*)value);
 188             break;
 189         case typeLong:
 190             stmt->setLongParam(parampos, *(long*)value);
 191             break;
 192         case typeLongLong:
 193             stmt->setLongLongParam(parampos, *(long long*)value);
 194             break;
 195         case typeShort:
 196             stmt->setShortParam(parampos, *(short*)value);
 197             break;
 198         case typeByteInt:
 199             stmt->setByteIntParam(parampos, *(char*)value);
 200             break;
 201         case typeDouble:
 202             stmt->setDoubleParam(parampos, *(double*)value);
 203             break;
 204         case typeFloat:
 205             stmt->setFloatParam(parampos, *(float*)value);
 206             break;
 207         case typeDate:
 208             stmt->setDateParam(parampos, *(Date*)value);
 209             break;
 210         case typeTime:
 211             stmt->setTimeParam(parampos, *(Time*)value);
 212             break;
 213         case typeTimeStamp:
 214             stmt->setTimeStampParam(parampos, *(TimeStamp*)value);
 215             break;
 216         case typeString:
 217             {
 218                 char *d =(char*)value;
 219                 d[length-1] = '\0';
 220                 stmt->setStringParam(parampos, (char*)value);
 221                 break;
 222             }
 223     }
 224     return;
 225 }
 226