projects/stargazer/plugins/store/postgresql/postgresql_store_messages.cpp

   1 /*
   2  *    This program is free software; you can redistribute it and/or modify
   3  *    it under the terms of the GNU General Public License as published by
   4  *    the Free Software Foundation; either version 2 of the License, or
   5  *    (at your option) any later version.
   6  *
   7  *    This program is distributed in the hope that it will be useful,
   8  *    but WITHOUT ANY WARRANTY; without even the implied warranty of
   9  *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  10  *    GNU General Public License for more details.
  11  *
  12  *    You should have received a copy of the GNU General Public License
  13  *    along with this program; if not, write to the Free Software
  14  *    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  15  */
  16
  17 /*
  18  *    Author : Maxim Mamontov <faust@stargazer.dp.ua>
  19  */
  20
  21
  22 /*
  23  *  Messages manipualtion methods
  24  *
  25  *  $Revision: 1.6 $
  26  *  $Date: 2009/07/15 11:19:42 $
  27  *
  28  */
  29
  30 #include <string>
  31 #include <vector>
  32 #include <sstream>
  33
  34 #include <libpq-fe.h>
  35
  36 #include "stg/common.h"
  37 #include "postgresql_store.h"
  38 #include "stg/locker.h"
  39 #include "stg/message.h"
  40
  41 //-----------------------------------------------------------------------------
  42 int POSTGRESQL_STORE::AddMessage(STG_MSG * msg, const std::string & login) const
  43 {
  44 STG_LOCKER lock(&mutex);
  45
  46 if (PQstatus(connection) != CONNECTION_OK)
  47     {
  48     printfd(__FILE__, "POSTGRESQL_STORE::AddMessage(): 'Connection lost. Trying to reconnect...'\n", strError.c_str());
  49     if (Reset())
  50         {
  51         strError = "Connection lost";
  52         printfd(__FILE__, "POSTGRESQL_STORE::AddMessage(): '%s'\n", strError.c_str());
  53         return -1;
  54         }
  55     }
  56
  57 PGresult * result;
  58
  59 if (StartTransaction())
  60     {
  61     printfd(__FILE__, "POSTGRESQL_STORE::AddMessage(): 'Failed to start transaction'\n");
  62     return -1;
  63     }
  64
  65 std::string elogin = login;
  66 std::string etext = msg->text;
  67
  68 if (EscapeString(elogin))
  69     {
  70     printfd(__FILE__, "POSTGRESQL_STORE::AddMessage(): 'Failed to escape login'\n");
  71     if (RollbackTransaction())
  72         {
  73         printfd(__FILE__, "POSTGRESQL_STORE::AddMessage(): 'Failed to rollback transaction'\n");
  74         }
  75     return -1;
  76     }
  77
  78 if (EscapeString(etext))
  79     {
  80     printfd(__FILE__, "POSTGRESQL_STORE::AddMessage(): 'Failed to escape message text'\n");
  81     if (RollbackTransaction())
  82         {
  83         printfd(__FILE__, "POSTGRESQL_STORE::AddMessage(): 'Failed to rollback transaction'\n");
  84         }
  85     return -1;
  86     }
  87
  88 std::ostringstream query;
  89 query << "SELECT sp_add_message("
  90       << "'" << elogin << "', "
  91       << "CAST(1 AS SMALLINT), " // Here need to be a version, but, it's uninitiated actually
  92       << "CAST(" << msg->header.type << " AS SMALLINT), "
  93       << "CAST('" << formatTime(msg->header.lastSendTime) << "' AS TIMESTAMP), "
  94       << "CAST('" << formatTime(msg->header.creationTime) << "' AS TIMESTAMP), "
  95       << msg->header.showTime << ", "
  96       << "CAST(" << msg->header.repeat << " AS SMALLINT), "
  97       << msg->header.repeatPeriod << ", "
  98       << "'" << etext << "')";
  99
 100 result = PQexec(connection, query.str().c_str());
 101
 102 if (PQresultStatus(result) != PGRES_TUPLES_OK)
 103     {
 104     strError = PQresultErrorMessage(result);
 105     PQclear(result);
 106     printfd(__FILE__, "POSTGRESQL_STORE::AddMessage(): '%s'\n", strError.c_str());
 107     if (RollbackTransaction())
 108         {
 109         printfd(__FILE__, "POSTGRESQL_STORE::AddMessage(): 'Failed to rollback transaction'\n");
 110         }
 111     return -1;
 112     }
 113
 114 int tuples = PQntuples(result);
 115
 116 if (tuples != 1)
 117     {
 118     strError = "Failed to fetch newlly added message ID";
 119     printfd(__FILE__, "POSTGRESQL_STORE::AddMessage(): 'Invalid number of tuples. Wanted 1, actulally %d'\n", tuples);
 120     PQclear(result);
 121     if (RollbackTransaction())
 122         {
 123         printfd(__FILE__, "POSTGRESQL_STORE::AddMessage(): 'Failed to rollback transaction'\n");
 124         }
 125     return -1;
 126     }
 127
 128 std::stringstream tuple;
 129 tuple << PQgetvalue(result, 0, 0);
 130
 131 PQclear(result);
 132
 133 tuple >> msg->header.id;
 134
 135 if (CommitTransaction())
 136     {
 137     printfd(__FILE__, "POSTGRESQL_STORE::AddMessage(): 'Failed to commit transaction'\n");
 138     return -1;
 139     }
 140
 141 return 0;
 142 }
 143 //-----------------------------------------------------------------------------
 144 int POSTGRESQL_STORE::EditMessage(const STG_MSG & msg,
 145                                   const std::string & login) const
 146 {
 147 STG_LOCKER lock(&mutex);
 148
 149 if (PQstatus(connection) != CONNECTION_OK)
 150     {
 151     printfd(__FILE__, "POSTGRESQL_STORE::EditMessage(): 'Connection lost. Trying to reconnect...'\n", strError.c_str());
 152     if (Reset())
 153         {
 154         strError = "Connection lost";
 155         printfd(__FILE__, "POSTGRESQL_STORE::EditMessage(): '%s'\n", strError.c_str());
 156         return -1;
 157         }
 158     }
 159
 160 PGresult * result;
 161
 162 if (StartTransaction())
 163     {
 164     printfd(__FILE__, "POSTGRESQL_STORE::EditMessage(): 'Failed to start transaction'\n");
 165     return -1;
 166     }
 167
 168 std::string elogin = login;
 169 std::string etext = msg.text;
 170
 171 if (EscapeString(elogin))
 172     {
 173     printfd(__FILE__, "POSTGRESQL_STORE::EditMessage(): 'Failed to escape login'\n");
 174     if (RollbackTransaction())
 175         {
 176         printfd(__FILE__, "POSTGRESQL_STORE::EditMessage(): 'Failed to rollback transaction'\n");
 177         }
 178     return -1;
 179     }
 180
 181 if (EscapeString(etext))
 182     {
 183     printfd(__FILE__, "POSTGRESQL_STORE::EditMessage(): 'Failed to escape message text'\n");
 184     if (RollbackTransaction())
 185         {
 186         printfd(__FILE__, "POSTGRESQL_STORE::EditMessage(): 'Failed to rollback transaction'\n");
 187         }
 188     return -1;
 189     }
 190
 191 std::ostringstream query;
 192 query << "UPDATE tb_messages SET "
 193           << "fk_user = (SELECT pk_user FROM tb_users WHERE name = '" << elogin << "'), "
 194           << "ver = " << msg.header.ver << ", "
 195           << "msg_type = " << msg.header.type << ", "
 196           << "last_send_time = CAST('" << formatTime(msg.header.lastSendTime) << "' AS TIMESTAMP), "
 197           << "creation_time = CAST('" << formatTime(msg.header.creationTime) << "' AS TIMESTAMP), "
 198           << "show_time = " << msg.header.showTime << ", "
 199           << "repeat = " << msg.header.repeat << ", "
 200           << "repeat_period = " << msg.header.repeatPeriod << ", "
 201           << "msg_text = '" << etext << "' "
 202       << "WHERE pk_message = " << msg.header.id;
 203
 204 result = PQexec(connection, query.str().c_str());
 205
 206 if (PQresultStatus(result) != PGRES_COMMAND_OK)
 207     {
 208     strError = PQresultErrorMessage(result);
 209     PQclear(result);
 210     printfd(__FILE__, "POSTGRESQL_STORE::EditMessage(): '%s'\n", strError.c_str());
 211     if (RollbackTransaction())
 212         {
 213         printfd(__FILE__, "POSTGRESQL_STORE::EditMessage(): 'Failed to rollback transaction'\n");
 214         }
 215     return -1;
 216     }
 217
 218 PQclear(result);
 219
 220 if (CommitTransaction())
 221     {
 222     printfd(__FILE__, "POSTGRESQL_STORE::EditMessage(): 'Failed to commit transaction'\n");
 223     return -1;
 224     }
 225
 226 return 0;
 227 }
 228 //-----------------------------------------------------------------------------
 229 int POSTGRESQL_STORE::GetMessage(uint64_t id,
 230                                STG_MSG * msg,
 231                                const std::string &) const
 232 {
 233 STG_LOCKER lock(&mutex);
 234
 235 if (PQstatus(connection) != CONNECTION_OK)
 236     {
 237     printfd(__FILE__, "POSTGRESQL_STORE::GetMessage(): 'Connection lost. Trying to reconnect...'\n", strError.c_str());
 238     if (Reset())
 239         {
 240         strError = "Connection lost";
 241         printfd(__FILE__, "POSTGRESQL_STORE::GetMessage(): '%s'\n", strError.c_str());
 242         return -1;
 243         }
 244     }
 245
 246 PGresult * result;
 247
 248 if (StartTransaction())
 249     {
 250     printfd(__FILE__, "POSTGRESQL_STORE::GetMessage(): 'Failed to start transaction'\n");
 251     return -1;
 252     }
 253
 254 std::ostringstream query;
 255 query << "SELECT ver, msg_type, last_send_time, \
 256                  creation_time, show_time, repeat, \
 257                  repeat_period, msg_text \
 258           FROM tb_messages \
 259           WHERE pk_message = " << id;
 260
 261 result = PQexec(connection, query.str().c_str());
 262
 263 if (PQresultStatus(result) != PGRES_TUPLES_OK)
 264     {
 265     strError = PQresultErrorMessage(result);
 266     PQclear(result);
 267     printfd(__FILE__, "POSTGRESQL_STORE::GetMessage(): '%s'\n", strError.c_str());
 268     if (RollbackTransaction())
 269         {
 270         printfd(__FILE__, "POSTGRESQL_STORE::GetMessage(): 'Failed to rollback transaction'\n");
 271         }
 272     return -1;
 273     }
 274
 275 int tuples = PQntuples(result);
 276
 277 if (tuples != 1)
 278     {
 279     strError = "Failed to fetch message data";
 280     printfd(__FILE__, "POSTGRESQL_STORE::GetMessage(): 'Invalid number of tuples. Wanted 1, actulally %d'\n", tuples);
 281     PQclear(result);
 282     if (RollbackTransaction())
 283         {
 284         printfd(__FILE__, "POSTGRESQL_STORE::GetMessage(): 'Failed to rollback transaction'\n");
 285         }
 286     return -1;
 287     }
 288
 289 str2x(PQgetvalue(result, 0, 0), msg->header.ver);
 290 str2x(PQgetvalue(result, 0, 1), msg->header.type);
 291 msg->header.lastSendTime = static_cast<unsigned int>(readTime(PQgetvalue(result, 0, 2)));
 292 msg->header.creationTime = static_cast<unsigned int>(readTime(PQgetvalue(result, 0, 3)));
 293 str2x(PQgetvalue(result, 0, 4), msg->header.showTime);
 294 str2x(PQgetvalue(result, 0, 5), msg->header.repeat);
 295 str2x(PQgetvalue(result, 0, 6), msg->header.repeatPeriod);
 296 msg->text = PQgetvalue(result, 0, 7);
 297
 298 PQclear(result);
 299
 300 if (CommitTransaction())
 301     {
 302     printfd(__FILE__, "POSTGRESQL_STORE::GetMessage(): 'Failed to commit transaction'\n");
 303     return -1;
 304     }
 305
 306 return 0;
 307 }
 308 //-----------------------------------------------------------------------------
 309 int POSTGRESQL_STORE::DelMessage(uint64_t id, const std::string &) const
 310 {
 311 STG_LOCKER lock(&mutex);
 312
 313 if (PQstatus(connection) != CONNECTION_OK)
 314     {
 315     printfd(__FILE__, "POSTGRESQL_STORE::DelMessage(): 'Connection lost. Trying to reconnect...'\n", strError.c_str());
 316     if (Reset())
 317         {
 318         strError = "Connection lost";
 319         printfd(__FILE__, "POSTGRESQL_STORE::DelMessage(): '%s'\n", strError.c_str());
 320         return -1;
 321         }
 322     }
 323
 324 PGresult * result;
 325
 326 if (StartTransaction())
 327     {
 328     printfd(__FILE__, "POSTGRESQL_STORE::DelMessage(): 'Failed to start transaction'\n");
 329     return -1;
 330     }
 331
 332 std::ostringstream query;
 333 query << "DELETE FROM tb_messages WHERE pk_message = " << id;
 334
 335 result = PQexec(connection, query.str().c_str());
 336
 337 if (PQresultStatus(result) != PGRES_COMMAND_OK)
 338     {
 339     strError = PQresultErrorMessage(result);
 340     PQclear(result);
 341     printfd(__FILE__, "POSTGRESQL_STORE::DelMessage(): '%s'\n", strError.c_str());
 342     if (RollbackTransaction())
 343         {
 344         printfd(__FILE__, "POSTGRESQL_STORE::DelMessage(): 'Failed to rollback transaction'\n");
 345         }
 346     return -1;
 347     }
 348
 349 PQclear(result);
 350
 351 if (CommitTransaction())
 352     {
 353     printfd(__FILE__, "POSTGRESQL_STORE::DelMessage(): 'Failed to commit transaction'\n");
 354     return -1;
 355     }
 356
 357 return 0;
 358 }
 359 //-----------------------------------------------------------------------------
 360 int POSTGRESQL_STORE::GetMessageHdrs(std::vector<STG_MSG_HDR> * hdrsList,
 361                                    const std::string & login) const
 362 {
 363 STG_LOCKER lock(&mutex);
 364
 365 if (PQstatus(connection) != CONNECTION_OK)
 366     {
 367     printfd(__FILE__, "POSTGRESQL_STORE::GetMessageHdrs(): 'Connection lost. Trying to reconnect...'\n", strError.c_str());
 368     if (Reset())
 369         {
 370         strError = "Connection lost";
 371         printfd(__FILE__, "POSTGRESQL_STORE::GetMessageHdrs(): '%s'\n", strError.c_str());
 372         return -1;
 373         }
 374     }
 375
 376 PGresult * result;
 377
 378 if (StartTransaction())
 379     {
 380     printfd(__FILE__, "POSTGRESQL_STORE::GetMessageHdrs(): 'Failed to start transaction'\n");
 381     return -1;
 382     }
 383
 384 std::string elogin = login;
 385
 386 if (EscapeString(elogin))
 387     {
 388     printfd(__FILE__, "POSTGRESQL_STORE::GetMessageHdrs(): 'Failed to escape login'\n");
 389     if (RollbackTransaction())
 390         {
 391         printfd(__FILE__, "POSTGRESQL_STORE::GetMessageHdrs(): 'Failed to rollback transaction'\n");
 392         }
 393     return -1;
 394     }
 395
 396 std::ostringstream query;
 397 query << "SELECT pk_message, ver, msg_type, \
 398                  last_send_time, creation_time, show_time, \
 399                  repeat, repeat_period \
 400           FROM tb_messages \
 401           WHERE fk_user IN \
 402                 (SELECT pk_user FROM tb_users \
 403           WHERE name = '" << elogin << "')";
 404
 405 result = PQexec(connection, query.str().c_str());
 406
 407 if (PQresultStatus(result) != PGRES_TUPLES_OK)
 408     {
 409     strError = PQresultErrorMessage(result);
 410     PQclear(result);
 411     printfd(__FILE__, "POSTGRESQL_STORE::GetMessageHdrs(): '%s'\n", strError.c_str());
 412     if (RollbackTransaction())
 413         {
 414         printfd(__FILE__, "POSTGRESQL_STORE::GetMessageHdrs(): 'Failed to rollback transaction'\n");
 415         }
 416     return -1;
 417     }
 418
 419 int tuples = PQntuples(result);
 420
 421 for (int i = 0; i < tuples; ++i)
 422     {
 423     std::stringstream tuple;
 424     STG_MSG_HDR header;
 425     tuple << PQgetvalue(result, i, 0) << " ";
 426     tuple << PQgetvalue(result, i, 1) << " ";
 427     tuple << PQgetvalue(result, i, 2) << " ";
 428     header.lastSendTime = static_cast<unsigned int>(readTime(PQgetvalue(result, i, 3)));
 429     header.creationTime = static_cast<unsigned int>(readTime(PQgetvalue(result, i, 4)));
 430     tuple << PQgetvalue(result, i, 5) << " ";
 431     tuple << PQgetvalue(result, i, 6) << " ";
 432     tuple << PQgetvalue(result, i, 7) << " ";
 433
 434     tuple >> header.id;
 435     tuple >> header.ver;
 436     tuple >> header.type;
 437     tuple >> header.showTime;
 438     tuple >> header.repeat;
 439     tuple >> header.repeatPeriod;
 440     hdrsList->push_back(header);
 441     }
 442
 443 PQclear(result);
 444
 445 if (CommitTransaction())
 446     {
 447     printfd(__FILE__, "POSTGRESQL_STORE::GetMessageHdrs(): 'Failed to commit transaction'\n");
 448     return -1;
 449     }
 450
 451 return 0;
 452 }
 453 //-----------------------------------------------------------------------------
 454