Merge pull request #32 from EduFrazao/async-callback-threads

Author: Oscar Franco

android/CMakeLists.txt

@@ -49,6 +49,10 @@ add_library(
   ../cpp/JSIHelper.cpp
   ../cpp/ThreadPool.h
   ../cpp/ThreadPool.cpp
+  ../cpp/sqlfileloader.h
+  ../cpp/sqlfileloader.cpp
+  ../cpp/sqlbatchexecutor.h
+  ../cpp/sqlbatchexecutor.cpp
   cpp-adapter.cpp
 )
 

cpp/installer.cpp

@@ -12,9 +12,9 @@
 #include "logs.h"
 #include "JSIHelper.h"
 #include "ThreadPool.h"
+#include "sqlfileloader.h"
+#include "sqlbatchexecutor.h"
 #include <vector>
-#include <iostream>
-#include <fstream>
 #include <string>
 
 using namespace std;
@@ -23,55 +23,6 @@ using namespace facebook;
 string docPathStr;
 std::shared_ptr<react::CallInvoker> invoker;
 
-/**
- * Local function to handle SQL File Import in order to reuse with Sync and Async operations
- */
-SequelBatchOperationResult importSQLFile(string dbName, string fileLocation)
-{
-  string line;
-  ifstream sqFile(fileLocation);
-  if (sqFile.is_open())
-  {
-    try
-    {
-      int affectedRows = 0;
-      int commands = 0;
-      sequel_execute_literal_update(dbName, "BEGIN EXCLUSIVE TRANSACTION");
-      while (std::getline(sqFile, line, '\n'))
-      {
-        if (!line.empty())
-        {
-          SequelLiteralUpdateResult result = sequel_execute_literal_update(dbName, line);
-          if (result.type == SequelResultError)
-          {
-            sequel_execute_literal_update(dbName, "ROLLBACK");
-            sqFile.close();
-            return {SequelResultError, result.message, 0, commands};
-          }
-          else
-          {
-            affectedRows += result.affectedRows;
-            commands++;
-          }
-        }
-      }
-      sqFile.close();
-      sequel_execute_literal_update(dbName, "COMMIT");
-      return {SequelResultOk, "", affectedRows, commands};
-    }
-    catch (...)
-    {
-      sqFile.close();
-      sequel_execute_literal_update(dbName, "ROLLBACK");
-      return {SequelResultError, "[react-native-quick-sqlite][loadSQLFile] Unexpected error, transaction was rolledback", 0, 0};
-    }
-  }
-  else
-  {
-    return {SequelResultError, "[react-native-quick-sqlite][loadSQLFile] Could not open file", 0, 0};
-  }
-}
-
 jsi::Object createError(jsi::Runtime &rt, string message)
 {
   auto res = jsi::Object(rt);
@@ -235,7 +186,7 @@ void install(jsi::Runtime &rt, std::shared_ptr<react::CallInvoker> jsCallInvoker
 
         // Converting results into a JSI Response
         auto jsiResult = createSequelQueryExecutionResult(rt, status, &results);
-        return jsiResult;
+        return move(jsiResult);
       });
 
   // Execute a batch of SQL queries in a transaction
@@ -251,80 +202,90 @@ void install(jsi::Runtime &rt, std::shared_ptr<react::CallInvoker> jsCallInvoker
           return createError(rt, "[react-native-quick-sqlite][execSQLBatch] - Incorrect parameter count");
         }
 
-        const string dbName = args[0].asString(rt).utf8(rt);
         const jsi::Value &params = args[1];
         if (params.isNull() || params.isUndefined())
         {
           return createError(rt, "[react-native-quick-sqlite][execSQLBatch] - An array of SQL commands or parameters is needed");
         }
+        const string dbName = args[0].asString(rt).utf8(rt);
+        const jsi::Array &batchParams = params.asObject(rt).asArray(rt);
+        vector<QuickQueryArguments> commands;
+        jsiBatchParametersToQuickArguments(rt, batchParams, &commands);
+
+        auto batchResult = executeBatch(dbName, &commands);
+        if(batchResult.type == SequelResultOk)
+        {
+          auto res = jsi::Object(rt);
+          res.setProperty(rt, "status", jsi::Value(0));
+          res.setProperty(rt, "rowsAffected", jsi::Value(batchResult.affectedRows));
+          return move(res);
+        } else
+        {
+          return createError(rt, batchResult.message);
+        }
+      });
+
+  auto execSQLBatchAsync = jsi::Function::createFromHostFunction(
+      rt,
+      jsi::PropNameID::forAscii(rt, "sequel_execSQLBatchAsync"),
+      3,
+      [pool](jsi::Runtime &rt, const jsi::Value &thisValue, const jsi::Value *args, size_t count) -> jsi::Value
+      {
+        if (sizeof(args) < 3)
+        {
+          jsi::detail::throwJSError(rt, "[react-native-quick-sqlite][asyncExecuteSqlBatch] Incorrect parameter count");
+          return {};
+        }
+
+        const jsi::Value &params = args[1];
+        const jsi::Value &callbackHolder = args[2];
+        if(!callbackHolder.isObject() || !callbackHolder.asObject(rt).isFunction(rt)) {
+          jsi::detail::throwJSError(rt, "[react-native-quick-sqlite][asyncExecuteSqlBatch] The callback argument must be a function");
+          return {};
+        }
+
+        if (params.isNull() || params.isUndefined())
+        {
+          jsi::detail::throwJSError(rt, "[react-native-quick-sqlite][asyncExecuteSqlBatch] - An array of SQL commands or parameters is needed");
+          return {};
+        }
 
-        int rowsAffected = 0;
+        const string dbName = args[0].asString(rt).utf8(rt);
         const jsi::Array &batchParams = params.asObject(rt).asArray(rt);
-        try
+        auto callback = make_shared<jsi::Value>(callbackHolder.asObject(rt));
+
+        vector<QuickQueryArguments> commands;
+        jsiBatchParametersToQuickArguments(rt, batchParams, &commands);
+
+        auto task =
+            [&rt, dbName, commands = make_shared<vector<QuickQueryArguments>>(commands), callback]()
         {
-          sequel_execute(rt, dbName, "BEGIN TRANSACTION", jsi::Value::undefined());
-          for (int i = 0; i < batchParams.length(rt); i++)
+          try
           {
-            const jsi::Array &command = batchParams.getValueAtIndex(rt, i).asObject(rt).asArray(rt);
-            if (command.length(rt) == 0)
-            {
-              sequel_execute(rt, dbName, "ROLLBACK", jsi::Value::undefined());
-              return createError(rt, "[react-native-quick-sqlite][execSQLBatch] - No SQL Commands found on batch index " + std::to_string(i));
-            }
-            const string query = command.getValueAtIndex(rt, 0).asString(rt).utf8(rt);
-            const jsi::Value &commandParams = command.length(rt) > 1 ? command.getValueAtIndex(rt, 1) : jsi::Value::undefined();
-            if (!commandParams.isUndefined() && commandParams.asObject(rt).isArray(rt) && commandParams.asObject(rt).asArray(rt).length(rt) > 0 && commandParams.asObject(rt).asArray(rt).getValueAtIndex(rt, 0).isObject())
-            {
-              // This arguments are an array of arrays, like a batch update of a single sql command.
-              const jsi::Array &batchUpdateParams = commandParams.asObject(rt).asArray(rt);
-              for (int x = 0; x < batchUpdateParams.length(rt); x++)
-              {
-                const jsi::Value &p = batchUpdateParams.getValueAtIndex(rt, x);
-                SequelResult result = sequel_execute(rt, dbName, query, p);
-                if (result.type == SequelResultError)
-                {
-                  sequel_execute(rt, dbName, "ROLLBACK", jsi::Value::undefined());
-                  return createError(rt, result.message.c_str());
-                }
-                else
-                {
-                  if (result.value.getObject(rt).hasProperty(rt, jsi::PropNameID::forAscii(rt, "rowsAffected")))
-                  {
-                    rowsAffected += result.value.getObject(rt).getProperty(rt, jsi::PropNameID::forAscii(rt, "rowsAffected")).asNumber();
-                  }
-                }
-              }
-            }
-            else
+            // Inside the new worker thread, we can now call sqlite operations
+            auto batchResult = executeBatch(dbName, commands.get());
+            invoker->invokeAsync([&rt, batchResult = move(batchResult), callback] 
             {
-              SequelResult result = sequel_execute(rt, dbName, query, commandParams);
-              if (result.type == SequelResultError)
+              if(batchResult.type == SequelResultOk)
               {
-                sequel_execute(rt, dbName, "ROLLBACK", jsi::Value::undefined());
-
-                return createError(rt, result.message.c_str());
-              }
-              else
+                auto res = jsi::Object(rt);
+                res.setProperty(rt, "status", jsi::Value(0));
+                res.setProperty(rt, "rowsAffected", jsi::Value(batchResult.affectedRows));
+                callback->asObject(rt).asFunction(rt).call(rt, move(res));
+              } else
               {
-                if (result.value.getObject(rt).hasProperty(rt, jsi::PropNameID::forAscii(rt, "rowsAffected")))
-                {
-                  rowsAffected += result.value.getObject(rt).getProperty(rt, jsi::PropNameID::forAscii(rt, "rowsAffected")).asNumber();
-                }
+                callback->asObject(rt).asFunction(rt).call(rt, createError(rt, batchResult.message));
               }
-            }
+            });
           }
-          sequel_execute(rt, dbName, "COMMIT", jsi::Value::undefined());
-        }
-        catch (...)
-        {
-          sequel_execute(rt, dbName, "ROLLBACK", jsi::Value::undefined());
-          return createError(rt, "[react-native-quick-sqlite][execSQLBatch] - Unexpected error");
-        }
-
-        auto res = jsi::Object(rt);
-        res.setProperty(rt, "status", jsi::Value(0));
-        res.setProperty(rt, "rowsAffected", jsi::Value(rowsAffected));
-        return move(res);
+          catch (std::exception &exc)
+          {
+            invoker->invokeAsync([&rt, callback, &exc]
+                                 { callback->asObject(rt).asFunction(rt).call(rt, createError(rt, exc.what())); });
+          }
+        };
+        pool->queueWork(task);
+        return {};
       });
 
   // Load SQL File from disk
@@ -359,9 +320,21 @@ void install(jsi::Runtime &rt, std::shared_ptr<react::CallInvoker> jsCallInvoker
       3,
       [pool](jsi::Runtime &rt, const jsi::Value &thisValue, const jsi::Value *args, size_t count) -> jsi::Value
       {
+        if (sizeof(args) < 3)
+        {
+          jsi::detail::throwJSError(rt, "[react-native-quick-sqlite][asyncLoadSqlFile] Incorrect parameter count");
+          return {};
+        }
+
+        const jsi::Value &callbackHolder = args[2];
+        if(!callbackHolder.isObject() || !callbackHolder.asObject(rt).isFunction(rt)) {
+          jsi::detail::throwJSError(rt, "[react-native-quick-sqlite][asyncLoadSqlFile] The callback argument must be a function");
+          return {};
+        }
+
         const string dbName = args[0].asString(rt).utf8(rt);
         const string sqlFileName = args[1].asString(rt).utf8(rt);
-        auto callback = make_shared<jsi::Value>((args[2].asObject(rt)));
+        auto callback = make_shared<jsi::Value>(callbackHolder.asObject(rt));
 
         auto task =
             [&rt, dbName, sqlFileName, callback]()
@@ -405,14 +378,20 @@ void install(jsi::Runtime &rt, std::shared_ptr<react::CallInvoker> jsCallInvoker
       {
         if (count < 4)
         {
-          jsi::detail::throwJSError(rt, "[react-native-quick-sqlite] Incorrect arguments for asyncExecuteSQL");
+          jsi::detail::throwJSError(rt, "[react-native-quick-sqlite][asyncExecuteSql] Incorrect arguments for asyncExecuteSQL");
+          return {};
+        }
+
+        const jsi::Value &callbackHolder = args[3];
+        if(!callbackHolder.isObject() || !callbackHolder.asObject(rt).isFunction(rt)) {
+          jsi::detail::throwJSError(rt, "[react-native-quick-sqlite][asyncExecuteSql] The callback argument must be a function");
           return {};
         }
 
         const string dbName = args[0].asString(rt).utf8(rt);
         const string query = args[1].asString(rt).utf8(rt);
         const jsi::Value &originalParams = args[2];
-        auto callback = make_shared<jsi::Value>(args[3].asObject(rt));
+        auto callback = make_shared<jsi::Value>(callbackHolder.asObject(rt));
 
         // Converting query parameters inside the javascript caller thread
         vector<QuickValue> params;
@@ -456,6 +435,7 @@ void install(jsi::Runtime &rt, std::shared_ptr<react::CallInvoker> jsCallInvoker
   module.setProperty(rt, "executeSql", move(execSQL));
   module.setProperty(rt, "asyncExecuteSql", move(asyncExecSQL));
   module.setProperty(rt, "executeSqlBatch", move(execSQLBatch));
+  module.setProperty(rt, "asyncExecuteSqlBatch", move(execSQLBatchAsync));
   module.setProperty(rt, "loadSqlFile", move(loadSQLFile));
   module.setProperty(rt, "asyncLoadSqlFile", move(loadSQLFileAsync));
 

cpp/sqlbatchexecutor.cpp

@@ -0,0 +1,89 @@
+/**
+ * Batch execution implementation
+*/
+#include "sqlbatchexecutor.h"
+
+void jsiBatchParametersToQuickArguments(jsi::Runtime &rt, jsi::Array const &batchParams, vector<QuickQueryArguments> *commands)
+{
+  for (int i = 0; i < batchParams.length(rt); i++)
+  {
+    const jsi::Array &command = batchParams.getValueAtIndex(rt, i).asObject(rt).asArray(rt);
+    if (command.length(rt) == 0)
+    {
+      continue;
+    }
+
+    const string query = command.getValueAtIndex(rt, 0).asString(rt).utf8(rt);
+    const jsi::Value &commandParams = command.length(rt) > 1 ? command.getValueAtIndex(rt, 1) : jsi::Value::undefined();
+    if (!commandParams.isUndefined() && commandParams.asObject(rt).isArray(rt) && commandParams.asObject(rt).asArray(rt).length(rt) > 0 && commandParams.asObject(rt).asArray(rt).getValueAtIndex(rt, 0).isObject())
+    {
+      // This arguments is an array of arrays, like a batch update of a single sql command.
+      const jsi::Array &batchUpdateParams = commandParams.asObject(rt).asArray(rt);
+      for (int x = 0; x < batchUpdateParams.length(rt); x++)
+      {
+        const jsi::Value &p = batchUpdateParams.getValueAtIndex(rt, x);
+        vector<QuickValue> params;
+        jsiQueryArgumentsToSequelParam(rt, p, &params);
+        commands->push_back(QuickQueryArguments{
+          query,
+          make_shared<vector<QuickValue>>(params)
+        });
+      }
+    }
+    else
+    {
+      vector<QuickValue> params;
+      jsiQueryArgumentsToSequelParam(rt, commandParams, &params);
+      commands->push_back(QuickQueryArguments{
+        query,
+        make_shared<vector<QuickValue>>(params)
+      });
+    }
+  }
+}
+
+SequelBatchOperationResult executeBatch(std::string dbName, vector<QuickQueryArguments> *commands)
+{
+  size_t commandCount = commands->size();
+  if(commandCount <= 0)
+  {
+    return SequelBatchOperationResult {
+      .type = SequelResultError,
+      .message = "No SQL commands provided",
+    };
+  }
+
+  try 
+  {
+    int affectedRows = 0;
+    sequel_execute_literal_update(dbName, "BEGIN EXCLUSIVE TRANSACTION");
+    for(int i = 0; i<commandCount; i++) {
+      auto command = commands->at(i);
+      // We do not provide a datastructure to receive query data because we don't need/want to handle this results in a batch execution
+      auto result = sequel_execute3(dbName, command.sql, command.params.get(), NULL);
+      if(result.type == SequelResultError)
+      {
+        return SequelBatchOperationResult {
+          .type = SequelResultError,
+          .message = result.errorMessage,
+        };
+      } else 
+      {
+        affectedRows += result.rowsAffected;
+      }
+    }
+    sequel_execute_literal_update(dbName, "COMMIT");
+    return SequelBatchOperationResult {
+      .type = SequelResultOk,
+      .affectedRows = affectedRows,
+      .commands = (int) commandCount,
+    };
+  } catch(std::exception &exc)
+  {
+    sequel_execute_literal_update(dbName, "ROLLBACK");
+    return SequelBatchOperationResult {
+      .type = SequelResultError,
+      .message = exc.what(),
+    };
+  }
+}