JobQueue

A persistent job queue with a simple API that does not depend on Operation/OperationQueue, is storage agnostic, supports for manual execution order, per job type concurrency limits, delayed jobs, and more.

Features

• No dependence on Operation or OperationQueue
• Concurrency per job type
• Manual processing order
• Delayed jobs
• Paused jobs
• Scheduled jobs
• Repeating jobs
• Rate limiting
• Custom execution sorting
• Storage agnostic
• In memory storage implementation
• YapDatabase storage implementation
• Couchbase Lite storage implementation
• Core Data storage implementation
• Realm storage implementation

Dependencies

ReactiveSwift

I didn't want to make a hard choice here, but I didn't have time to come up with a more flexible implementation. Ideally, I'd like to refactor it to have a core implementation that is interacted with via an adapter that implements the asynchronous behavior so adapters could be written for GCD, RxSwift, ReactiveSwift, Combine, etc.

Install Notes

For CouchbaseLite storage support, you must use Cocoapods to install.

Roadmap

This is alpha level code at this time. It needs more testing and profiling before being used in production.

Priorities are:

• Implement the remaining storage providers (YapDB & Realm)
• Testing
• Profiling

How to use

Create a Job

struct Email: Codable {
  var recipientAddress: String
  var senderAddress: String
  var subject: String
  var body: String
}

// If you inherit from the `DefaultJob<T>` class, all you need to do is override and implement the `process(?)` function
class SendAnEmail: DefaultJob<Email> {
  override func process(
    details: JobDetails,
    payload: Email,
    queue: JobQueueProtocol,
    done: @escaping JobCompletion
  ) {
    send(email: payload) { error in
      if let error = error {
        done(.failure(error))
      } else {
        done(.success(()))
      }
    }
  }
}

Create a JobQueue instance

let schedulers = JobQueueSchedulers()
let storage = JobQueueInMemoryStorage()

let queue = JobQueue(
  name: "email",
  schedulers: schedulers,
  storage: storage
)

// Register the processor and allow for processing up to 4 concurrent SendEmail jobs
queue.register(SendAnEmail.self, concurrency: 4)

// Resume the queue
queue.resume().startWithCompleted {
  print("Resumed queue.")
}

Add a job

let email = Email(
  recipient: "[email protected]",
  sender: "[email protected]",
  subject: "Job Queue question",
  body: "Do you even use a job queue?"
)
let details = try! JobDetails(SendAnEmail.self, id: "email 1", queueName: queue.name, payload: email)
queue.store(details)
  .startWithCompleted {
    print("Stored a job")
  }