Added initial package files.

Author: billmalarky

.gitignore

@@ -1,3 +1,4 @@
 /node_modules
 /coverage
-/.idea
+/.idea
+/reactNativeQueue.realm*

Models/Queue.js

@@ -0,0 +1,254 @@
+/**
+ *
+ * Queue Model
+ *
+ * Job Realm Schema defined in ../config/Database
+ *
+ */
+
+import Database from '../config/Database';
+import uuid from 'react-native-uuid';
+import Worker from './Worker';
+import promiseReflect from 'promise-reflect';
+
+
+export class Queue {
+
+  constructor() {
+    this.realm = null;
+    this.worker = new Worker();
+    this.status = 'inactive';
+  }
+
+  async init() {
+    if (this.realm === null) {
+      this.realm = await Database.getRealmInstance();
+    }
+  }
+
+  addWorker(jobName, worker, options = {}) {
+    this.worker.addWorker(jobName, worker, options);
+  }
+
+  removeWorker(jobName) {
+    this.worker.removeWorker(jobName);
+  }
+
+  createJob(name, payload = {}, options = {}, startQueue = true) {
+
+    if (!name) {
+      throw new Error('Job name must be supplied.');
+    }
+
+    this.realm.write(() => {
+
+      this.realm.create('Job', {
+        id: uuid.v4(),
+        name,
+        payload: JSON.stringify(payload),
+        data: JSON.stringify({
+          timeout: (options.timeout > 0) ? options.timeout : 0,
+          attempts: options.attempts || 1
+        }),
+        priority: options.priority || 0,
+        active: false,
+        created: new Date(),
+        failed: null
+      });
+
+    });
+
+    // Start queue on job creation if it isn't running by default.
+    if (startQueue && this.status == 'inactive') {
+      this.start();
+    }
+
+  }
+
+  async start() {
+
+    // If queue is already running, don't fire up concurrent loop.
+    if (this.status == 'active') {
+      return;
+    }
+
+    this.status = 'active';
+
+    let concurrentJobs = await this.getConcurrentJobs();
+
+    while (this.status == 'active' && concurrentJobs.length) {
+
+      // Loop over jobs and process them concurrently.
+      const processingJobs = concurrentJobs.map( job => {
+        return this.processJob(job);
+      });
+
+      // Promise Reflect ensures all processingJobs resolve so
+      // we don't break await early if one of the jobs fails.
+      await Promise.all(processingJobs.map(promiseReflect));
+
+      // Get next batch of jobs.
+      concurrentJobs = await this.getConcurrentJobs();
+
+    }
+
+    this.status = 'inactive';
+
+  }
+
+  stop() {
+    this.status = 'inactive';
+  }
+
+  async getJobs(sync = false) {
+
+    if (sync) {
+
+      let jobs = null;
+      this.realm.write(() => {
+
+        jobs = this.realm.objects('Job');
+
+      });
+
+      return jobs;
+
+    } else {
+      return await this.realm.objects('Job');
+    }
+
+  }
+
+  async getConcurrentJobs() {
+
+    let concurrentJobs = [];
+
+    this.realm.write(() => {
+
+      // Get next job from queue.
+      let nextJob = null;
+
+      let jobs = this.realm.objects('Job')
+        .filtered('active == FALSE AND failed == null')
+        .sorted([['priority', true], ['created', false]]);
+
+      if (jobs.length) {
+        nextJob = jobs[0];
+      }
+
+      // If next job exists, get concurrent related jobs appropriately.
+      if (nextJob) {
+
+        const concurrency = this.worker.getConcurrency(nextJob.name);
+
+        const allRelatedJobs = this.realm.objects('Job')
+          .filtered('name == "'+ nextJob.name +'" AND active == FALSE AND failed == null')
+          .sorted([['priority', true], ['created', false]]);
+
+        let jobsToMarkActive = allRelatedJobs.slice(0, concurrency);
+
+        // Grab concurrent job ids to reselect jobs as marking these jobs as active will remove
+        // them from initial selection when write transaction exits.
+        // See: https://stackoverflow.com/questions/47359368/does-realm-support-select-for-update-style-read-locking/47363356#comment81772710_47363356
+        const concurrentJobIds = jobsToMarkActive.map( job => job.id);
+
+        // Mark concurrent jobs as active
+        jobsToMarkActive = jobsToMarkActive.map( job => {
+          job.active = true;
+        });
+
+        // Reselect now-active concurrent jobs by id.
+        const query = concurrentJobIds.map( jobId => 'id == "' + jobId + '"').join(' OR ');
+        const reselectedJobs = this.realm.objects('Job')
+          .filtered(query)
+          .sorted([['priority', true], ['created', false]]);
+
+        concurrentJobs = reselectedJobs.slice(0, concurrency);
+
+      }
+
+    });
+
+    return concurrentJobs;
+
+  }
+
+  async processJob(job) {
+
+    try {
+
+      await this.worker.executeJob(job);
+
+      // On job completion, remove job
+      this.realm.write(() => {
+
+        this.realm.delete(job);
+
+      });
+
+    } catch (error) {
+
+      // Handle job failure logic, including retries.
+      this.realm.write(() => {
+
+        // Increment failed attempts number
+        let jobData = JSON.parse(job.data);
+
+        if (!jobData.failedAttempts) {
+          jobData.failedAttempts = 1;
+        } else {
+          jobData.failedAttempts++;
+        }
+
+        job.data = JSON.stringify(jobData);
+
+        // Reset active status
+        job.active = false;
+
+        // Mark job as failed if too many attempts
+        if (jobData.failedAttempts >= jobData.attempts) {
+          job.failed = new Date();
+        }
+
+      });
+
+    }
+
+  }
+
+  flushQueue(jobName = null) {
+
+    if (jobName) {
+
+      this.realm.write(() => {
+
+        let jobs = this.realm.objects('Job')
+          .filtered('name == "' + jobName + '"');
+
+        if (jobs.length) {
+          this.realm.delete(jobs);
+        }
+
+      });
+
+    } else {
+      this.realm.write(() => {
+
+        this.realm.deleteAll();
+
+      });
+    }
+
+  }
+
+
+}
+
+export default async function queueFactory() {
+
+  const queue = new Queue();
+  await queue.init();
+
+  return queue;
+
+}

Models/Worker.js

@@ -0,0 +1,63 @@
+/**
+ * Created by mayor on 11/16/17.
+ */
+
+
+export default class Worker {
+
+  static workers = {};
+
+  addWorker(jobName, worker, options = {}) {
+
+    // Attach options to worker
+    worker.options = {
+      concurrency: options.concurrency || 1
+    };
+
+    Worker.workers[jobName] = worker;
+  }
+
+  removeWorker(jobName) {
+    delete Worker.workers[jobName];
+  }
+
+  getConcurrency(jobName) {
+
+    // If no worker assigned to job name, throw error.
+    if (!Worker.workers[jobName]) {
+      throw new Error('Job ' + jobName + ' does not have a worker assigned to it.');
+    }
+
+    return Worker.workers[jobName].options.concurrency;
+
+  }
+
+  async executeJob(job) {
+
+    // If no worker assigned to job name, throw error.
+    if (!Worker.workers[job.name]) {
+      throw new Error('Job ' + job.name + ' does not have a worker assigned to it.');
+    }
+
+    // Timeout Logic
+    const jobTimeout = JSON.parse(job.data).timeout;
+
+    if (jobTimeout > 0) {
+
+      let timeoutPromise = new Promise((resolve, reject) => {
+
+        setTimeout(() => {
+          reject(new Error('TIMEOUT: Job id: ' + job.id + ' timed out in ' + jobTimeout  + 'ms.'));
+        }, jobTimeout);
+
+      });
+
+      await Promise.race([timeoutPromise, Worker.workers[job.name](job.id, JSON.parse(job.payload))]);
+
+    } else {
+      await Worker.workers[job.name](job.id, JSON.parse(job.payload));
+    }
+
+  }
+
+};

config/Database.js

@@ -0,0 +1,47 @@
+/**
+ * Realm database bootstrap
+ */
+
+import { Config } from './config';
+import Realm from 'realm';
+
+const JobSchema = {
+  name: 'Job',
+  primaryKey: 'id',
+  properties: {
+    id:  'string', // UUID.
+    name: 'string', // Job name to be matched with handler function.
+    payload: 'string', // Job payload stored as JSON.
+    data: 'string', // Store arbitrary data like "failed attempts" as JSON.
+    priority: 'int', // -5 to 5 to indicate low to high priority.
+    active: { type: 'bool', default: false},
+    created: 'date', // Job creation timestamp.
+    failed: 'date?' // Job failure timestamp (null until failure).
+  }
+};
+
+export default class Database {
+
+  static realmInstance = null; // Use a singleton connection to realm for performance.
+
+  static async getRealmInstance(options = {}) {
+
+    // Connect to realm if database singleton instance has not already been created.
+    if (Database.realmInstance === null) {
+
+      Database.realmInstance = await Realm.open({
+        path: options.realmPath || Config.REALM_PATH,
+        schemaVersion: Config.REALM_SCHEMA_VERSION,
+        schema: [JobSchema]
+
+        // Look up shouldCompactOnLaunch to auto-vacuum https://github.com/realm/realm-js/pull/1209/files
+
+      });
+
+    }
+
+    return Database.realmInstance;
+
+  }
+
+};

config/config.js

@@ -0,0 +1,10 @@
+/**
+ *
+ * Config constants
+ *
+ */
+
+export const Config = {
+  REALM_PATH: 'reactNativeQueue.realm',
+  REALM_SCHEMA_VERSION: 0 // Must be incremented if data model updates.
+};