This repository has been archived by the owner on Sep 22, 2020. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Added support for Faktory (https://github.com/contribsys/faktory)
- Loading branch information
Showing
3 changed files
with
209 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,137 @@ | ||
import './BaseDriver' | ||
import { MissingPackageError } from 'grind-framework' | ||
|
||
let Client = null | ||
let Manager = null | ||
|
||
/** | ||
* Loads the faktory-client + faktory-worker packages | ||
* or throws an error if they haven’t been added | ||
*/ | ||
function loadPackage() { | ||
if(!Client.isNil) { | ||
return | ||
} | ||
|
||
try { | ||
Manager = require('faktory-worker/lib/manager') | ||
} catch(err) { | ||
throw new MissingPackageError('faktory-worker') | ||
} | ||
|
||
try { | ||
Client = require('faktory-client') | ||
} catch(err) { | ||
throw new MissingPackageError('faktory-worker') | ||
} | ||
} | ||
|
||
/** | ||
* Faktory backed Queue Driver | ||
*/ | ||
export class FaktoryDriver extends BaseDriver { | ||
|
||
connection = null | ||
channel = null | ||
uri = null | ||
|
||
constructor(app, config) { | ||
super(app, config) | ||
|
||
loadPackage() | ||
|
||
this.config = config | ||
} | ||
|
||
connect() { | ||
this.client = new Client(this.config) | ||
return this.client.connect() | ||
} | ||
|
||
dispatch(job) { | ||
const payload = this.buildPayload(job) | ||
const at = payload.delay.isNil ? null : (new Date(Date.now() + payload.delay)) | ||
payload.try = 1 | ||
payload.queued_at = (at || new Date).getTime() | ||
|
||
return this._push(payload, at) | ||
} | ||
|
||
async listen(queues, concurrency, jobHandler, errorHandler) { | ||
this.manager = new Manager({ | ||
...this.config, | ||
queues, | ||
concurrency, | ||
registry: { | ||
'grind-job': this._receiveMessage.bind(this, jobHandler, errorHandler) | ||
} | ||
}) | ||
|
||
return this.manager.run() | ||
} | ||
|
||
async _receiveMessage(jobHandler, errorHandler, payload) { | ||
try { | ||
await jobHandler(payload) | ||
} catch(err) { | ||
try { | ||
await this._retryMessageOrRethrow(payload, err) | ||
} catch(err2) { | ||
await errorHandler(payload, err) | ||
} | ||
} | ||
} | ||
|
||
async _retryMessageOrRethrow(payload, err) { | ||
const tries = Number(payload.tries) || 1 | ||
|
||
if(tries <= 1) { | ||
throw err | ||
} | ||
|
||
const tryCount = Number(payload.try) || 1 | ||
|
||
if(tryCount >= tries) { | ||
throw err | ||
} | ||
|
||
const timeout = Number(payload.timeout) || 0 | ||
|
||
if(timeout > 0) { | ||
const at = Number(payload.at) || 0 | ||
|
||
if(at > 0 && (Date.now() + timeout) > at) { | ||
throw err | ||
} | ||
} | ||
|
||
const delay = payload.retry_delay || payload.delay | ||
const at = delay.isNil ? null : (new Date(Date.now() + delay)) | ||
payload.try = tryCount + 1 | ||
|
||
return this._push(payload, at) | ||
} | ||
|
||
_push(payload, at) { | ||
return this.client.push({ | ||
jobtype: 'grind-job', | ||
queue: payload.queue, | ||
retry: 0, | ||
at: at === null ? null : at.toISOString(), | ||
args: [ payload ] | ||
}) | ||
} | ||
|
||
async destroy() { | ||
if(!this.manager.isNil) { | ||
await this.manager.stop() | ||
this.manager = null | ||
} | ||
|
||
if(!this.client.isNil) { | ||
await this.client.close() | ||
this.client = null | ||
} | ||
} | ||
|
||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,66 @@ | ||
import { serial as test } from 'ava' | ||
|
||
import './helpers/TestJob' | ||
import './helpers/Listener' | ||
import './helpers/Service' | ||
|
||
import '../src/Drivers/FaktoryDriver' | ||
|
||
const service = new Service(test, 'faktory', { | ||
image: 'contribsys/faktory', | ||
port: 7419 | ||
}) | ||
|
||
test.beforeEach(t => { | ||
t.context.driver = new FaktoryDriver(null, { | ||
host: 'localhost', | ||
port: service.port | ||
}) | ||
|
||
return t.context.driver.connect() | ||
}) | ||
|
||
test.afterEach.always(t => t.context.driver.destroy()) | ||
|
||
test('dispatch', async t => { | ||
const payload = { time: Date.now() } | ||
const job = new TestJob({ ...payload }) | ||
|
||
await t.context.driver.dispatch(job) | ||
|
||
return Listener(t.context.driver, job => t.deepEqual(job.data.data, payload)) | ||
}) | ||
|
||
test('delayed dispatch', async t => { | ||
const payload = { time: Date.now() } | ||
const job = new TestJob({ ...payload }) | ||
|
||
const dispatchedAt = Date.now() | ||
await t.context.driver.dispatch(job.$delay(5000)) | ||
|
||
return Listener(t.context.driver, job => { | ||
// NOTE: Faktory’s spec says it will dispatch "within a few seconds" | ||
// of the delay time, so for test purposes, we can only ensure | ||
// that there was a delay at all, rather than the requested delay | ||
// as it does not appear to fully honor the request. | ||
t.is(Date.now() - dispatchedAt >= 1000, true) | ||
t.deepEqual(job.data.data, payload) | ||
}) | ||
}) | ||
|
||
test('retry dispatch', async t => { | ||
const payload = { time: Date.now() } | ||
const job = new TestJob({ ...payload }) | ||
let tries = 0 | ||
|
||
await t.context.driver.dispatch(job.$tries(2)) | ||
|
||
return Listener(t.context.driver, job => { | ||
t.is(job.tries, 2) | ||
t.deepEqual(job.data.data, payload) | ||
|
||
if(++tries === 1 || tries > 2) { | ||
throw new Error | ||
} | ||
}) | ||
}) |