src/modules/extra/m_mssql.cpp

   1 /*               +------------------------------------+
   2  *               | Inspire Internet Relay Chat Daemon |
   3  *               +------------------------------------+
   4  *
   5  *      InspIRCd: (C) 2002-2009 InspIRCd Development Team
   6  * See: http://www.inspircd.org/wiki/index.php/Credits
   7  *
   8  * This program is free but copyrighted software; see
   9  *                        the file COPYING for details.
  10  *
  11  * ---------------------------------------------------
  12  */
  13
  14 #include "inspircd.h"
  15 #include <tds.h>
  16 #include <tdsconvert.h>
  17 #include "users.h"
  18 #include "channels.h"
  19 #include "modules.h"
  20
  21 #include "m_sqlv2.h"
  22
  23 /* $ModDesc: MsSQL provider */
  24 /* $CompileFlags: exec("grep VERSION_NO /usr/include/tdsver.h 2>/dev/null | perl -e 'print "-D_TDSVER=".((<> =~ /freetds v(\d+\.\d+)/i) ? $1*100 : 0);'") */
  25 /* $LinkerFlags: -ltds */
  26 /* $ModDep: m_sqlv2.h */
  27
  28 class SQLConn;
  29 class MsSQLResult;
  30 class ResultNotifier;
  31 class MsSQLListener;
  32 class ModuleMsSQL;
  33
  34 typedef std::map<std::string, SQLConn*> ConnMap;
  35 typedef std::deque<MsSQLResult*> ResultQueue;
  36
  37 unsigned long count(const char * const str, char a)
  38 {
  39         unsigned long n = 0;
  40         const char *p = reinterpret_cast<const char *>(str);
  41
  42         while ((p = strchr(p, a)) != NULL)
  43         {
  44                 ++p;
  45                 ++n;
  46         }
  47         return n;
  48 }
  49
  50 ResultNotifier* notifier = NULL;
  51 MsSQLListener* listener = NULL;
  52 int QueueFD = -1;
  53
  54 ConnMap connections;
  55 Mutex* QueueMutex;
  56 Mutex* ResultsMutex;
  57 Mutex* LoggingMutex;
  58
  59 class QueryThread : public Thread
  60 {
  61   private:
  62         ModuleMsSQL* Parent;
  63         InspIRCd* ServerInstance;
  64   public:
  65         QueryThread(InspIRCd* si, ModuleMsSQL* mod)
  66         : Thread(), Parent(mod), ServerInstance(si)
  67         {
  68         }
  69         ~QueryThread() { }
  70         virtual void Run();
  71 };
  72
  73 class ResultNotifier : public BufferedSocket
  74 {
  75         ModuleMsSQL* mod;
  76
  77  public:
  78         ResultNotifier(ModuleMsSQL* m, InspIRCd* SI, int newfd, char* ip) : BufferedSocket(SI, newfd, ip), mod(m)
  79         {
  80         }
  81
  82         virtual bool OnDataReady()
  83         {
  84                 char data = 0;
  85                 if (ServerInstance->SE->Recv(this, &data, 1, 0) > 0)
  86                 {
  87                         Dispatch();
  88                         return true;
  89                 }
  90                 return false;
  91         }
  92
  93         void Dispatch();
  94 };
  95
  96 class MsSQLListener : public ListenSocketBase
  97 {
  98         ModuleMsSQL* Parent;
  99         irc::sockets::insp_sockaddr sock_us;
 100         socklen_t uslen;
 101         FileReader* index;
 102
 103  public:
 104         MsSQLListener(ModuleMsSQL* P, InspIRCd* Instance, int port, const std::string &addr) : ListenSocketBase(Instance, port, addr), Parent(P)
 105         {
 106                 uslen = sizeof(sock_us);
 107                 if (getsockname(this->fd,(sockaddr*)&sock_us,&uslen))
 108                 {
 109                         throw ModuleException("Could not getsockname() to find out port number for ITC port");
 110                 }
 111         }
 112
 113         virtual void OnAcceptReady(const std::string &ipconnectedto, int nfd, const std::string &incomingip)
 114         {
 115                 new ResultNotifier(this->Parent, this->ServerInstance, nfd, (char *)ipconnectedto.c_str()); // XXX unsafe casts suck
 116         }
 117
 118         /* Using getsockname and ntohs, we can determine which port number we were allocated */
 119         int GetPort()
 120         {
 121 #ifdef IPV6
 122                 return ntohs(sock_us.sin6_port);
 123 #else
 124                 return ntohs(sock_us.sin_port);
 125 #endif
 126         }
 127 };
 128
 129
 130 class MsSQLResult : public SQLresult
 131 {
 132  private:
 133         int currentrow;
 134         int rows;
 135         int cols;
 136
 137         std::vector<std::string> colnames;
 138         std::vector<SQLfieldList> fieldlists;
 139         SQLfieldList emptyfieldlist;
 140
 141         SQLfieldList* fieldlist;
 142         SQLfieldMap* fieldmap;
 143
 144  public:
 145         MsSQLResult(Module* self, Module* to, unsigned int rid)
 146         : SQLresult(self, to, rid), currentrow(0), rows(0), cols(0), fieldlist(NULL), fieldmap(NULL)
 147         {
 148         }
 149
 150         ~MsSQLResult()
 151         {
 152         }
 153
 154         void AddRow(int colsnum, char **dat, char **colname)
 155         {
 156                 colnames.clear();
 157                 cols = colsnum;
 158                 for (int i = 0; i < colsnum; i++)
 159                 {
 160                         fieldlists.resize(fieldlists.size()+1);
 161                         colnames.push_back(colname[i]);
 162                         SQLfield sf(dat[i] ? dat[i] : "", dat[i] ? false : true);
 163                         fieldlists[rows].push_back(sf);
 164                 }
 165                 rows++;
 166         }
 167
 168         void UpdateAffectedCount()
 169         {
 170                 rows++;
 171         }
 172
 173         virtual int Rows()
 174         {
 175                 return rows;
 176         }
 177
 178         virtual int Cols()
 179         {
 180                 return cols;
 181         }
 182
 183         virtual std::string ColName(int column)
 184         {
 185                 if (column < (int)colnames.size())
 186                 {
 187                         return colnames[column];
 188                 }
 189                 else
 190                 {
 191                         throw SQLbadColName();
 192                 }
 193                 return "";
 194         }
 195
 196         virtual int ColNum(const std::string &column)
 197         {
 198                 for (unsigned int i = 0; i < colnames.size(); i++)
 199                 {
 200                         if (column == colnames[i])
 201                                 return i;
 202                 }
 203                 throw SQLbadColName();
 204                 return 0;
 205         }
 206
 207         virtual SQLfield GetValue(int row, int column)
 208         {
 209                 if ((row >= 0) && (row < rows) && (column >= 0) && (column < Cols()))
 210                 {
 211                         return fieldlists[row][column];
 212                 }
 213
 214                 throw SQLbadColName();
 215
 216                 /* XXX: We never actually get here because of the throw */
 217                 return SQLfield("",true);
 218         }
 219
 220         virtual SQLfieldList& GetRow()
 221         {
 222                 if (currentrow < rows)
 223                         return fieldlists[currentrow];
 224                 else
 225                         return emptyfieldlist;
 226         }
 227
 228         virtual SQLfieldMap& GetRowMap()
 229         {
 230                 /* In an effort to reduce overhead we don't actually allocate the map
 231                  * until the first time it's needed...so...
 232                  */
 233                 if(fieldmap)
 234                 {
 235                         fieldmap->clear();
 236                 }
 237                 else
 238                 {
 239                         fieldmap = new SQLfieldMap;
 240                 }
 241
 242                 if (currentrow < rows)
 243                 {
 244                         for (int i = 0; i < Cols(); i++)
 245                         {
 246                                 fieldmap->insert(std::make_pair(ColName(i), GetValue(currentrow, i)));
 247                         }
 248                         currentrow++;
 249                 }
 250
 251                 return *fieldmap;
 252         }
 253
 254         virtual SQLfieldList* GetRowPtr()
 255         {
 256                 fieldlist = new SQLfieldList();
 257
 258                 if (currentrow < rows)
 259                 {
 260                         for (int i = 0; i < Rows(); i++)
 261                         {
 262                                 fieldlist->push_back(fieldlists[currentrow][i]);
 263                         }
 264                         currentrow++;
 265                 }
 266                 return fieldlist;
 267         }
 268
 269         virtual SQLfieldMap* GetRowMapPtr()
 270         {
 271                 fieldmap = new SQLfieldMap();
 272
 273                 if (currentrow < rows)
 274                 {
 275                         for (int i = 0; i < Cols(); i++)
 276                         {
 277                                 fieldmap->insert(std::make_pair(colnames[i],GetValue(currentrow, i)));
 278                         }
 279                         currentrow++;
 280                 }
 281
 282                 return fieldmap;
 283         }
 284
 285         virtual void Free(SQLfieldMap* fm)
 286         {
 287                 delete fm;
 288         }
 289
 290         virtual void Free(SQLfieldList* fl)
 291         {
 292                 delete fl;
 293         }
 294
 295
 296 };
 297
 298 class SQLConn : public classbase
 299 {
 300  private:
 301         ResultQueue results;
 302         InspIRCd* ServerInstance;
 303         Module* mod;
 304         SQLhost host;
 305         TDSLOGIN* login;
 306         TDSSOCKET* sock;
 307         TDSCONTEXT* context;
 308
 309  public:
 310         QueryQueue queue;
 311
 312         SQLConn(InspIRCd* SI, Module* m, const SQLhost& hi)
 313         : ServerInstance(SI), mod(m), host(hi), login(NULL), sock(NULL), context(NULL)
 314         {
 315                 if (OpenDB())
 316                 {
 317                         std::string query("USE " + host.name);
 318                         if (tds_submit_query(sock, query.c_str()) == TDS_SUCCEED)
 319                         {
 320                                 if (tds_process_simple_query(sock) != TDS_SUCCEED)
 321                                 {
 322                                         LoggingMutex->Lock();
 323                                         ServerInstance->Logs->Log("m_mssql",DEFAULT, "WARNING: Could not select database " + host.name + " for DB with id: " + host.id);
 324                                         LoggingMutex->Unlock();
 325                                         CloseDB();
 326                                 }
 327                         }
 328                         else
 329                         {
 330                                 LoggingMutex->Lock();
 331                                 ServerInstance->Logs->Log("m_mssql",DEFAULT, "WARNING: Could not select database " + host.name + " for DB with id: " + host.id);
 332                                 LoggingMutex->Unlock();
 333                                 CloseDB();
 334                         }
 335                 }
 336                 else
 337                 {
 338                         LoggingMutex->Lock();
 339                         ServerInstance->Logs->Log("m_mssql",DEFAULT, "WARNING: Could not connect to DB with id: " + host.id);
 340                         LoggingMutex->Unlock();
 341                         CloseDB();
 342                 }
 343         }
 344
 345         ~SQLConn()
 346         {
 347                 CloseDB();
 348         }
 349
 350         SQLerror Query(SQLrequest &req)
 351         {
 352                 if (!sock)
 353                         return SQLerror(SQL_BAD_CONN, "Socket was NULL, check if SQL server is running.");
 354
 355                 /* Pointer to the buffer we screw around with substitution in */
 356                 char* query;
 357
 358                 /* Pointer to the current end of query, where we append new stuff */
 359                 char* queryend;
 360
 361                 /* Total length of the unescaped parameters */
 362                 unsigned long maxparamlen, paramcount;
 363
 364                 /* Total length of query, used for binary-safety */
 365                 unsigned long querylength = 0;
 366
 367                 /* The length of the longest parameter */
 368                 maxparamlen = 0;
 369
 370                 for(ParamL::iterator i = req.query.p.begin(); i != req.query.p.end(); i++)
 371                 {
 372                         if (i->size() > maxparamlen)
 373                                 maxparamlen = i->size();
 374                 }
 375
 376                 /* How many params are there in the query? */
 377                 paramcount = count(req.query.q.c_str(), '?');
 378
 379                 /* This stores copy of params to be inserted with using numbered params 1;3B*/
 380                 ParamL paramscopy(req.query.p);
 381
 382                 /* To avoid a lot of allocations, allocate enough memory for the biggest the escaped query could possibly be.
 383                  * sizeofquery + (maxtotalparamlength*2) + 1
 384                  *
 385                  * The +1 is for null-terminating the string
 386                  */
 387
 388                 query = new char[req.query.q.length() + (maxparamlen*paramcount*2) + 1];
 389                 queryend = query;
 390
 391                 for(unsigned long i = 0; i < req.query.q.length(); i++)
 392                 {
 393                         if(req.query.q[i] == '?')
 394                         {
 395                                 /* We found a place to substitute..what fun.
 396                                  * use mssql calls to escape and write the
 397                                  * escaped string onto the end of our query buffer,
 398                                  * then we "just" need to make sure queryend is
 399                                  * pointing at the right place.
 400                                  */
 401
 402                                 /* Is it numbered parameter?
 403                                  */
 404
 405                                 bool numbered;
 406                                 numbered = false;
 407
 408                                 /* Numbered parameter number :|
 409                                  */
 410                                 unsigned int paramnum;
 411                                 paramnum = 0;
 412
 413                                 /* Let's check if it's a numbered param. And also calculate it's number.
 414                                  */
 415
 416                                 while ((i < req.query.q.length() - 1) && (req.query.q[i+1] >= '0') && (req.query.q[i+1] <= '9'))
 417                                 {
 418                                         numbered = true;
 419                                         ++i;
 420                                         paramnum = paramnum * 10 + req.query.q[i] - '0';
 421                                 }
 422
 423                                 if (paramnum > paramscopy.size() - 1)
 424                                 {
 425                                         /* index is out of range!
 426                                          */
 427                                         numbered = false;
 428                                 }
 429
 430                                 if (numbered)
 431                                 {
 432                                         /* Custom escaping for this one. converting ' to '' should make SQL Server happy. Ugly but fast :]
 433                                          */
 434                                         char* escaped = new char[(paramscopy[paramnum].length() * 2) + 1];
 435                                         char* escend = escaped;
 436                                         for (std::string::iterator p = paramscopy[paramnum].begin(); p < paramscopy[paramnum].end(); p++)
 437                                         {
 438                                                 if (*p == '\'')
 439                                                 {
 440                                                         *escend = *p;
 441                                                         escend++;
 442                                                         *escend = *p;
 443                                                 }
 444                                                 *escend = *p;
 445                                                 escend++;
 446                                         }
 447                                         *escend = 0;
 448
 449                                         for (char* n = escaped; *n; n++)
 450                                         {
 451                                                 *queryend = *n;
 452                                                 queryend++;
 453                                         }
 454                                         delete[] escaped;
 455                                 }
 456                                 else if (req.query.p.size())
 457                                 {
 458                                         /* Custom escaping for this one. converting ' to '' should make SQL Server happy. Ugly but fast :]
 459                                          */
 460                                         char* escaped = new char[(req.query.p.front().length() * 2) + 1];
 461                                         char* escend = escaped;
 462                                         for (std::string::iterator p = req.query.p.front().begin(); p < req.query.p.front().end(); p++)
 463                                         {
 464                                                 if (*p == '\'')
 465                                                 {
 466                                                         *escend = *p;
 467                                                         escend++;
 468                                                         *escend = *p;
 469                                                 }
 470                                                 *escend = *p;
 471                                                 escend++;
 472                                         }
 473                                         *escend = 0;
 474
 475                                         for (char* n = escaped; *n; n++)
 476                                         {
 477                                                 *queryend = *n;
 478                                                 queryend++;
 479                                         }
 480                                         delete[] escaped;
 481                                         req.query.p.pop_front();
 482                                 }
 483                                 else
 484                                         break;
 485                         }
 486                         else
 487                         {
 488                                 *queryend = req.query.q[i];
 489                                 queryend++;
 490                         }
 491                         querylength++;
 492                 }
 493                 *queryend = 0;
 494                 req.query.q = query;
 495
 496                 MsSQLResult* res = new MsSQLResult((Module*)mod, req.GetSource(), req.id);
 497                 res->dbid = host.id;
 498                 res->query = req.query.q;
 499
 500                 char* msquery = strdup(req.query.q.data());
 501                 LoggingMutex->Lock();
 502                 ServerInstance->Logs->Log("m_mssql",DEBUG,"doing Query: %s",msquery);
 503                 LoggingMutex->Unlock();
 504                 if (tds_submit_query(sock, msquery) != TDS_SUCCEED)
 505                 {
 506                         std::string error("failed to execute: "+std::string(req.query.q.data()));
 507                         delete[] query;
 508                         delete res;
 509                         free(msquery);
 510                         return SQLerror(SQL_QSEND_FAIL, error);
 511                 }
 512                 delete[] query;
 513                 free(msquery);
 514
 515                 int tds_res;
 516                 while (tds_process_tokens(sock, &tds_res, NULL, TDS_TOKEN_RESULTS) == TDS_SUCCEED)
 517                 {
 518                         //ServerInstance->Logs->Log("m_mssql",DEBUG,"<******> result type: %d", tds_res);
 519                         //ServerInstance->Logs->Log("m_mssql",DEBUG,"AFFECTED ROWS: %d", sock->rows_affected);
 520                         switch (tds_res)
 521                         {
 522                                 case TDS_ROWFMT_RESULT:
 523                                         break;
 524
 525                                 case TDS_DONE_RESULT:
 526                                         if (sock->rows_affected > -1)
 527                                         {
 528                                                 for (int c = 0; c < sock->rows_affected; c++)  res->UpdateAffectedCount();
 529                                                 continue;
 530                                         }
 531                                         break;
 532
 533                                 case TDS_ROW_RESULT:
 534                                         while (tds_process_tokens(sock, &tds_res, NULL, TDS_STOPAT_ROWFMT|TDS_RETURN_DONE|TDS_RETURN_ROW) == TDS_SUCCEED)
 535                                         {
 536                                                 if (tds_res != TDS_ROW_RESULT)
 537                                                         break;
 538
 539                                                 if (!sock->current_results)
 540                                                         continue;
 541
 542                                                 if (sock->res_info->row_count > 0)
 543                                                 {
 544                                                         int cols = sock->res_info->num_cols;
 545                                                         char** name = new char*[MAXBUF];
 546                                                         char** data = new char*[MAXBUF];
 547                                                         for (int j=0; j<cols; j++)
 548                                                         {
 549                                                                 TDSCOLUMN* col = sock->current_results->columns[j];
 550                                                                 name[j] = col->column_name;
 551
 552                                                                 int ctype;
 553                                                                 int srclen;
 554                                                                 unsigned char* src;
 555                                                                 CONV_RESULT dres;
 556                                                                 ctype = tds_get_conversion_type(col->column_type, col->column_size);
 557 #if _TDSVER >= 82
 558                                                                         src = col->column_data;
 559 #else
 560                                                                         src = &(sock->current_results->current_row[col->column_offset]);
 561 #endif
 562                                                                 srclen = col->column_cur_size;
 563                                                                 tds_convert(sock->tds_ctx, ctype, (TDS_CHAR *) src, srclen, SYBCHAR, &dres);
 564                                                                 data[j] = (char*)dres.ib;
 565                                                         }
 566                                                         ResultReady(res, cols, data, name);
 567                                                 }
 568                                         }
 569                                         break;
 570
 571                                 default:
 572                                         break;
 573                         }
 574                 }
 575                 ResultsMutex->Lock();
 576                 results.push_back(res);
 577                 ResultsMutex->Unlock();
 578                 SendNotify();
 579                 return SQLerror();
 580         }
 581
 582         static int HandleMessage(const TDSCONTEXT * pContext, TDSSOCKET * pTdsSocket, TDSMESSAGE * pMessage)
 583         {
 584                 SQLConn* sc = (SQLConn*)pContext->parent;
 585                 LoggingMutex->Lock();
 586                 sc->ServerInstance->Logs->Log("m_mssql", DEBUG, "Message for DB with id: %s -> %s", sc->host.id.c_str(), pMessage->message);
 587                 LoggingMutex->Unlock();
 588                 return 0;
 589         }
 590
 591         static int HandleError(const TDSCONTEXT * pContext, TDSSOCKET * pTdsSocket, TDSMESSAGE * pMessage)
 592         {
 593                 SQLConn* sc = (SQLConn*)pContext->parent;
 594                 LoggingMutex->Lock();
 595                 sc->ServerInstance->Logs->Log("m_mssql", DEFAULT, "Error for DB with id: %s -> %s", sc->host.id.c_str(), pMessage->message);
 596                 LoggingMutex->Unlock();
 597                 return 0;
 598         }
 599
 600         void ResultReady(MsSQLResult *res, int cols, char **data, char **colnames)
 601         {
 602                 res->AddRow(cols, data, colnames);
 603         }
 604
 605         void AffectedReady(MsSQLResult *res)
 606         {
 607                 res->UpdateAffectedCount();
 608         }
 609
 610         bool OpenDB()
 611         {
 612                 CloseDB();
 613
 614                 TDSCONNECTION* conn = NULL;
 615
 616                 login = tds_alloc_login();
 617                 tds_set_app(login, "TSQL");
 618                 tds_set_library(login,"TDS-Library");
 619                 tds_set_host(login, "");
 620                 tds_set_server(login, host.host.c_str());
 621                 tds_set_server_addr(login, host.host.c_str());
 622                 tds_set_user(login, host.user.c_str());
 623                 tds_set_passwd(login, host.pass.c_str());
 624                 tds_set_port(login, host.port);
 625                 tds_set_packet(login, 512);
 626
 627                 context = tds_alloc_context(this);
 628                 context->msg_handler = HandleMessage;
 629                 context->err_handler = HandleError;
 630
 631                 sock = tds_alloc_socket(context, 512);
 632                 tds_set_parent(sock, NULL);
 633
 634                 conn = tds_read_config_info(NULL, login, context->locale);
 635
 636                 if (tds_connect(sock, conn) == TDS_SUCCEED)
 637                 {
 638                         tds_free_connection(conn);
 639                         return 1;
 640                 }
 641                 tds_free_connection(conn);
 642                 return 0;
 643         }
 644
 645         void CloseDB()
 646         {
 647                 if (sock)
 648                 {
 649                         tds_free_socket(sock);
 650                         sock = NULL;
 651                 }
 652                 if (context)
 653                 {
 654                         tds_free_context(context);
 655                         context = NULL;
 656                 }
 657                 if (login)
 658                 {
 659                         tds_free_login(login);
 660                         login = NULL;
 661                 }
 662         }
 663
 664         SQLhost GetConfHost()
 665         {
 666                 return host;
 667         }
 668
 669         void SendResults()
 670         {
 671                 while (results.size())
 672                 {
 673                         MsSQLResult* res = results[0];
 674                         ResultsMutex->Lock();
 675                         if (res->GetDest())
 676                         {
 677                                 res->Send();
 678                         }
 679                         else
 680                         {
 681                                 /* If the client module is unloaded partway through a query then the provider will set
 682                                  * the pointer to NULL. We cannot just cancel the query as the result will still come
 683                                  * through at some point...and it could get messy if we play with invalid pointers...
 684                                  */
 685                                 delete res;
 686                         }
 687                         results.pop_front();
 688                         ResultsMutex->Unlock();
 689                 }
 690         }
 691
 692         void ClearResults()
 693         {
 694                 while (results.size())
 695                 {
 696                         MsSQLResult* res = results[0];
 697                         delete res;
 698                         results.pop_front();
 699                 }
 700         }
 701
 702         void SendNotify()
 703         {
 704                 if (QueueFD < 0)
 705                 {
 706                         if ((QueueFD = socket(AF_FAMILY, SOCK_STREAM, 0)) == -1)
 707                         {
 708                                 /* crap, we're out of sockets... */
 709                                 return;
 710                         }
 711
 712                         irc::sockets::insp_sockaddr addr;
 713
 714 #ifdef IPV6
 715                         irc::sockets::insp_aton("::1", &addr.sin6_addr);
 716                         addr.sin6_family = AF_FAMILY;
 717                         addr.sin6_port = htons(listener->GetPort());
 718 #else
 719                         irc::sockets::insp_inaddr ia;
 720                         irc::sockets::insp_aton("127.0.0.1", &ia);
 721                         addr.sin_family = AF_FAMILY;
 722                         addr.sin_addr = ia;
 723                         addr.sin_port = htons(listener->GetPort());
 724 #endif
 725
 726                         if (connect(QueueFD, (sockaddr*)&addr,sizeof(addr)) == -1)
 727                         {
 728                                 /* wtf, we cant connect to it, but we just created it! */
 729                                 return;
 730                         }
 731                 }
 732                 char id = 0;
 733                 send(QueueFD, &id, 1, 0);
 734         }
 735
 736         void DoLeadingQuery()
 737         {
 738                 SQLrequest& req = queue.front();
 739                 req.error = Query(req);
 740         }
 741
 742 };
 743
 744
 745 class ModuleMsSQL : public Module
 746 {
 747  private:
 748         unsigned long currid;
 749         QueryThread* queryDispatcher;
 750
 751  public:
 752         ModuleMsSQL(InspIRCd* Me)
 753         : Module(Me), currid(0)
 754         {
 755                 LoggingMutex = ServerInstance->Mutexes->CreateMutex();
 756                 ResultsMutex = ServerInstance->Mutexes->CreateMutex();
 757                 QueueMutex = ServerInstance->Mutexes->CreateMutex();
 758
 759                 ServerInstance->Modules->UseInterface("SQLutils");
 760
 761                 if (!ServerInstance->Modules->PublishFeature("SQL", this))
 762                 {
 763                         throw ModuleException("m_mssql: Unable to publish feature 'SQL'");
 764                 }
 765
 766                 /* Create a socket on a random port. Let the tcp stack allocate us an available port */
 767 #ifdef IPV6
 768                 listener = new MsSQLListener(this, ServerInstance, 0, "::1");
 769 #else
 770                 listener = new MsSQLListener(this, ServerInstance, 0, "127.0.0.1");
 771 #endif
 772
 773                 if (listener->GetFd() == -1)
 774                 {
 775                         ServerInstance->Modules->DoneWithInterface("SQLutils");
 776                         throw ModuleException("m_mssql: unable to create ITC pipe");
 777                 }
 778                 else
 779                 {
 780                         LoggingMutex->Lock();
 781                         ServerInstance->Logs->Log("m_mssql", DEBUG, "MsSQL: Interthread comms port is %d", listener->GetPort());
 782                         LoggingMutex->Unlock();
 783                 }
 784
 785                 ReadConf();
 786
 787                 queryDispatcher = new QueryThread(ServerInstance, this);
 788                 ServerInstance->Threads->Create(queryDispatcher);
 789
 790                 ServerInstance->Modules->PublishInterface("SQL", this);
 791                 Implementation eventlist[] = { I_OnRequest, I_OnRehash };
 792                 ServerInstance->Modules->Attach(eventlist, this, 2);
 793         }
 794
 795         virtual ~ModuleMsSQL()
 796         {
 797                 delete queryDispatcher;
 798                 ClearQueue();
 799                 ClearAllConnections();
 800
 801                 ServerInstance->SE->DelFd(listener);
 802                 ServerInstance->BufferedSocketCull();
 803
 804                 if (QueueFD >= 0)
 805                 {
 806                         shutdown(QueueFD, 2);
 807                         close(QueueFD);
 808                 }
 809
 810                 if (notifier)
 811                 {
 812                         ServerInstance->SE->DelFd(notifier);
 813                         notifier->Close();
 814                         ServerInstance->BufferedSocketCull();
 815                 }
 816
 817                 ServerInstance->Modules->UnpublishInterface("SQL", this);
 818                 ServerInstance->Modules->UnpublishFeature("SQL");
 819                 ServerInstance->Modules->DoneWithInterface("SQLutils");
 820
 821                 delete LoggingMutex;
 822                 delete ResultsMutex;
 823                 delete QueueMutex;
 824         }
 825
 826
 827         void SendQueue()
 828         {
 829                 for (ConnMap::iterator iter = connections.begin(); iter != connections.end(); iter++)
 830                 {
 831                         iter->second->SendResults();
 832                 }
 833         }
 834
 835         void ClearQueue()
 836         {
 837                 for (ConnMap::iterator iter = connections.begin(); iter != connections.end(); iter++)
 838                 {
 839                         iter->second->ClearResults();
 840                 }
 841         }
 842
 843         bool HasHost(const SQLhost &host)
 844         {
 845                 for (ConnMap::iterator iter = connections.begin(); iter != connections.end(); iter++)
 846                 {
 847                         if (host == iter->second->GetConfHost())
 848                                 return true;
 849                 }
 850                 return false;
 851         }
 852
 853         bool HostInConf(const SQLhost &h)
 854         {
 855                 ConfigReader conf(ServerInstance);
 856                 for(int i = 0; i < conf.Enumerate("database"); i++)
 857                 {
 858                         SQLhost host;
 859                         host.id         = conf.ReadValue("database", "id", i);
 860                         host.host       = conf.ReadValue("database", "hostname", i);
 861                         host.port       = conf.ReadInteger("database", "port", "1433", i, true);
 862                         host.name       = conf.ReadValue("database", "name", i);
 863                         host.user       = conf.ReadValue("database", "username", i);
 864                         host.pass       = conf.ReadValue("database", "password", i);
 865                         if (h == host)
 866                                 return true;
 867                 }
 868                 return false;
 869         }
 870
 871         void ReadConf()
 872         {
 873                 ClearOldConnections();
 874
 875                 ConfigReader conf(ServerInstance);
 876                 for(int i = 0; i < conf.Enumerate("database"); i++)
 877                 {
 878                         SQLhost host;
 879
 880                         host.id         = conf.ReadValue("database", "id", i);
 881                         host.host       = conf.ReadValue("database", "hostname", i);
 882                         host.port       = conf.ReadInteger("database", "port", "1433", i, true);
 883                         host.name       = conf.ReadValue("database", "name", i);
 884                         host.user       = conf.ReadValue("database", "username", i);
 885                         host.pass       = conf.ReadValue("database", "password", i);
 886
 887                         if (HasHost(host))
 888                                 continue;
 889
 890                         this->AddConn(host);
 891                 }
 892         }
 893
 894         void AddConn(const SQLhost& hi)
 895         {
 896                 if (HasHost(hi))
 897                 {
 898                         LoggingMutex->Lock();
 899                         ServerInstance->Logs->Log("m_mssql",DEFAULT, "WARNING: A MsSQL connection with id: %s already exists. Aborting database open attempt.", hi.id.c_str());
 900                         LoggingMutex->Unlock();
 901                         return;
 902                 }
 903
 904                 SQLConn* newconn;
 905
 906                 newconn = new SQLConn(ServerInstance, this, hi);
 907
 908                 connections.insert(std::make_pair(hi.id, newconn));
 909         }
 910
 911         void ClearOldConnections()
 912         {
 913                 ConnMap::iterator iter,safei;
 914                 for (iter = connections.begin(); iter != connections.end(); iter++)
 915                 {
 916                         if (!HostInConf(iter->second->GetConfHost()))
 917                         {
 918                                 delete iter->second;
 919                                 safei = iter;
 920                                 --iter;
 921                                 connections.erase(safei);
 922                         }
 923                 }
 924         }
 925
 926         void ClearAllConnections()
 927         {
 928                 ConnMap::iterator i;
 929                 while ((i = connections.begin()) != connections.end())
 930                 {
 931                         connections.erase(i);
 932                         delete i->second;
 933                 }
 934         }
 935
 936         virtual void OnRehash(User* user, const std::string &parameter)
 937         {
 938                 QueueMutex->Lock();
 939                 ReadConf();
 940                 QueueMutex->Unlock();
 941         }
 942
 943         virtual const char* OnRequest(Request* request)
 944         {
 945                 if(strcmp(SQLREQID, request->GetId()) == 0)
 946                 {
 947                         SQLrequest* req = (SQLrequest*)request;
 948
 949                         QueueMutex->Lock();
 950
 951                         ConnMap::iterator iter;
 952
 953                         const char* returnval = NULL;
 954
 955                         if((iter = connections.find(req->dbid)) != connections.end())
 956                         {
 957                                 req->id = NewID();
 958                                 iter->second->queue.push(*req);
 959                                 returnval= SQLSUCCESS;
 960                         }
 961                         else
 962                         {
 963                                 req->error.Id(SQL_BAD_DBID);
 964                         }
 965
 966                         QueueMutex->Unlock();
 967
 968                         return returnval;
 969                 }
 970                 return NULL;
 971         }
 972
 973         unsigned long NewID()
 974         {
 975                 if (currid+1 == 0)
 976                         currid++;
 977
 978                 return ++currid;
 979         }
 980
 981         virtual Version GetVersion()
 982         {
 983                 return Version("$Id$", VF_VENDOR | VF_SERVICEPROVIDER, API_VERSION);
 984         }
 985
 986 };
 987
 988 void ResultNotifier::Dispatch()
 989 {
 990         mod->SendQueue();
 991 }
 992
 993 void QueryThread::Run()
 994 {
 995         while (this->GetExitFlag() == false)
 996         {
 997                 SQLConn* conn = NULL;
 998                 QueueMutex->Lock();
 999                 for (ConnMap::iterator i = connections.begin(); i != connections.end(); i++)
1000                 {
1001                         if (i->second->queue.totalsize())
1002                         {
1003                                 conn = i->second;
1004                                 break;
1005                         }
1006                 }
1007                 QueueMutex->Unlock();
1008                 if (conn)
1009                 {
1010                         conn->DoLeadingQuery();
1011                         QueueMutex->Lock();
1012                         conn->queue.pop();
1013                         QueueMutex->Unlock();
1014                 }
1015                 usleep(1000);
1016         }
1017 }
1018
1019 MODULE_INIT(ModuleMsSQL)