Queue got jobCount & better scheduling though storage requirement (fetchNextJobTimeInterval). Also, fixed critical bug - a race condition in queue. So, job might be executed two times :/ Hola TDD!

Author: gelosi

EDQueue.podspec

@@ -3,9 +3,9 @@ Pod::Spec.new do |s|
   s.version      = '1.0'
   s.license      = 'MIT'
   s.summary      = 'A persistent background job queue for iOS.'
-  s.homepage     = 'https://github.com/thisandagain/queue'
+  s.homepage     = 'https://github.com/gelosi/queue'
   s.authors      = {'Andrew Sliwinski' => '[email protected]', 'Francois Lambert' => '[email protected]', 'Oleg Shanyuk' => '[email protected]'}
-  s.source       = { :git => 'https://github.com/thisandagain/queue.git', :tag => 'v1.0' }
+  s.source       = { :git => 'https://github.com/gelosi/queue.git', :tag => 'v1.0' }
   s.platform     = :ios, '7.0'
   s.source_files = 'EDQueue'
   s.library      = 'sqlite3.0'

EDQueue/EDQueue.h

@@ -56,6 +56,8 @@ extern NSString *const EDQueueDidDrain;
 - (void)stop;
 - (void)empty;
 
+- (NSInteger)jobCount;
+
 - (BOOL)jobExistsForTag:(NSString *)tag;
 - (BOOL)jobIsActiveForTag:(NSString *)tag;
 - (nullable EDQueueJob *)nextJobForTag:(NSString *)tag;

EDQueue/EDQueue.m

@@ -36,7 +36,7 @@ + (instancetype)defaultQueue
     static dispatch_once_t onceToken;
 
     dispatch_once(&onceToken, ^{
-        EDQueueStorageEngine *fmdbBasedStorage = [[EDQueueStorageEngine alloc] initWithName:@"edqueue.default.v7.3.sqlite"];
+        EDQueueStorageEngine *fmdbBasedStorage = [[EDQueueStorageEngine alloc] initWithName:@"edqueue.default.v1.0.sqlite"];
 
         defaultQueue = [[EDQueue alloc] initWithPersistentStore:fmdbBasedStorage];
     });
@@ -56,6 +56,16 @@ - (instancetype)initWithPersistentStore:(id<EDQueuePersistentStorage>)persistent
 
 #pragma mark - Public methods
 
+/**
+ * Total number of enqueued & valid jobs.
+ *
+ * @return {NSInteger}
+ */
+- (NSInteger)jobCount
+{
+    return [self.storage jobCount];
+}
+
 /**
  * Adds a new job to the queue.
  *
@@ -152,6 +162,8 @@ - (void)stop
 - (void)empty
 {
     [self.storage removeAllJobs];
+
+    [self postNotificationOnMainThread:@{ EDQueueNameKey : EDQueueDidDrain }];
 }
 
 
@@ -164,15 +176,31 @@ - (void)empty
  */
 - (void)tick
 {
-    dispatch_queue_t gcd = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_BACKGROUND, 0);
-    dispatch_async(gcd, ^{
-        if (self.isRunning && !self.isActive && [self.storage jobCount] > 0) {
+    static dispatch_once_t onceToken;
+    static dispatch_queue_t gcd;
+
+    dispatch_once(&onceToken, ^{
+        gcd = dispatch_queue_create("edqueue.serial", DISPATCH_QUEUE_SERIAL);
+    });
+
+    dispatch_barrier_async(gcd, ^{
+
+        if (!self.isRunning) {
+            return;
+        }
+
+        if (self.isActive) {
+            return;
+        }
+
+        if ([self.storage jobCount] > 0) {
 
             id<EDQueueStorageItem> storedJob = [self.storage fetchNextJobValidForDate:[NSDate date]];
 
             if (!storedJob) {
                 __weak typeof(self) weakSelf = self;
-                dispatch_after(dispatch_time(DISPATCH_TIME_NOW, (int64_t)(5 * NSEC_PER_SEC)), dispatch_get_main_queue(), ^{
+                NSTimeInterval nextTime = [self.storage fetchNextJobTimeInterval];
+                dispatch_after(dispatch_time(DISPATCH_TIME_NOW, (int64_t)(nextTime * NSEC_PER_SEC)), dispatch_get_main_queue(), ^{
                     [weakSelf performSelectorOnMainThread:@selector(tick) withObject:nil waitUntilDone:false];
                 });
 
@@ -182,6 +210,7 @@ - (void)tick
             // Start job & Pass job to delegate
             self.activeJobTag = storedJob.job.tag;
             _isActive = YES;
+
             [self.delegate queue:self processJob:storedJob.job completion:^(EDQueueResult result) {
                 [self processJob:storedJob withResult:result];
                 self.activeJobTag = nil;
@@ -202,6 +231,7 @@ - (void)processJob:(id<EDQueueStorageItem>)storedJob withResult:(EDQueueResult)r
                                                  }];
 
             [self.storage removeJob:storedJob];
+
             break;
 
         case EDQueueResultFail:
@@ -224,6 +254,7 @@ - (void)processJob:(id<EDQueueStorageItem>)storedJob withResult:(EDQueueResult)r
             } else {
                 [self.storage removeJob:storedJob];
             }
+
             break;
         case EDQueueResultCritical:
 
@@ -234,12 +265,13 @@ - (void)processJob:(id<EDQueueStorageItem>)storedJob withResult:(EDQueueResult)r
 
             [self errorWithMessage:@"Critical error. Job canceled."];
             [self.storage removeJob:storedJob];
+
             break;
     }
 
     // Clean-up
     _isActive = NO;
-    
+
     // Drain
     if ([self.storage jobCount] == 0) {
 

EDQueue/EDQueuePersistentStorageProtocol.h

@@ -35,6 +35,7 @@ NS_ASSUME_NONNULL_BEGIN
 - (NSUInteger)jobCount;
 - (nullable id<EDQueueStorageItem>)fetchNextJobValidForDate:(NSDate *)date;
 - (nullable id<EDQueueStorageItem>)fetchNextJobForTag:(NSString *)tag validForDate:(NSDate *)date;
+- (NSTimeInterval)fetchNextJobTimeInterval;
 
 @end
 

EDQueue/EDQueueStorageEngine.m

@@ -15,6 +15,10 @@
 
 #import "EDQueueJob.h"
 
+static NSTimeInterval const DefaultJobSleepInteval = 5.0;
+static NSTimeInterval const MaxJobSleepInterval = 60.0;
+
+
 NS_ASSUME_NONNULL_BEGIN
 
 static NSString *pathForStorageName(NSString *storage)
@@ -58,6 +62,11 @@ - (instancetype)initWithTag:(NSString *)tag
     return self;
 }
 
+- (NSString *)description
+{
+    return [NSString stringWithFormat:@"%@ : id:%@, tag: %@",NSStringFromClass([self class]), _jobID, _job.tag];
+}
+
 @end
 
 @interface EDQueueStorageEngine()
@@ -239,7 +248,7 @@ - (NSUInteger)jobCount
 }
 
 /**
- * Returns the oldest job from the datastore.
+ * Returns the oldest valid job from the datastore.
  *
  * @return {id<EDQueueStorageItem>}
  */
@@ -263,28 +272,60 @@ - (NSUInteger)jobCount
 }
 
 /**
- * Returns the oldest job for the with tag from the datastore.
- *
- * @param {id} tag
+ * Returns the oldest valid job from the datastore with specific tag
  *
  * @return {id<EDQueueStorageItem>}
  */
 - (nullable id<EDQueueStorageItem>)fetchNextJobForTag:(NSString *)tag validForDate:(NSDate *)date
 {
     __block id<EDQueueStorageItem> job;
-    
+
     [self.queue inDatabase:^(FMDatabase *db) {
-        FMResultSet *rs = [db executeQuery:@"SELECT * FROM queue WHERE tag = ? AND lastAttempt <= ? AND expiration >= ? ORDER BY id ASC LIMIT 1", tag, @(date.timeIntervalSince1970), @(date.timeIntervalSince1970)];
+        NSTimeInterval timestamp = date.timeIntervalSince1970;
+        FMResultSet *rs = [db executeQuery:@"SELECT * FROM queue WHERE tag = ? AND lastAttempt <= ? AND expiration >= ? ORDER BY id ASC LIMIT 1", tag, @(timestamp), @(timestamp)];
         [self _databaseHadError:[db hadError] fromDatabase:db];
-        
+
         while ([rs next]) {
             job = [self _jobFromResultSet:rs];
         }
+
+        [rs close];
+    }];
+
+    return job;
+}
+
+/**
+ * Returns the minumum timeout for starting the next job.
+ *
+ * @param {id} tag
+ *
+ * @return {id<EDQueueStorageItem>}
+ */
+- (NSTimeInterval)fetchNextJobTimeInterval
+{
+
+    __block NSTimeInterval timeInterval = DefaultJobSleepInteval;
+
+    [self.queue inDatabase:^(FMDatabase *db) {
+
+        FMResultSet *rs = [db executeQuery:@"SELECT * FROM queue WHERE lastAttempt < expiration ORDER BY lastAttempt ASC LIMIT 1"];
+        [self _databaseHadError:[db hadError] fromDatabase:db];
+        
+        while ([rs next]) {
+            timeInterval = [rs doubleForColumn:@"lastAttempt"];
+
+            timeInterval = fabs(timeInterval - [NSDate date].timeIntervalSince1970);
+
+            if (timeInterval > MaxJobSleepInterval) {
+                timeInterval = MaxJobSleepInterval;
+            }
+        }
 
         [rs close];
     }];
 
-    return job;
+    return timeInterval;
 }
 
 #pragma mark - Private methods
@@ -298,7 +339,7 @@ - (NSUInteger)jobCount
                                                                           jobID:@([rs intForColumn:@"id"])
                                                                         atempts:@([rs intForColumn:@"attempts"])];
 
-    storedItem.job.maxRetryCount = [rs unsignedLongLongIntForColumn:@"maxAttempts"];
+    storedItem.job.maxRetryCount = [rs intForColumn:@"maxAttempts"];
     storedItem.job.retryTimeInterval = [rs doubleForColumn:@"retryTimeInterval"];
 
     NSTimeInterval expiration = [rs doubleForColumn:@"expiration"];

Project/queue/EDViewController.h

@@ -11,10 +11,12 @@
 
 @interface EDViewController : UIViewController
 
+@property (weak, nonatomic) IBOutlet UILabel *activityTitle;
 @property (nonatomic) IBOutlet UITextView *activity;
 
 - (IBAction)addSuccess:(id)sender;
 - (IBAction)addFail:(id)sender;
 - (IBAction)addCritical:(id)sender;
+- (IBAction)clearQueue:(id)sender;
 
 @end

Project/queue/EDViewController.m

@@ -43,6 +43,7 @@ - (IBAction)addSuccess:(id)sender
 - (IBAction)addFail:(id)sender
 {
     EDQueueJob *fail = [[EDQueueJob alloc] initWithTag:@"fail" userInfo:nil];
+    fail.maxRetryCount = 10;
     [[EDQueue defaultQueue] enqueueJob:fail];
 }
 
@@ -51,13 +52,20 @@ - (IBAction)addCritical:(id)sender
     EDQueueJob *critical = [[EDQueueJob alloc] initWithTag:@"critical" userInfo:nil];
     [[EDQueue defaultQueue] enqueueJob:critical];
 }
+
+- (IBAction)clearQueue:(id)sender
+{
+    [[EDQueue defaultQueue] empty];
+}
 
 #pragma mark - Notifications
 
 - (void)receivedNotification:(NSNotification *)notification
 {
     self.activity.text = [NSString stringWithFormat:@"%@%@\n", self.activity.text, notification];
     [self.activity scrollRangeToVisible:NSMakeRange([self.activity.text length], 0)];
+
+    self.activityTitle.text = [NSString stringWithFormat:@"Activity: %ld",(long)[[EDQueue defaultQueue] jobCount]];
 }
 
 #pragma mark - Dealloc