Merge pull request #1 from gatewaytechnology/apply-runafter-support

Apply runafter support

Author: ehlersd

EDQueue/EDQueue.h

@@ -34,6 +34,8 @@ extern NSString *const EDQueueDidDrain;
 @property (nonatomic) NSUInteger retryLimit;
 
 - (void)enqueueWithData:(id)data forTask:(NSString *)task;
+- (void)enqueueWithData:(id)data forTask:(NSString *)task runAfter:(NSDate*)runAfter;
+
 - (void)start;
 - (void)stop;
 - (void)empty;

EDQueue/EDQueue.m

@@ -83,7 +83,28 @@ - (void)enqueueWithData:(id)data forTask:(NSString *)task
 }
 
 /**
- * Returns true if a job exists for this task.
+ * Adds a new scheduled job to the queue.
+ *
+ * @param {id} Data
+ * @param {NSString} Task label
+ * @param {NSDate} Run After
+ *
+ * @return {void}
+ */
+- (void)enqueueWithData:(id)data forTask:(NSString *)task runAfter:(NSDate*)runAfter
+{
+    if (!runAfter)
+    {
+        [self enqueueWithData:data forTask:task];
+        return;
+    }
+    
+    if (data == nil) data = @{};
+    [self.engine createJob:data forTask:task runAfter:runAfter];
+    [self tick];
+}
+
+/** * Returns true if a job exists for this task.
  *
  * @param {NSString} Task label
  *

EDQueue/EDQueueStorageEngine.h

@@ -14,6 +14,7 @@
 @property (retain) FMDatabaseQueue *queue;
 
 - (void)createJob:(id)data forTask:(id)task;
+- (void)createJob:(id)data forTask:(id)task runAfter:(NSDate*)runAfter;
 - (BOOL)jobExistsForTask:(id)task;
 - (void)incrementAttemptForJob:(NSNumber *)jid;
 - (void)removeJob:(NSNumber *)jid;

EDQueue/EDQueueStorageEngine.m

@@ -24,12 +24,12 @@ - (id)init
         // Database path
         NSArray *paths                  = NSSearchPathForDirectoriesInDomains(NSDocumentDirectory, NSUserDomainMask,YES);
         NSString *documentsDirectory    = [paths objectAtIndex:0];
-        NSString *path                  = [documentsDirectory stringByAppendingPathComponent:@"edqueue_0.5.0d.db"];
+        NSString *path                  = [documentsDirectory stringByAppendingPathComponent:@"edqueue_0.5.0e.db"];
 
         // Allocate the queue
         _queue                          = [[FMDatabaseQueue alloc] initWithPath:path];
         [self.queue inDatabase:^(FMDatabase *db) {
-            [db executeUpdate:@"CREATE TABLE IF NOT EXISTS queue (id INTEGER PRIMARY KEY, task TEXT NOT NULL, data TEXT NOT NULL, attempts INTEGER DEFAULT 0, stamp STRING DEFAULT (strftime('%s','now')) NOT NULL, udef_1 TEXT, udef_2 TEXT)"];
+            [db executeUpdate:@"CREATE TABLE IF NOT EXISTS queue (id INTEGER PRIMARY KEY, task TEXT NOT NULL, data TEXT NOT NULL, attempts INTEGER DEFAULT 0, stamp STRING DEFAULT (strftime('%s','now')) NOT NULL, runafter TEXT DEFAULT (strftime('%s','now')) NOT NULL, udef_1 TEXT, udef_2 TEXT)"];
             [self _databaseHadError:[db hadError] fromDatabase:db];
         }];
     }
@@ -62,6 +62,25 @@ - (void)createJob:(id)data forTask:(id)task
     }];
 }
 
+/**
+ * Creates a new scheduled job within the datastore.
+ *
+ * @param {NSString} Data (JSON string)
+ * @param {NSString} Task name
+ * @param {NSDate} Run After
+ *
+ * @return {void}
+ */
+- (void)createJob:(id)data forTask:(id)task runAfter:(NSDate*)runAfter
+{
+    NSString *dataString = [[NSString alloc] initWithData:[NSJSONSerialization dataWithJSONObject:data options:NSJSONWritingPrettyPrinted error:nil] encoding:NSUTF8StringEncoding];
+    
+    [self.queue inDatabase:^(FMDatabase *db) {
+        [db executeUpdate:@"INSERT INTO queue (task, data, runAfter) VALUES (?, ?, ?)", task, dataString, runAfter];
+        [self _databaseHadError:[db hadError] fromDatabase:db];
+    }];
+}
+
 /**
  * Tells if a job exists for the specified task name.
  *
@@ -140,7 +159,7 @@ - (NSUInteger)fetchJobCount
     __block NSUInteger count = 0;
 
     [self.queue inDatabase:^(FMDatabase *db) {
-        FMResultSet *rs = [db executeQuery:@"SELECT count(id) AS count FROM queue"];
+        FMResultSet *rs = [db executeQuery:@"SELECT count(id) AS count FROM queue WHERE runafter <= strftime('%s','now')"];
         [self _databaseHadError:[db hadError] fromDatabase:db];
 
         while ([rs next]) {
@@ -163,7 +182,7 @@ - (NSDictionary *)fetchJob
     __block id job;
 
     [self.queue inDatabase:^(FMDatabase *db) {
-        FMResultSet *rs = [db executeQuery:@"SELECT * FROM queue ORDER BY id ASC LIMIT 1"];
+        FMResultSet *rs = [db executeQuery:@"SELECT * FROM queue WHERE runafter <= strftime('%s','now') ORDER BY id ASC LIMIT 1"];
         [self _databaseHadError:[db hadError] fromDatabase:db];
 
         while ([rs next]) {
@@ -188,7 +207,7 @@ - (NSDictionary *)fetchJobForTask:(id)task
     __block id job;
 
     [self.queue inDatabase:^(FMDatabase *db) {
-        FMResultSet *rs = [db executeQuery:@"SELECT * FROM queue WHERE task = ? ORDER BY id ASC LIMIT 1", task];
+        FMResultSet *rs = [db executeQuery:@"SELECT * FROM queue WHERE task = ? AND runafter <= strftime('%s','now')ORDER BY id ASC LIMIT 1", task];
         [self _databaseHadError:[db hadError] fromDatabase:db];
 
         while ([rs next]) {
@@ -210,7 +229,8 @@ - (NSDictionary *)_jobFromResultSet:(FMResultSet *)rs
         @"task":        [rs stringForColumn:@"task"],
         @"data":        [NSJSONSerialization JSONObjectWithData:[[rs stringForColumn:@"data"] dataUsingEncoding:NSUTF8StringEncoding] options:NSJSONReadingMutableContainers error:nil],
         @"attempts":    [NSNumber numberWithInt:[rs intForColumn:@"attempts"]],
-        @"stamp":       [rs stringForColumn:@"stamp"]
+        @"stamp":       [rs stringForColumn:@"stamp"],
+        @"runafter":    [rs dateForColumn:@"runafter"]
     };
     return job;
 }