src/tools/csqlcacheserver.cxx

   1 /***************************************************************************
   2  *   Copyright (C) 2007 by www.databasecache.com                           *
   3  *   Contact: praba_tuty@databasecache.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   ***************************************************************************/
  16 #include <AbsSqlConnection.h>
  17 #include <AbsSqlStatement.h>
  18 #include <SqlOdbcStatement.h>
  19 #include <SqlFactory.h>
  20 #include <CSql.h>
  21
  22 int insert(Table *table, int pkid);
  23 int remove(Table *table, int pkid);
  24 int getRecordsFromTargetDb(int mode);
  25
  26 int srvStop =0;
  27 static void sigTermHandler(int sig)
  28 {
  29     printf("Received signal %d\nStopping the server\n", sig);
  30     srvStop = 1;
  31 }
  32
  33 void printUsage()
  34 {
  35    printf("Usage: csqlcacheserver \n");
  36    printf("Description: Start the csql caching server.\n");
  37    return;
  38 }
  39 AbsSqlConnection *targetconn;
  40 Connection conn;
  41 int main(int argc, char **argv)
  42 {
  43     int c = 0, opt = 0;
  44     while ((c = getopt(argc, argv, "?")) != EOF)
  45     {
  46         switch (c)
  47         {
  48             case '?' : { opt = 10; break; } //print help
  49             default: opt=10;
  50
  51         }
  52     }//while options
  53
  54     if (opt == 10) {
  55         printUsage();
  56         return 0;
  57     }
  58
  59     os::signal(SIGINT, sigTermHandler);
  60     os::signal(SIGTERM, sigTermHandler);
  61
  62     DbRetVal rv = conn.open("root", "manager");
  63     if (rv != OK) return 1;
  64     targetconn = SqlFactory::createConnection(CSqlAdapter);
  65     rv = targetconn->connect("root", "manager");
  66     if (rv != OK) return 1;
  67     if (!Conf::config.useCache())
  68     {
  69         printf("Cache is set to OFF in csql.conf file\n");
  70         return 1;
  71     }
  72     AbsSqlStatement *stmt = SqlFactory::createStatement(CSqlAdapter);
  73     stmt->setConnection(targetconn);
  74     /*rv = stmt->prepare("create table csql_log_int(tablename char(64), pkid int, operation int, id int not null unique auto_increment)engine='innodb';");
  75     targetconn->beginTrans();
  76     int rows=0;
  77     stmt->execute(rows);
  78     targetconn->commit();
  79     stmt->free();
  80     delete stmt;*/
  81
  82     printf("Cache server started\n");
  83     int ret = 0;
  84     struct timeval timeout, tval;
  85     timeout.tv_sec = Conf::config.getCacheWaitSecs();
  86     timeout.tv_usec = 0;
  87
  88     while(!srvStop)
  89     {
  90         tval.tv_sec = timeout.tv_sec;
  91         tval.tv_usec = timeout.tv_usec;
  92         ret = os::select(0, NULL, 0, 0, &tval);
  93         printf("Checking for cache updates\n");
  94         ret = getRecordsFromTargetDb(1);
  95         if (ret !=0) srvStop = 1;
  96         //ret = getRecordsFromTargetDb(2);
  97         if (ret !=0) srvStop = 1;
  98     }
  99     printf("Cache Server Exiting\n");
 100     conn.close();
 101     targetconn->disconnect();
 102     return 0;
 103 }
 104 int getRecordsFromTargetDb(int mode)
 105 {
 106     int pkid;
 107     char tablename[64];
 108     int op, id,caId;
 109     int rows =0;
 110     DbRetVal rv = OK;
 111     char StmtStr[1024];
 112     caId =Conf::config.getCacheID();
 113     AbsSqlStatement *stmt = SqlFactory::createStatement(CSqlAdapter);
 114     stmt->setConnection(targetconn);
 115     AbsSqlStatement *delstmt = SqlFactory::createStatement(CSqlAdapter);
 116     delstmt->setConnection(targetconn);
 117     if (mode == 1 ) {
 118         //rv = delstmt->prepare("DELETE from csql_log_int where id=?;");
 119         sprintf(StmtStr, "SELECT * FROM csql_log_int where cacheid = %d;", caId);
 120         rv = stmt->prepare(StmtStr);
 121         if (rv != OK) {printf("Stmt prepare failed\n"); return 1; }
 122     }
 123     else {
 124         rv = stmt->prepare("SELECT * FROM csql_log_char;");
 125         if (rv != OK) {printf("Stmt prepare failed\n"); return 1; }
 126         //rv = delstmt->prepare("DELETE from csql_log_char where id=?;");
 127     }
 128     if (rv != OK) {printf("Stmt prepare failed\n"); return 1; }
 129     stmt->bindField(1, tablename);
 130     stmt->bindField(2, &pkid);
 131     stmt->bindField(3, &op);
 132     stmt->bindField(4, &caId);
 133     stmt->bindField(5, &id);
 134
 135     DatabaseManager *dbMgr = conn.getDatabaseManager();
 136     while(true) {
 137       rv = targetconn->beginTrans();
 138       rv = stmt->execute(rows);
 139       if (rv != OK)
 140       {
 141           printError(ErrSysInit, "Unable to execute stmt in target db");
 142           targetconn->rollback();
 143           stmt->free();
 144           delstmt->free();
 145           delete stmt;
 146           delete delstmt;
 147           return 1;
 148       }
 149       if (stmt->fetch() != NULL) {
 150           printf("Row value is %s %d %d %d\n", tablename, pkid, op,caId);
 151
 152                 Table *table = dbMgr->openTable(tablename);
 153                 int ret = 0;
 154                 if (table == NULL)
 155                 {
 156                         printError(ErrSysInit, "Table %s not exist in csql", tablename);
 157                         targetconn->rollback();
 158                         stmt->free();
 159                         delstmt->free();
 160                         delete stmt;
 161                         delete delstmt;
 162                         break;
 163                 }
 164                 if (op == 2)//DELETE
 165                 {
 166                         ret = remove(table,pkid);
 167                 }
 168                 else //INSERT
 169                 {
 170                         ret = insert(table, pkid);
 171                 }
 172                 dbMgr->closeTable(table);
 173                 rv = targetconn->commit();
 174                 rv = targetconn->beginTrans();
 175           //Remove record from csql_log_XXX table
 176                 sprintf(StmtStr, "DELETE from csql_log_int where id=%d ;", id);
 177                 rv = delstmt->prepare(StmtStr);
 178                 if (rv != OK) {printf("FAILED\n"); return 1; }
 179                 // delstmt->setIntParam(1, id);
 180                 rv = delstmt->execute(rows);
 181                 if (rv != OK)
 182                 {
 183                         printf("log record not deleted from the target db %d\n", rv);
 184                         targetconn->rollback();
 185                         stmt->free();
 186                         delstmt->free();
 187                         delete stmt;
 188                         delete delstmt;
 189                 }
 190                 delstmt->free();
 191
 192                 rv = targetconn->commit();
 193
 194      }
 195      else {
 196          stmt->close();
 197          break;
 198      }
 199      stmt->close();
 200    }
 201    stmt->free();
 202    delstmt->free();
 203    delete stmt;
 204    delete delstmt;
 205    return 0;
 206 }
 207 int insert(Table *table, int pkid)
 208 {
 209     AbsSqlStatement *stmt = SqlFactory::createStatement(CSqlAdapter);
 210     stmt->setConnection(targetconn);
 211     SqlOdbcStatement *ostmt = (SqlOdbcStatement*) stmt;
 212     char pkfieldname[128];
 213     ostmt->getPrimaryKeyFieldName(table->getName(), pkfieldname);
 214     char sbuf[1024];
 215     sprintf(sbuf, "SELECT * FROM %s where %s = %d;", table->getName(), pkfieldname, pkid);
 216     //TODO::get the primary key field name from the table interface. need to implement it
 217     DbRetVal rv = stmt->prepare(sbuf);
 218     if (rv != OK) return 1;
 219
 220     List fNameList = table->getFieldNameList();
 221     ListIterator fNameIter = fNameList.getIterator();
 222     FieldInfo *info = new FieldInfo();
 223     int fcount =1; void *valBuf; int fieldsize=0;
 224     void *buf[128];//TODO:resticts to support only 128 fields in table
 225     Identifier *elem = NULL;
 226     while (fNameIter.hasElement()) {
 227         elem = (Identifier*) fNameIter.nextElement();
 228         table->getFieldInfo((const char*)elem->name, info);
 229         valBuf = AllDataType::alloc(info->type, info->length);
 230         buf[fcount] = valBuf;
 231         table->bindFld(elem->name, valBuf);
 232         stmt->bindField(fcount++, valBuf);
 233
 234     }
 235     delete info;
 236     int rows=0;
 237     int retValue = stmt->execute(rows);
 238     if (retValue && rows != 1) {printError(ErrSysInit, "Unable to execute statement at target db\n"); return ErrSysInit; }
 239     conn.startTransaction();
 240     if (stmt->fetch() != NULL) {
 241         table->insertTuple();
 242         //Note:insert may fail if the record is inserted from this cache
 243     }
 244     for (int i=1; i < fcount; i++) {
 245         free(buf[i]);
 246     }
 247     stmt->free();
 248     delete stmt;
 249     conn.commit();
 250     return 0;
 251 }
 252 int remove(Table *table, int pkid)
 253 {
 254     DbRetVal rv = OK;
 255     AbsSqlStatement *stmt = SqlFactory::createStatement(CSqlAdapter);
 256     stmt->setConnection(targetconn);
 257     SqlOdbcStatement *ostmt = (SqlOdbcStatement*) stmt;
 258     char pkfieldname[128];
 259     ostmt->getPrimaryKeyFieldName(table->getName(), pkfieldname);
 260     delete stmt;
 261     Condition p1;
 262     p1.setTerm(pkfieldname, OpEquals, &pkid);
 263     table->setCondition(&p1);
 264     rv = conn.startTransaction();
 265     if (rv != OK) return 1;
 266     rv = table->execute();
 267     if (rv != OK)
 268     {
 269         table->setCondition(NULL);
 270         conn.rollback();
 271         return 1;
 272     }
 273     if (table->fetch() != NULL)
 274         rv = table->deleteTuple();
 275         //Note:Delete may fail if the record is deleted from this cache
 276     table->setCondition(NULL);
 277     rv = conn.commit();
 278     if (rv != OK) return 1;
 279     return 0;
 280 }
 281