Skip to content

Feature/queue performance #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jacobgardner wants to merge 8 commits into
base: develop
Choose a base branch
from
Open

Feature/queue performance #5

Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
d512a96
Add index to queue when starting polling
jacobgardner Oct 9, 2019
05a1822
Merge branch 'feature/modernize-a-bit' into feature/queue-performance
jacobgardner Oct 9, 2019
ea48302
Hide queue creation behind flag
jacobgardner Oct 9, 2019
a4022bb
Extract index creation code into its own private method
jacobgardner Oct 9, 2019
61f50a4
Merge branch 'feature/modernize-a-bit' into feature/queue-performance
jacobgardner Oct 9, 2019
215c2b1
Use _getCollection in _createQueueIndex
jacobgardner Oct 9, 2019
1e85297
Merge remote-tracking branch 'origin/develop' into feature/queue-perf…
jacobgardner Oct 10, 2019
25b2bc3
Address some review feedback
jacobgardner Oct 10, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 28 additions & 4 deletions lib/message-queue.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,14 +1,14 @@
const _ = require('lodash');

// TODO... what indexes should be created and should this be responsible for creating them?

function MessageQueue() {
const self = this;

self.errorHandler = console.error;

self.databasePromise = null;
self.collectionName = '_queue';
self.createIndex = true;
self.indexName = 'poll-index';

self.pollingInterval = 1000;
self.processingTimeout = 30 * 1000;
Expand All @@ -20,7 +20,7 @@ function MessageQueue() {

self.registerWorker = function(type, promise) {
_workers[type] = promise;
_startPolling();
return _startPolling();
Comment thread
jacobgardner marked this conversation as resolved.
};

self.stopPolling = function() {
Expand Down Expand Up @@ -80,13 +80,37 @@ function MessageQueue() {

//region Private Helper Methods

function _startPolling() {
async function _startPolling() {
if (!_pollingIntervalId) {
// Temporarily assigning polling interval to prevent multiple polls starting while index is being created
_pollingIntervalId = -1;
Copy link
Copy Markdown
Member

@wonderlic-stephene wonderlic-stephene Oct 9, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure I understand this. Does the await on line 93 not actually keep line 97 from starting the polling?

Copy link
Copy Markdown
Contributor Author

@jacobgardner jacobgardner Oct 10, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's to avoid a "race condition" when registerWorker is called twice in quick succession, which is what everyone will probably be doing anyway, including us.

// _pollingIntervalId is undefined
mq.registerWorker('typeOne', (queueItem) => {...});
// _startPolling called => !_pollingIntervalId === true => createIndex awaiting
mq.registerWorker('typeTwo', (queueItem) => {...});
// _startPolling called => !_pollingIntervalId === true => createIndex awaiting again
// first createIndex finishes => _pollingIntervalId updated and setInterval called
// second createIndex finishes => _pollingIntervalId changed (first one lost) and setInterval called again.

This is a pretty awkward place to be creating the index as it is...
I think I would rather do a major version bump, have the database promise passed in through the constructor and do a createIndex once that gets fulfilled and then start the poller (if register worker has been called).

Copy link
Copy Markdown
Member

@wonderlic-stephene wonderlic-stephene Oct 10, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, that makes more sense now. Let's sleep on it and discuss further tomorrow.


if (self.createIndex) {
await _createQueueIndex();
}

// Try and find work at least once every pollingInterval
_pollingIntervalId = setInterval(_poll, self.pollingInterval);
}
}

async function _createQueueIndex() {
const collection = await _getCollection();

await collection.createIndex(
{
type: 1,
rejectedTime: 1,
nextReceivableTime: 1,
receivedTime: 1
},
{
sparse: false,
name: self.indexName
}
);
}

function _stopPolling() {
if (_pollingIntervalId) {
clearInterval(_pollingIntervalId);
Expand Down