Merge branch 'master' of https://github.com/m-barthelemy/vapor-queues…

…-fluent-driver

README.md

@@ -40,7 +40,7 @@ let package = Package(
     ...
     dependencies: [
         ...
-        .package(url: "https://github.com/m-barthelemy/vapor-queues-fluent-driver.git", from: "0.3.6"),
+        .package(name: "QueuesFluentDriver", url: "https://github.com/m-barthelemy/vapor-queues-fluent-driver.git", from: "0.3.6"),
         ...
     ],
     targets: [

Sources/QueuesFluentDriver/PopQueries/SqlitePopQuery.swift

@@ -2,35 +2,30 @@ import Foundation
 import SQLKit
 import Fluent
 
-// Currently broken
 final class SqlitePop : PopQueryProtocol {
     func pop(db: Database, select: SQLExpression) -> EventLoopFuture<String?> {
-        let db = db as! SQLDatabase
-        //let beginImmediateTrxn = database.raw("BEGIN IMMEDIATE").
-
-        //return database.raw(SQLQueryString("BEGIN IMMEDIATE")).run().flatMap { void in
+        db.transaction { transaction in
+            let database = transaction as! SQLDatabase
             var id: String?
-            return db.execute(sql: select) { (row) -> Void in
+
+            return database.execute(sql: select) { (row) -> Void in
                 id = try? row.decode(column: "\(FieldKey.jobId)", as: String.self)
             }
             .flatMap {
                 guard let id = id else {
-                    return db.eventLoop.makeSucceededFuture(nil)
+                    return database.eventLoop.makeSucceededFuture(nil)
                 }
-                let updateQuery = db
+                let updateQuery = database
                     .update(JobModel.schema)
                     .set(SQLColumn("\(FieldKey.state)"), to: SQLBind(QueuesFluentJobState.processing))
                     .set(SQLColumn("\(FieldKey.updatedAt)"), to: SQLBind(Date()))
                     .where(SQLColumn("\(FieldKey.jobId)"), .equal, SQLBind(id))
                     .where(SQLColumn("\(FieldKey.state)"), .equal, SQLBind(QueuesFluentJobState.pending))
                     .query
-                return db.execute(sql: updateQuery) { (row) in }
-                    .flatMap {
-                        return db.raw(SQLQueryString("COMMIT")).run().map {
-                            return id
-                        }
-                    }
+                return database.execute(sql: updateQuery) { (row) in }
+                    .map { id }
             }
-        //}
+
+        }
     }
 }