@astodi/embedded-queue v0.0.3

embedded-queue

embedded-queue is job/message queue for any platform. It does not required any other repository for storing data, like Redis, MySQL, and so on.

It currently has the following repository implementations:

• nedb embedded repository,
• vanilla in-memory repository,
• or you can implement your own repository

Installation

npm install embedded-queue 

or

yarn add embedded-queue 

Basic Usage

import { Queue, InMemoryJobRepository } from 'embedded-queue'

(async () => {
    const queue = await Queue.createQueue(
        new InMemoryJobRepository()
        // or new NedbJobRepository({ /* you can pass the nedb options here */ })
    );

    // set up job processor for "adder" type, concurrency is 1
    queue.process(
        "adder",
        async (job) => job.data.a + job.data.b,
        1
    );

    // handle job complete event
    queue.on(
        EmbeddedQueue.Event.Complete,
        (job, result) => {
            console.log("Job Completed.");
            console.log(`    job.id: ${job.id}`);
            console.log(`    job.type: ${job.type}`);
            console.log(`    job.data: ${JSON.stringify(job.data)}`);
            console.log(`    result: ${result}`);
        }
    );

    // create "adder" type job
    await queue.createJob({
        type: "adder",
        data: { a: 1, b: 2 },
    });

    // shutdown queue
    setTimeout(async () => { await queue.shutdown(1000); }, 1);
})();

Basic

• Create Queue
• Set Job Processor
• Set Job Event Handler
• Create Job
• Shutdown Queue

Create Queue

You can create a new queue by calling Queue.createQueue(repository). Queue.createQueue returns a Promise, await it for initialize finish.

Set Job Processor

Job processor is a function that process single job. It is called by Worker and pass Job argument, it must return Promise<any>. It runs any process(calculation, network access, etc...) and call resolve(result). Required data can pass by Job.data object. Also you can call Job.setProgress for notify progress, Job.addLog for logging. You can set any number of job processors, each processor is associate to single job type, it processes only jobs of that type. If you want to need process many jobs that same type in concurrency, you can launch any number of job processor of same type.

Finally, queue.process method signature is quque.process(type, processor, concurrency).

Set Job Event Handler

Queue implements EventEmitter, when job is completed or job is failed, job progress updated, etc..., you can observe these events by set handlers Queue.on(Event, Handler).

Event.Complete event handler is most commonly used, it can receive job result from job processor.

Create Job

You can create a job by calling Queue.createJob(data). data argument is object that contains type, priority and data.

Queue.createJob(data) returns Promise<Job> object, this job is associated to Queue.

Priority is any of Priority.LOW, Priority.NORMAL, Priority.MEDIUM, Priority.HIGH, Priority.CRITICAL.

Shutdown Queue

If you want stop processing jobs, you have to call Queue.shutdown(timeoutMilliseconds, type) => Promise<void>. Queue starts to stop running job processor, and all job processors are stopped, Promise is resolved. If stopping job processor takes long time, after timeoutMilliseconds Queue terminate job processor, and set Job.state to State.FAILURE.
You can stop specified type job processor by passing second argument type. If undefined is passed, stop all type job processors.

API

Queue API

• createJob(data): Create a new Job, see above for usage.
• process(type, processor, concurrency): Set job processor, see above for usage.
• shutdown(timeoutMilliseconds, type): Start shutting down Queue, see above for usage.
• findJob(id): Search queue by Job.id. If found return Job, otherwise return null.
• listJobs(state): List all jobs that has specified state. If passed undefined return all jobs.
• removeJobById(id): Remove a Job from queue that specified id.
• removeJobsByCallback(callback): Remove all jobs that callback returns true. Callback signature is (job) => boolean.

Job API

• setProgress(completed, total): Set progress, arguments are convert to percentage value(completed / total).
• addLog(message): Add log.
• save(): After call it, job put into associate Queue, and waiting for process by job processor.
• remove(): Remove job from Queue, it will not be processed anymore.
• setPriority(value): Set priority value.
• isExist(): Return Job is in Queue. Before calling save() or after calling remove() returns false, otherwise true.
• Getters
  • id: String that identifies Job.
  • type: String that Identifier for select job processor.
  • data: Object that is used by job processor, you can set any data.
  • priority: Number that determines processing order.
  • createdAt: Date that job is created.
  • updatedAt: Date that job is updated.
  • startedAt: Date that job processor start process. Before job start, value is undefined.
  • completedAt: Date that job processor complete process. Before job complete or job failed, value is undefined.
  • failedAt: Date that job processor occurred error. Before job complete or job complete successfully, value is undefined.
  • state: String that represents current Job state, any of State.INACTIVE, State.ACTIVE, STATE.COMPLETE, State.FAILURE.
  • duration: Number that processing time of Job in milliseconds. Before job complete or job failed, value is undefined.
  • progress: Number that Job progress in percentage. You can set value by calling Job.setProgress. When job complete, set 100 automatically.
  • logs: Array of String. You can add log by calling Job.addLog.

Advanced

Unexpectedly Termination

If your program suddenly terminated without calling Queue.shutdown while your processor was processing jobs. These jobs remain State.ACTIVE in queue. When next time Queue.createQueue is called, these jobs are updated to State.FAILURE automatically.
If you want reprocessing these jobs, please call Queue.createJob with same parameter.

License

MIT