Skip to content
This repository was archived by the owner on Oct 24, 2024. It is now read-only.

Update for mongodb 4.0 nodejs driver #45

Merged
merged 4 commits into from
Dec 26, 2022
Merged

Update for mongodb 4.0 nodejs driver #45

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ed30448
Updated to mongodb nodejs 4.0.0 driver
eugeniop Jul 27, 2021
d56c9ff
Bumped version
eugeniop Jul 28, 2021
8d61d35
Handle fixed date delay
Nov 16, 2021
29fe51c
Merge pull request #1 from bizmate-oss/master
eugeniop Dec 24, 2022
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
21 changes: 12 additions & 9 deletions mongodb-queue.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ Queue.prototype.add = function(payload, opts, callback) {
opts = {}
}
var delay = opts.delay || self.delay
var visible = delay ? nowPlusSecs(delay) : now()
var visible = delay ? (delay instanceof Date ? delay.toISOString() : nowPlusSecs(delay)) : now()

var msgs = []
if (payload instanceof Array) {
Expand All @@ -92,9 +92,9 @@ Queue.prototype.add = function(payload, opts, callback) {
}

self.col.insertMany(msgs, function(err, results) {
if (err) return callback(err)
if (payload instanceof Array) return callback(null, '' + results.insertedIds)
callback(null, '' + results.ops[0]._id)
if (err) return callback(err);
if (payload instanceof Array) return callback(null, '' + results.insertedIds);
callback(null, '' + results.insertedIds["0"]);
})
}

Expand All @@ -121,8 +121,11 @@ Queue.prototype.get = function(opts, callback) {
}
}

self.col.findOneAndUpdate(query, update, { sort: sort, returnOriginal : false }, function(err, result) {
if (err) return callback(err)
self.col.findOneAndUpdate(query, update, { sort: sort, returnDocument : 'after' }, function(err, result) {
if (err){
return callback(err);
}

var msg = result.value
if (!msg) return callback()

Expand Down Expand Up @@ -152,7 +155,6 @@ Queue.prototype.get = function(opts, callback) {
return
}
}

callback(null, msg)
})
}
Expand All @@ -175,7 +177,7 @@ Queue.prototype.ping = function(ack, opts, callback) {
visible : nowPlusSecs(visibility)
}
}
self.col.findOneAndUpdate(query, update, { returnOriginal : false }, function(err, msg, blah) {
self.col.findOneAndUpdate(query, update, { returnDocument : 'after' }, function(err, msg, blah) {
if (err) return callback(err)
if ( !msg.value ) {
return callback(new Error("Queue.ping(): Unidentified ack : " + ack))
Expand All @@ -197,7 +199,8 @@ Queue.prototype.ack = function(ack, callback) {
deleted : now(),
}
}
self.col.findOneAndUpdate(query, update, { returnOriginal : false }, function(err, msg, blah) {

self.col.findOneAndUpdate(query, update, { returnDocument : 'after' }, function(err, msg, blah) {
if (err) return callback(err)
if ( !msg.value ) {
return callback(new Error("Queue.ack(): Unidentified ack : " + ack))
Expand Down
Loading