transport/sqlite/src/aaSqlitePump.cpp

   1 /* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
   2 /* vim:set ts=2 sw=2 sts=2 et cindent tw=79 ft=cpp: */
   3 /*
   4  * Copyright (C) 2008 Sergey Yanovich <ynvich@gmail.com>
   5  *
   6  * This program is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU General Public License as
   8  * published by the Free Software Foundation; either version 2 of the
   9  * License, or (at your option) any later version.
  10  *
  11  * This program is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU General Public
  17  * License along with this program; if not, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 02111-1307, USA.
  20  */
  21
  22 #include "xpcom-config.h"
  23
  24 #include "sqlite3.h"
  25 #include "nsCOMPtr.h"
  26 #include "nsAutoPtr.h"
  27 #include "nsComponentManagerUtils.h"
  28 #include "nsServiceManagerUtils.h"
  29 #include "nsStringAPI.h"
  30 #include "nsNetCID.h"
  31 #include "nsIAsyncInputStream.h"
  32 #include "nsIAsyncOutputStream.h"
  33 #include "nsIPipe.h"
  34 #include "nsIStreamListener.h"
  35
  36 /* Unfrozen API */
  37 #include "nsThreadUtils.h"
  38 #include "nsIRunnable.h"
  39
  40 #include "aaSqlitePump.h"
  41 #include "aaSqliteChannel.h"
  42 #include "aaTransportXML.h"
  43 #include "aaTransportUtils.h"
  44
  45 class aaStartEvent : public nsIRunnable
  46 {
  47 public:
  48   aaStartEvent(aaSqliteChannel *channel)
  49     :mChannel(channel) {}
  50   NS_DECL_ISUPPORTS
  51   NS_DECL_NSIRUNNABLE
  52 private:
  53   nsRefPtr<aaSqliteChannel> mChannel;
  54 private:
  55   ~aaStartEvent() {}
  56 };
  57
  58 NS_IMPL_THREADSAFE_ISUPPORTS1(aaStartEvent,
  59                               nsIRunnable)
  60
  61 NS_IMETHODIMP
  62 aaStartEvent::Run()
  63 {
  64   NS_ENSURE_STATE(mChannel);
  65   return mChannel->OnStartRequest(nsnull, nsnull);
  66 }
  67
  68 class aaDataEvent : public nsIRunnable
  69 {
  70 public:
  71   aaDataEvent(aaSqliteChannel *channel, nsIInputStream *stream, PRUint32
  72       offset, PRUint32 count)
  73     :mChannel(channel), mStream(stream), mOffset(offset), mCount(count) {}
  74   NS_DECL_ISUPPORTS
  75   NS_DECL_NSIRUNNABLE
  76 private:
  77   nsRefPtr<aaSqliteChannel> mChannel;
  78   nsCOMPtr<nsIInputStream> mStream;
  79   PRUint32 mOffset;
  80   PRUint32 mCount;
  81 private:
  82   ~aaDataEvent() {}
  83 };
  84
  85 NS_IMPL_THREADSAFE_ISUPPORTS1(aaDataEvent,
  86                               nsIRunnable)
  87
  88 NS_IMETHODIMP
  89 aaDataEvent::Run()
  90 {
  91   NS_ENSURE_STATE(mChannel);
  92   if (mChannel->IsCancelled())
  93     return NS_OK;
  94
  95   return mChannel->OnDataAvailable(nsnull, nsnull, mStream,
  96       mOffset, mCount);
  97 }
  98
  99 class aaStopEvent : public nsIRunnable
 100 {
 101 public:
 102   aaStopEvent(aaSqliteChannel *channel, nsresult status)
 103     :mChannel(channel), mStatus(status) {}
 104   NS_DECL_ISUPPORTS
 105   NS_DECL_NSIRUNNABLE
 106 private:
 107   nsRefPtr<aaSqliteChannel> mChannel;
 108   nsresult mStatus;
 109 private:
 110   ~aaStopEvent() {}
 111 };
 112
 113 NS_IMPL_THREADSAFE_ISUPPORTS1(aaStopEvent,
 114                               nsIRunnable)
 115
 116 NS_IMETHODIMP
 117 aaStopEvent::Run()
 118 {
 119   NS_ENSURE_STATE(mChannel);
 120   return mChannel->OnStopRequest(nsnull, nsnull, mStatus);
 121 }
 122
 123 aaSqlitePump::aaSqlitePump(nsIFile *dbFile, const char *query, aaSqliteChannel
 124     *channel)
 125   :mDBFile(dbFile), mQuery(query), mChannel(channel), mOffset(0), mCount(0),
 126   mRowIndex(0)
 127 {
 128   mTarget = do_GetCurrentThread();
 129 }
 130
 131 NS_IMPL_THREADSAFE_ISUPPORTS1(aaSqlitePump,
 132                               nsIRunnable)
 133
 134 NS_IMETHODIMP
 135 aaSqlitePump::Run()
 136 {
 137   nsresult rv;
 138
 139   {
 140     nsRefPtr<aaStartEvent> event = new aaStartEvent(mChannel);
 141     mTarget->Dispatch(event, NS_DISPATCH_NORMAL);
 142   }
 143
 144   rv = start();
 145
 146   PRInt32 code = 0;
 147   db.ErrorCode(&code);
 148   if (NS_FAILED(rv) && SQLITE_OK != code)
 149   {
 150     feed(kTransportXmlHeader);
 151     feedError();
 152     feed(kTransportXmlFooter);
 153   }
 154
 155   if (NS_SUCCEEDED(rv))
 156     rv = read();
 157
 158   {
 159     nsRefPtr<aaStopEvent> event = new aaStopEvent(mChannel, rv);
 160     rv = mTarget->Dispatch(event, NS_DISPATCH_NORMAL);
 161     NS_ENSURE_SUCCESS(rv, rv);
 162   }
 163
 164   return NS_OK;
 165 }
 166
 167 nsresult
 168 aaSqlitePump::write(const char *data)
 169 {
 170   return writen(data, strlen(data));
 171 }
 172
 173 nsresult
 174 aaSqlitePump::writen(const char *data, PRUint32 length)
 175 {
 176   NS_ENSURE_STATE(mSink);
 177   nsresult rv;
 178
 179   PRUint32 writtenCount;
 180   rv = mSink->Write(data, length, &writtenCount);
 181   NS_ENSURE_SUCCESS(rv, rv);
 182
 183   mCount += writtenCount;
 184   return NS_OK;
 185 }
 186
 187 static const char kRaw[] = "<>&";
 188 static const char* kEscaped[] =
 189 {
 190   "&lt;"
 191   ,"&rt;"
 192   ,"&amp;"
 193 };
 194
 195 nsresult
 196 aaSqlitePump::writeAndEscape(const char *data)
 197 {
 198   nsresult rv;
 199
 200   PRUint32 i = 0;
 201   const char *last = data;
 202
 203   while (NS_LIKELY(0 != last[i++]));
 204   {
 205     PRUint32 y = 0;
 206     for (; y < 3; y++)
 207     {
 208       if (NS_UNLIKELY(kRaw[y] == last[i - 1]))
 209       {
 210         if (i > 1) {
 211           rv = writen(last, i - 1);
 212           NS_ENSURE_SUCCESS(rv, rv);
 213         }
 214         rv = write(kEscaped[y]);
 215         NS_ENSURE_SUCCESS(rv, rv);
 216
 217         last = &last[i];
 218         i = 0;
 219       }
 220     }
 221   }
 222
 223   rv = writen(last, i - 1);
 224   NS_ENSURE_SUCCESS(rv, rv);
 225
 226   return NS_OK;
 227 }
 228
 229 nsresult
 230 aaSqlitePump::notifyListener()
 231 {
 232   NS_ENSURE_STATE(mTarget);
 233   nsresult rv;
 234
 235   nsRefPtr<aaDataEvent> event = new aaDataEvent(mChannel, mFeeder, mOffset,
 236       mCount);
 237   NS_ENSURE_TRUE(event, NS_ERROR_OUT_OF_MEMORY);
 238
 239   rv = mTarget->Dispatch(event, NS_DISPATCH_NORMAL);
 240   NS_ENSURE_SUCCESS(rv, rv);
 241
 242   mOffset += mCount;
 243   mCount = 0;
 244   return NS_OK;
 245 }
 246
 247 nsresult
 248 aaSqlitePump::feed(const char *data)
 249 {
 250   nsresult rv;
 251
 252   rv = write(data);
 253   NS_ENSURE_SUCCESS(rv, rv);
 254
 255   rv = notifyListener();
 256   NS_ENSURE_SUCCESS(rv, rv);
 257
 258   return NS_OK;
 259 }
 260
 261 nsresult
 262 aaSqlitePump::start()
 263 {
 264   NS_ENSURE_STATE(mDBFile);
 265   nsresult rv;
 266
 267   nsAutoString path;
 268   rv = mDBFile->GetPath(path);
 269   NS_ENSURE_SUCCESS(rv, rv);
 270
 271   rv = AA_CreatePipe(getter_AddRefs(mFeeder),
 272                   getter_AddRefs(mSink));
 273   NS_ENSURE_SUCCESS(rv, rv);
 274
 275   rv = db.Connect(NS_ConvertUTF16toUTF8(path).get());
 276   if (NS_SUCCEEDED(rv))
 277   {
 278     rv = db.Create(mQuery.get());
 279   }
 280
 281   return rv;
 282 }
 283
 284 nsresult
 285 aaSqlitePump::read()
 286 {
 287   nsresult rv;
 288
 289   rv = feed(kTransportXmlHeader);
 290   NS_ENSURE_SUCCESS(rv, rv);
 291
 292   rv = feedResult();
 293   feed(kTransportXmlFooter);
 294
 295   return rv;
 296 }
 297
 298 nsresult
 299 aaSqlitePump::feedResult()
 300 {
 301   nsresult rv;
 302   int rc;
 303
 304   PRBool isItemCollected = PR_FALSE;
 305   do {
 306     rc = sqlite3_step(db.GetStmt());
 307     if (rc != SQLITE_ROW) {
 308       if (rc == SQLITE_DONE || rc == SQLITE_OK)
 309         break;
 310       return feedError();
 311     }
 312
 313     rv = feedItem();
 314     NS_ENSURE_SUCCESS(rv, rv);
 315
 316     mRowIndex++;
 317     isItemCollected = PR_TRUE;
 318   } while (PR_TRUE);
 319   if (!isItemCollected)
 320   {
 321     feedChanges();
 322   }
 323
 324   return NS_OK;
 325 }
 326
 327 nsresult
 328 aaSqlitePump::feedItem()
 329 {
 330   nsresult rv;
 331   PRUint32 i, count = sqlite3_column_count(db.GetStmt());
 332   nsCAutoString xml(kTransportRowHeader);
 333
 334   xml.AppendInt(mRowIndex);
 335   xml.Append("\">\n");
 336   rv = write(xml.get());
 337   NS_ENSURE_SUCCESS(rv, rv);
 338
 339   for (i = 0; i < count; i++) {
 340     rv = feedColumn(i);
 341     NS_ENSURE_SUCCESS(rv, rv);
 342   }
 343
 344   rv = write(kTransportRowFooter);
 345   NS_ENSURE_SUCCESS(rv, rv);
 346
 347   rv = notifyListener();
 348   NS_ENSURE_SUCCESS(rv, rv);
 349
 350   return NS_OK;
 351 }
 352
 353 nsresult
 354 aaSqlitePump::feedColumn(PRUint32 index)
 355 {
 356   nsresult rv;
 357   const char *name = (const char *) sqlite3_column_name(db.GetStmt(), index);
 358   const char *text = (const char *) sqlite3_column_text(db.GetStmt(), index);
 359
 360   rv = write("    <");
 361   NS_ENSURE_SUCCESS(rv, rv);
 362
 363   rv = writeAndEscape(name);
 364   NS_ENSURE_SUCCESS(rv, rv);
 365
 366   rv = write(">");
 367   NS_ENSURE_SUCCESS(rv, rv);
 368
 369   rv = writeAndEscape(text);
 370   NS_ENSURE_SUCCESS(rv, rv);
 371
 372   rv = write("</");
 373   NS_ENSURE_SUCCESS(rv, rv);
 374
 375   rv = writeAndEscape(name);
 376   NS_ENSURE_SUCCESS(rv, rv);
 377
 378   rv = write(">\n");
 379   NS_ENSURE_SUCCESS(rv, rv);
 380
 381   return NS_OK;
 382 }
 383
 384 nsresult
 385 aaSqlitePump::feedChanges()
 386 {
 387   nsresult rv;
 388   nsCAutoString xml(kTransportMessageHeader);
 389
 390   xml.AppendLiteral("Query OK: ");
 391
 392   PRInt32 changes = 0;
 393   rv = db.Changes(&changes);
 394   NS_ENSURE_SUCCESS(rv, rv);
 395
 396   xml.AppendInt(changes);
 397   xml.AppendLiteral(" row affected");
 398   xml.AppendLiteral(kTransportMessageFooter);
 399
 400   rv = write(xml.get());
 401   NS_ENSURE_SUCCESS(rv, rv);
 402
 403   rv = notifyListener();
 404   NS_ENSURE_SUCCESS(rv, rv);
 405
 406   return NS_OK;
 407 }
 408
 409 nsresult
 410 aaSqlitePump::feedError()
 411 {
 412   nsresult rv;
 413   nsCAutoString xml(kTransportErrorHeader);
 414
 415   xml.AppendLiteral("ERROR ");
 416
 417   PRInt32 code = 0;
 418   rv = db.ErrorCode(&code);
 419   NS_ENSURE_SUCCESS(rv, rv);
 420
 421   xml.AppendInt(code);
 422   xml.AppendLiteral(": ");
 423
 424   const char *msg = nsnull;
 425   rv = db.ErrorMsg(&msg);
 426   NS_ENSURE_SUCCESS(rv, rv);
 427
 428   xml.AppendLiteral(msg);
 429   xml.AppendLiteral(kTransportErrorFooter);
 430
 431   rv = write(xml.get());
 432   NS_ENSURE_SUCCESS(rv, rv);
 433
 434   rv = notifyListener();
 435   NS_ENSURE_SUCCESS(rv, rv);
 436
 437   return NS_OK;
 438 }
 439