Try workaround for Sqlite issue with concurrent requests

m-barthelemy · Apr 6, 2020 · 6141281 · 6141281
1 parent 5470158
commit 6141281
Show file tree

Hide file tree

Showing 3 changed files with 29 additions and 45 deletions.
diff --git a/Sources/QueuesFluentDriver/PopQueries/MySQLPopQuery.swift b/Sources/QueuesFluentDriver/PopQueries/MySQLPopQuery.swift
@@ -3,7 +3,7 @@ import SQLKit
 import Fluent
 import Queues
 
-class MySQLPop : PopQueryProtocol {
+final class MySQLPop : PopQueryProtocol {
     func pop(db: Database, select: SQLExpression) -> EventLoopFuture<UUID?> {
         db.transaction { transaction in
             let database = transaction as! SQLDatabase
@@ -12,7 +12,6 @@ class MySQLPop : PopQueryProtocol {
             return database.execute(sql: select) { (row) -> Void in
                 print("••• columns: \(row.allColumns)")
                 id = try? row.decode(column: "\(FluentQueue.model.$id.key)", as: UUID.self)
-                print("••• returned id \(id)")
             }
             .flatMap {
                 if (id != nil) {
@@ -27,7 +26,6 @@ class MySQLPop : PopQueryProtocol {
                 }
                 return database.eventLoop.makeSucceededFuture(nil)
             }
-
 
         }
     }

diff --git a/Sources/QueuesFluentDriver/PopQueries/SqlitePopQuery.swift b/Sources/QueuesFluentDriver/PopQueries/SqlitePopQuery.swift
@@ -3,6 +3,32 @@ import SQLKit
 import Fluent
 import Queues
 
-final class SqlitePop : MySQLPop {
-
+final class SqlitePop : PopQueryProtocol {
+    func pop(db: Database, select: SQLExpression) -> EventLoopFuture<UUID?> {
+        let database = db as! SQLDatabase
+        return database.raw(SQLQueryString("BEGIN IMMEDIATE")).run().flatMap { void in
+            var id: UUID?
+            return database.execute(sql: select) { (row) -> Void in
+                print("••• columns: \(row.allColumns)")
+                id = try? row.decode(column: "\(FluentQueue.model.$id.key)", as: UUID.self)
+            }
+            .flatMap {
+                if (id != nil) {
+                    let updateQuery = database
+                        .update(JobModel.schema)
+                        .set(SQLColumn.init("\(FluentQueue.model.$state.key)"), to: SQLBind.init(JobState.processing))
+                        .set(SQLColumn.init("\(FluentQueue.model.$updatedAt.path.first!)"), to: SQLBind.init(Date()))
+                        .where(SQLColumn.init("\(FluentQueue.model.$id.key)"), .equal, SQLBind.init(id!))
+                        .query
+                    return database.execute(sql: updateQuery) { (row) in }
+                        .flatMap {
+                            return database.raw(SQLQueryString("COMMIT")).run().map {
+                                return id
+                            }
+                        }
+                }
+                return database.eventLoop.makeSucceededFuture(nil)
+            }
+        }
+    }
 }
diff --git a/Sources/QueuesFluentDriver/SQLDatabase+query.swift b/Sources/QueuesFluentDriver/SQLDatabase+query.swift