From b53d59244e4b0fbd17095027fa40eedd67284b52 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Barbosa?= Date: Mon, 24 Oct 2016 15:23:45 +0100 Subject: [PATCH 1/5] Revert "connect after event handlers are added, fix tests" This reverts commit 76c4c40b4b52d60ffc5d1ae715bc8d36605b18e9. --- circle.yml | 8 +- lib/reader.js | 5 +- package.json | 5 +- test/acceptance/connection.js | 17 +- test/acceptance/reader.js | 341 ++++++++++++++++------------------ test/acceptance/writer.js | 29 ++- test/utils.js | 22 +-- 7 files changed, 191 insertions(+), 236 deletions(-) diff --git a/circle.yml b/circle.yml index 4a36d5d..5b05ffa 100644 --- a/circle.yml +++ b/circle.yml @@ -4,14 +4,14 @@ machine: dependencies: pre: - - wget https://s3.amazonaws.com/bitly-downloads/nsq/nsq-0.3.6.linux-amd64.go1.5.1.tar.gz - - tar xvzf nsq-0.3.6.linux-amd64.go1.5.1.tar.gz - - cp nsq-0.3.6.linux-amd64.go1.5.1/bin/* . + - wget https://github.com/bitly/nsq/releases/download/v0.2.30/nsq-0.2.30.linux-amd64.go1.3.tar.gz + - tar xvzf nsq-0.2.30.linux-amd64.go1.3.tar.gz + - cp nsq-0.2.30.linux-amd64.go1.3/bin/* . - ./nsqlookupd: background: true - ./nsqd --lookupd-tcp-address=127.0.0.1:4160: background: true cache_directories: - - nsq-0.3.6.linux-amd64.go1.5.1 + - nsq-0.2.30.linux-amd64.go1.3 override: - npm install diff --git a/lib/reader.js b/lib/reader.js index adaf9bb..80b831f 100644 --- a/lib/reader.js +++ b/lib/reader.js @@ -72,10 +72,7 @@ function Reader(opts) { // mixins close(this); - var self = this; - setImmediate(function(){ - self.connect(); - }); + this.connect(); } /** diff --git a/package.json b/package.json index 43fd2b9..912638f 100644 --- a/package.json +++ b/package.json @@ -21,12 +21,11 @@ "set-component": "~1.0.0" }, "devDependencies": { - "bytes": "~0.3.0", - "jstrace": "~0.1.0", "mocha": "*", "should": "*", "superagent": "~0.17.0", - "uid": "0.0.2" + "jstrace": "~0.1.0", + "bytes": "~0.3.0" }, "license": "MIT" } diff --git a/test/acceptance/connection.js b/test/acceptance/connection.js index d95d985..96ca565 100644 --- a/test/acceptance/connection.js +++ b/test/acceptance/connection.js @@ -2,13 +2,10 @@ var Connection = require('../../lib/connection'); var assert = require('assert'); var utils = require('../utils'); -var uid = require('uid'); describe('Connection', function(){ - var topic = uid(); - afterEach(function(done){ - utils.deleteTopic(topic, function(){ - topic = uid(); + beforeEach(function(done){ + utils.deleteTopic('test', function(){ done(); }); }) @@ -31,11 +28,11 @@ describe('Connection', function(){ var sub = new Connection; pub.on('ready', function(){ - pub.publish(topic, 'something'); + pub.publish('test', 'something'); }); sub.on('ready', function(){ - sub.subscribe(topic, 'tailer'); + sub.subscribe('test', 'tailer'); sub.ready(5); }); @@ -52,7 +49,7 @@ describe('Connection', function(){ var conn = new Connection; conn.on('ready', function(){ - conn.subscribe(topic, 'tailer', function(err){ + conn.subscribe('test', 'tailer', function(err){ assert(!err); conn.close(done); }); @@ -68,7 +65,7 @@ describe('Connection', function(){ conn.on('error', function(){}); conn.on('ready', function(){ conn.sock.destroy(); - conn.publish(topic, 'something', function(err){ + conn.publish('test', 'something', function(err){ called++; }); assert.equal(called, 1); @@ -94,7 +91,7 @@ describe('Connection', function(){ conn.connect(); conn.on('ready', function(){ conn.end(); - conn.publish(topic, 'stuff'); + conn.publish('test', 'stuff'); conn.on('error', done); conn.on('end', done); }); diff --git a/test/acceptance/reader.js b/test/acceptance/reader.js index babac84..bb4f18f 100644 --- a/test/acceptance/reader.js +++ b/test/acceptance/reader.js @@ -2,222 +2,203 @@ var utils = require('../utils'); var assert = require('assert'); var nsq = require('../..'); -var uid = require('uid'); describe('Reader', function(){ - var topic = uid(); - afterEach(function(done){ - utils.deleteTopic(topic, function(){ - topic = uid(); + beforeEach(function(done){ + utils.deleteTopic('test', function(){ done(); - }) + }); }) - describe('Reader()', function(){ - describe('with .nsqd addresses', function(){ - it('should subscribe to messages', function(done){ - var pub = nsq.writer(); - var sub = nsq.reader({ - topic: topic, - channel: 'reader', - nsqd: ['0.0.0.0:4150'] - }); - - sub.on('message', function(msg){ - msg.finish(); - done(); - }); - - pub.on('ready', function(){ - pub.publish(topic, 'something', function(err){ - if (err) return done(err); - }); - }); - }) + describe('with .nsqd addresses', function(){ + it('should subscribe to messages', function(done){ + var pub = nsq.writer(); - it('should connect after event handlers are added', function(done){ - var sub = nsq.reader({ - topic: topic, - channel: 'reader', - nsqd: ['0.0.0.0:4150'] - }); + var sub = nsq.reader({ + topic: 'test', + channel: 'reader', + nsqd: ['0.0.0.0:4150'] + }); - sub.connect = function(){ - sub.emit('done'); - }; + pub.on('ready', function(){ + pub.publish('test', 'something'); + }); - sub.on('done', done); + sub.once('message', function(msg){ + msg.finish(); + done(); }); }) + }) - describe('with .nsqlookupd addresses', function(){ - it('should subscribe to messages', function(done){ - var pub = nsq.writer(); - var sub = nsq.reader({ - topic: topic, - channel: 'reader', - nsqlookupd: ['0.0.0.0:4161'], - pollInterval: 100 - }); - sub.on('message', function(msg){ - msg.finish(); - done(); - }); + describe('with .nsqlookupd addresses', function(){ + it('should subscribe to messages', function(done){ + var pub = nsq.writer(); - pub.on('ready', function(){ - pub.publish(topic, 'something', function(err){ - if (err) return done(err); - }); - }); - }) + pub.on('ready', function(){ + pub.publish('test', 'something', function(err){ + if (err) return done(err); - it('should set .timer attribute', function(done){ - var sub = nsq.reader({ - topic: topic, - channel: 'reader', - nsqlookupd: ['0.0.0.0:4161'] - }); + var sub = nsq.reader({ + topic: 'test', + channel: 'reader', + nsqlookupd: ['0.0.0.0:4161'] + }); - setImmediate(function(){ - assert(sub.timer !== null); - done(); + sub.on('message', function(msg){ + msg.finish(); + done(); + }); }); - }) + }); }) - it('should discard messages after the max attempts', function(done){ - var pub = nsq.writer(); + it('should set .timer attribute', function(){ var sub = nsq.reader({ - topic: topic, + topic: 'test', channel: 'reader', - nsqd: ['0.0.0.0:4150'], - maxAttempts: 5, + nsqlookupd: ['0.0.0.0:4161'] }); - sub.once('discard', function(msg){ - sub.removeAllListeners('message'); - done(); - }); - - sub.on('message', function(msg){ - msg.requeue(0); - }); - - pub.on('ready', function(){ - pub.publish(topic, 'something'); - }); + assert(sub.timer !== null); }) }) - describe('Reader#close()', function(){ - var topic = uid(); - beforeEach(function(done){ - var oldTopic = topic; - var newTopic = uid(); - topic = newTopic; - utils.deleteTopic(oldTopic, function(){ - utils.createTopic(newTopic, done); - }) - }) - - it('should wait for pending messages and emit "close"', function(done){ - var pub = nsq.writer(); - var recv = 0; + it('should discard messages after the max attempts', function(done){ + var pub = nsq.writer(); - var sub = nsq.reader({ - topic: topic, - channel: 'reader', - nsqd: ['0.0.0.0:4150'], - maxInFlight: 10 - }); + var sub = nsq.reader({ + topic: 'test', + channel: 'reader', + nsqd: ['0.0.0.0:4150'], + maxAttempts: 5 + }); - var n = 0; - pub.on('ready', function(){ - var count = 30; - while (count--) pub.publish(topic, { n: n++ }); - }); + pub.on('ready', function(){ + pub.publish('test', 'something'); + }); - sub.on('close', function(){ - assert(recv < 30, 'received too many messages'); - done(); - }); + sub.on('message', function(msg){ + msg.requeue(); + }); - sub.on('message', function(msg){ - setTimeout(function(){ - if (recv++ == 10) sub.close(); - msg.finish(); - }, 50); - }); - }) + sub.once('discard', function(msg){ + sub.removeAllListeners('message'); + done(); + }); }) +}) - describe('Reader#close(fn)', function(){ - it('should wait for pending messages and invoke the callback', function(done){ - var pub = nsq.writer(); - var recv = 0; - - var sub = nsq.reader({ - topic: topic, - channel: 'reader', - nsqd: ['0.0.0.0:4150'], - maxInFlight: 10 - }); +describe('Reader#close()', function(){ + beforeEach(function(done){ + utils.deleteTopic('test', function(){ + done(); + }); + }) - var n = 0; - pub.on('ready', function(){ - var count = 30; - while (count--) pub.publish(topic, { n: n++ }); - }); + it('should wait for pending messages and emit "close"', function(done){ + var pub = nsq.writer(); + var recv = 0; + + var sub = nsq.reader({ + topic: 'test', + channel: 'reader', + nsqd: ['0.0.0.0:4150'], + maxInFlight: 10 + }); + + var n = 0; + pub.on('ready', function(){ + var count = 30; + while (count--) pub.publish('test', { n: n++ }); + }); + + sub.on('close', function(){ + assert(recv < 30, 'received too many messages'); + done(); + }); + + sub.on('message', function(msg){ + setTimeout(function(){ + if (recv++ == 10) sub.close(); + msg.finish(); + }, 50); + }); + }) +}) - sub.on('message', function(msg){ - setTimeout(function(){ - if (recv++ == 10) { - sub.close(function(){ - assert(recv < 30, 'received too many messages'); - done(); - }); - } - msg.finish(); - }, 50); - }); - }) +describe('Reader#close(fn)', function(){ + beforeEach(function(done){ + utils.deleteTopic('test', function(){ + done(); + }); + }) - it('should close if there are no in-flight messages', function(done){ - var pub = nsq.writer(); - var sub = nsq.reader({ - topic: topic, - channel: 'reader', - nsqd: ['0.0.0.0:4150'], - maxInFlight: 10 - }); + it('should wait for pending messages and invoke the callback', function(done){ + var pub = nsq.writer(); + var recv = 0; + + var sub = nsq.reader({ + topic: 'test', + channel: 'reader', + nsqd: ['0.0.0.0:4150'], + maxInFlight: 10 + }); + + var n = 0; + pub.on('ready', function(){ + var count = 30; + while (count--) pub.publish('test', { n: n++ }); + }); + + sub.on('message', function(msg){ + setTimeout(function(){ + if (recv++ == 10) { + sub.close(function(){ + assert(recv < 30, 'received too many messages'); + done(); + }); + } + msg.finish(); + }, 50); + }); + }) - sub.on('ready', function(){ - // give it some time to SUB... lame - // add subscribe event? - setTimeout(function(){ - sub.close(done); - }, 100); - }); - }) + it('should close if there are no in-flight messages', function(done){ + var pub = nsq.writer(); + + var sub = nsq.reader({ + topic: 'empty', + channel: 'reader', + nsqd: ['0.0.0.0:4150'], + maxInFlight: 10 + }); + + sub.on('ready', function(){ + // give it some time to SUB... lame + // add subscribe event? + setTimeout(function(){ + sub.close(done); + }, 100); + }); }) +}) - describe('Reader#close(fn)', function(){ - it('should stop polling nsqlookupd if reader had been closed', function(done){ - var sub = nsq.reader({ - topic: topic, - channel: 'reader', - nsqlookupd: ['0.0.0.0:4161'], - pollInterval: 100 - }); +describe('Reader#close(fn)', function(){ + it('should stop polling nsqlookupd if reader had been closed', function(done){ + var sub = nsq.reader({ + topic: 'test', + channel: 'reader', + nsqlookupd: ['0.0.0.0:4161'], + pollInterval: 100 + }); - setImmediate(function(){ - sub.close(); - sub.lookup = function(fn){ - done(new Error('setInterval() is still running')); - }; - }); + sub.close(); - setTimeout(done, 500); - }) + setTimeout(done, 500); + + sub.lookup = function(fn){ + done(new Error('setInterval() is still running')); + }; }) -}); \ No newline at end of file +}) diff --git a/test/acceptance/writer.js b/test/acceptance/writer.js index 034fcea..c5ac303 100644 --- a/test/acceptance/writer.js +++ b/test/acceptance/writer.js @@ -3,13 +3,10 @@ var Connection = require('../../lib/connection'); var utils = require('../utils'); var assert = require('assert'); var nsq = require('../..'); -var uid = require('uid'); describe('Writer#publish()', function(){ - var topic = uid(); - afterEach(function(done){ - utils.deleteTopic(topic, function(){ - topic = uid(); + beforeEach(function(done){ + utils.deleteTopic('test', function(){ done(); }); }) @@ -19,11 +16,11 @@ describe('Writer#publish()', function(){ var sub = new Connection; pub.on('ready', function(){ - pub.publish(topic, 'something'); + pub.publish('test', 'something'); }); sub.on('ready', function(){ - sub.subscribe(topic, 'tailer'); + sub.subscribe('test', 'tailer'); sub.ready(5); }); @@ -40,7 +37,7 @@ describe('Writer#publish()', function(){ pub.on('error', function(){}); - pub.publish(topic, 'something', function(err){ + pub.publish('test', 'something', function(err){ err.message.should.equal('no nsqd nodes connected'); done(); }); @@ -71,9 +68,9 @@ describe('Writer#publish()', function(){ } pub.on('ready', function(){ - pub.publish(topic, new Buffer(1024), next); - pub.publish(topic, new Buffer(1024), next); - pub.publish(topic, new Buffer(1024), next); + pub.publish('test', new Buffer(1024), next); + pub.publish('test', new Buffer(1024), next); + pub.publish('test', new Buffer(1024), next); pub.close(function(){ assert(n === 3); done(); @@ -81,7 +78,7 @@ describe('Writer#publish()', function(){ }); sub.on('ready', function(){ - sub.subscribe(topic, 'tailer'); + sub.subscribe('test', 'tailer'); }); sub.on('message', function(msg){ @@ -99,11 +96,11 @@ describe('Writer#publish()', function(){ var n = 0; pub.on('ready', function(){ - pub.publish(topic, ['foo', 'bar', 'baz']); + pub.publish('test', ['foo', 'bar', 'baz']); }); sub.on('ready', function(){ - sub.subscribe(topic, 'something'); + sub.subscribe('test', 'something'); sub.ready(5); }); @@ -127,11 +124,11 @@ describe('Writer#publish()', function(){ var sub = new Connection; pub.on('ready', function(){ - pub.publish(topic, Buffer('foobar')); + pub.publish('test', Buffer('foobar')); }); sub.on('ready', function(){ - sub.subscribe(topic, 'something'); + sub.subscribe('test', 'something'); sub.ready(5); }); diff --git a/test/utils.js b/test/utils.js index de181b3..e496bcc 100644 --- a/test/utils.js +++ b/test/utils.js @@ -13,30 +13,15 @@ var request = require('superagent'); */ exports.deleteTopic = function(topic, fn){ - req(topic, 'delete', fn) -}; - -/** - * Create `topic`. - * - * @param {String} topic - * @param {Function} fn - */ - -exports.createTopic = function(topic, fn){ - req(topic, 'create', fn); -}; - -function req(topic, cmd, fn){ request - .post('http://127.0.0.1:4151/topic/' + cmd) + .get('http://127.0.0.1:4151/delete_topic') .query({ topic: topic }) .end(function(err, res){ if (err) return fn(err); if (res.error) return fn(res.error); fn(); - }); -} + }); +}; exports.framerData = [ '000000b0000000007b226d61785f7264795f636f756e74223a323530302c2276657273696f6e223a22302e322e3234222c226d61785f6d73675f74696d656f7574223a3930303030302c226d73675f74696d656f7574223a36303030302c22746c735f7631223a66616c73652c226465666c617465223a66616c73652c226465666c6174655f6c6576656c223a302c226d61785f6465666c6174655f6c6576656c223a362c22736e61707079223a66616c73657d', @@ -44,4 +29,3 @@ exports.framerData = [ '0000002f00000002135a2ad167d76e45000130363236323534303166363566303038736f6d65206d6573736167652068657265', '062625401f65f008' ]; - From da772ceedd78bd2a8a299f79bbc714fead92bcdd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Barbosa?= Date: Thu, 20 Oct 2016 15:10:04 +0100 Subject: [PATCH 2/5] Update circle packages --- circle.yml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/circle.yml b/circle.yml index 5b05ffa..85e490f 100644 --- a/circle.yml +++ b/circle.yml @@ -4,14 +4,14 @@ machine: dependencies: pre: - - wget https://github.com/bitly/nsq/releases/download/v0.2.30/nsq-0.2.30.linux-amd64.go1.3.tar.gz - - tar xvzf nsq-0.2.30.linux-amd64.go1.3.tar.gz - - cp nsq-0.2.30.linux-amd64.go1.3/bin/* . + - wget https://github.com/nsqio/nsq/releases/download/v0.3.8/nsq-0.3.8.linux-amd64.go1.6.2.tar.gz + - tar xvzf nsq-0.3.8.linux-amd64.go1.6.2.tar.gz + - cp nsq-0.3.8.linux-amd64.go1.6.2/bin/* . - ./nsqlookupd: background: true - ./nsqd --lookupd-tcp-address=127.0.0.1:4160: background: true cache_directories: - - nsq-0.2.30.linux-amd64.go1.3 + - nsq-0.3.8.linux-amd64.go1.6.2 override: - npm install From a1afedf92beb82371aaf6e8fd6dd8dbc2f266c64 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Barbosa?= Date: Thu, 20 Oct 2016 15:24:28 +0100 Subject: [PATCH 3/5] Update dependencies --- package.json | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/package.json b/package.json index 912638f..53f68c0 100644 --- a/package.json +++ b/package.json @@ -12,20 +12,20 @@ "queue" ], "dependencies": { - "backo": "~1.0.1", + "backo": "~1.1.0", "bignum": "~0.12.5", - "debug": "~0.7.4", - "ms": "~0.6.2", - "node-int64": "~0.3.0", - "nsq-lookup": "~1.0.0", + "debug": "~2.2.0", + "ms": "~0.7.1", + "node-int64": "~0.4.0", + "nsq-lookup": "~1.0.2", "set-component": "~1.0.0" }, "devDependencies": { + "bytes": "~2.4.0", + "jstrace": "~0.3.0", "mocha": "*", "should": "*", - "superagent": "~0.17.0", - "jstrace": "~0.1.0", - "bytes": "~0.3.0" + "superagent": "~2.3.0" }, "license": "MIT" } From 4ecfa7911eefc4d24a01db317d0f728f3084773e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Barbosa?= Date: Mon, 24 Oct 2016 11:49:33 +0100 Subject: [PATCH 4/5] Wait for reader clients to close in test suite --- Readme.md | 2 +- lib/reader.js | 5 +++- package.json | 1 + test/acceptance/connection.js | 11 +++----- test/acceptance/hooks.js | 17 +++++++++++++ test/acceptance/reader.js | 47 +++++++++++++---------------------- test/acceptance/writer.js | 17 +++++-------- test/utils.js | 17 ++++++++++--- 8 files changed, 62 insertions(+), 55 deletions(-) create mode 100644 test/acceptance/hooks.js diff --git a/Readme.md b/Readme.md index 1c83236..c8a7c75 100644 --- a/Readme.md +++ b/Readme.md @@ -28,7 +28,7 @@ $ npm install nsq.js except fo the framer: ``` -$ DEBUG=nsq*,-nsq:framer node test +$ DEBUG=nsq*,-nsq:framer npm test nsq:reader connect nsqd 0.0.0.0:4150 events/ingestion [5] +0ms nsq:connection connect: 0.0.0.0:4150 V2 +0ms diff --git a/lib/reader.js b/lib/reader.js index 80b831f..adaf9bb 100644 --- a/lib/reader.js +++ b/lib/reader.js @@ -72,7 +72,10 @@ function Reader(opts) { // mixins close(this); - this.connect(); + var self = this; + setImmediate(function(){ + self.connect(); + }); } /** diff --git a/package.json b/package.json index 53f68c0..92d7ee5 100644 --- a/package.json +++ b/package.json @@ -23,6 +23,7 @@ "devDependencies": { "bytes": "~2.4.0", "jstrace": "~0.3.0", + "lodash.get": "^4.4.2", "mocha": "*", "should": "*", "superagent": "~2.3.0" diff --git a/test/acceptance/connection.js b/test/acceptance/connection.js index 96ca565..d7daf59 100644 --- a/test/acceptance/connection.js +++ b/test/acceptance/connection.js @@ -1,15 +1,10 @@ +require('./hooks'); + var Connection = require('../../lib/connection'); var assert = require('assert'); -var utils = require('../utils'); describe('Connection', function(){ - beforeEach(function(done){ - utils.deleteTopic('test', function(){ - done(); - }); - }) - it('should identify on connect', function(done){ var conn = new Connection; @@ -38,7 +33,7 @@ describe('Connection', function(){ sub.on('message', function(msg){ msg.finish(); - done(); + sub.close(done); }); pub.connect(); diff --git a/test/acceptance/hooks.js b/test/acceptance/hooks.js new file mode 100644 index 0000000..6eed516 --- /dev/null +++ b/test/acceptance/hooks.js @@ -0,0 +1,17 @@ + +var assert = require('assert'); +var get = require('lodash.get'); +var utils = require('../utils'); + +beforeEach(function(done){ + utils.deleteTopic('test', done); +}) + +afterEach(function(done){ + utils.stats('test', 'reader', function(err, stats){ + if (err) return done(err); + var state = get(stats.body.data, 'topics[0].channels[0].clients[0].state'); + assert(state != 3, 'client in subscribed state'); + done(); + }) +}); diff --git a/test/acceptance/reader.js b/test/acceptance/reader.js index bb4f18f..ab0372d 100644 --- a/test/acceptance/reader.js +++ b/test/acceptance/reader.js @@ -1,15 +1,10 @@ -var utils = require('../utils'); +require('./hooks'); + var assert = require('assert'); var nsq = require('../..'); describe('Reader', function(){ - beforeEach(function(done){ - utils.deleteTopic('test', function(){ - done(); - }); - }) - describe('with .nsqd addresses', function(){ it('should subscribe to messages', function(done){ var pub = nsq.writer(); @@ -26,7 +21,7 @@ describe('Reader', function(){ sub.once('message', function(msg){ msg.finish(); - done(); + sub.close(done); }); }) }) @@ -45,9 +40,9 @@ describe('Reader', function(){ nsqlookupd: ['0.0.0.0:4161'] }); - sub.on('message', function(msg){ + sub.once('message', function(msg){ msg.finish(); - done(); + sub.close(done); }); }); }); @@ -60,7 +55,10 @@ describe('Reader', function(){ nsqlookupd: ['0.0.0.0:4161'] }); - assert(sub.timer !== null); + setImmediate(function(){ + assert(sub.timer !== null); + sub.close(done); + }); }) }) @@ -71,7 +69,7 @@ describe('Reader', function(){ topic: 'test', channel: 'reader', nsqd: ['0.0.0.0:4150'], - maxAttempts: 5 + maxAttempts: 1 }); pub.on('ready', function(){ @@ -84,18 +82,12 @@ describe('Reader', function(){ sub.once('discard', function(msg){ sub.removeAllListeners('message'); - done(); + sub.close(done); }); }) }) describe('Reader#close()', function(){ - beforeEach(function(done){ - utils.deleteTopic('test', function(){ - done(); - }); - }) - it('should wait for pending messages and emit "close"', function(done){ var pub = nsq.writer(); var recv = 0; @@ -128,12 +120,6 @@ describe('Reader#close()', function(){ }) describe('Reader#close(fn)', function(){ - beforeEach(function(done){ - utils.deleteTopic('test', function(){ - done(); - }); - }) - it('should wait for pending messages and invoke the callback', function(done){ var pub = nsq.writer(); var recv = 0; @@ -193,12 +179,13 @@ describe('Reader#close(fn)', function(){ pollInterval: 100 }); - sub.close(); + setImmediate(function(){ + sub.close(); + sub.lookup = function(fn){ + done(new Error('setInterval() is still running')); + }; + }); setTimeout(done, 500); - - sub.lookup = function(fn){ - done(new Error('setInterval() is still running')); - }; }) }) diff --git a/test/acceptance/writer.js b/test/acceptance/writer.js index c5ac303..a82a408 100644 --- a/test/acceptance/writer.js +++ b/test/acceptance/writer.js @@ -1,16 +1,11 @@ +require('./hooks'); + var Connection = require('../../lib/connection'); -var utils = require('../utils'); var assert = require('assert'); var nsq = require('../..'); describe('Writer#publish()', function(){ - beforeEach(function(done){ - utils.deleteTopic('test', function(){ - done(); - }); - }) - it('should publish messages', function(done){ var pub = nsq.writer(); var sub = new Connection; @@ -26,7 +21,7 @@ describe('Writer#publish()', function(){ sub.on('message', function(msg){ msg.finish(); - done(); + sub.close(done); }); sub.connect(); @@ -73,7 +68,7 @@ describe('Writer#publish()', function(){ pub.publish('test', new Buffer(1024), next); pub.close(function(){ assert(n === 3); - done(); + sub.close(done); }); }); @@ -110,7 +105,7 @@ describe('Writer#publish()', function(){ if (++n == 3) { msgs.should.eql(['foo', 'bar', 'baz']); - done(); + sub.close(done); } }); @@ -135,7 +130,7 @@ describe('Writer#publish()', function(){ sub.on('message', function(msg){ msg.finish(); msg.body.toString().should.eql('foobar'); - done(); + sub.close(done); }); sub.connect(); diff --git a/test/utils.js b/test/utils.js index e496bcc..6cc42cb 100644 --- a/test/utils.js +++ b/test/utils.js @@ -14,12 +14,21 @@ var request = require('superagent'); exports.deleteTopic = function(topic, fn){ request - .get('http://127.0.0.1:4151/delete_topic') + .post('http://127.0.0.1:4151/topic/delete') .query({ topic: topic }) .end(function(err, res){ - if (err) return fn(err); - if (res.error) return fn(res.error); - fn(); + if (err && err.status == 404) err = null; + fn(err, res); + }); +}; + +exports.stats = function(topic, channel, fn){ + request + .get('http://127.0.0.1:4151/stats') + .query({ format: 'json', topic: topic, channel: channel }) + .end(function(err, res){ + if (err) return fn(err) + fn(null, res); }); }; From 49d452b1661ea9872d604db8217c31ecb94784a3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Barbosa?= Date: Mon, 24 Oct 2016 11:50:10 +0100 Subject: [PATCH 5/5] Add callbacks to Message methods --- Readme.md | 8 ++++---- lib/connection.js | 36 ++++++++++++++++++++++++------------ lib/message.js | 15 +++++++++------ 3 files changed, 37 insertions(+), 22 deletions(-) diff --git a/Readme.md b/Readme.md index c8a7c75..0e62afe 100644 --- a/Readme.md +++ b/Readme.md @@ -151,17 +151,17 @@ Close the writer's connection(s) and fire the optional [fn] when completed. A single message. -### Message#finish() +### Message#finish([fn]) - Mark message as complete. + Mark message as complete.. -### Message#requeue([delay]) +### Message#requeue([delay], [fn]) Re-queue the message immediately, or with the given `delay` in milliseconds, or a string such as "5s", "10m" etc. -### Message#touch() +### Message#touch([fn]) Reset the message's timeout, increasing the length of time before NSQD considers it timed out. diff --git a/lib/connection.js b/lib/connection.js index c46a368..2176be2 100644 --- a/lib/connection.js +++ b/lib/connection.js @@ -341,13 +341,17 @@ Connection.prototype.ready = function(n){ * @api private */ -Connection.prototype.finish = function(id){ +Connection.prototype.finish = function(id, fn){ assertValidMessageId(id); - var fn = this.onerror; var self = this; - if (!this._ready) return fn(new Error('cannot finish, connection not ready')); - this.command('FIN', [id], function(){ + if (!this._ready) { + var err = new Error('cannot finish, connection not ready'); + if (fn) fn(err); + return this.onerror(err); + } + this.command('FIN', [id], function(err){ self.emit('finish', id); + if (fn) fn(err); --self.inFlight; }); }; @@ -361,13 +365,17 @@ Connection.prototype.finish = function(id){ * @api private */ -Connection.prototype.requeue = function(id, timeout){ +Connection.prototype.requeue = function(id, timeout, fn){ assertValidMessageId(id); - var fn = this.onerror; var self = this; - if (!this._ready) return fn(new Error('cannot requeue, connection not ready')); - this.command('REQ', [id, timeout || 0], function(){ + if (!this._ready) { + var err = new Error('cannot requeue, connection not ready'); + if (fn) fn(err); + return this.onerror(err); + } + this.command('REQ', [id, timeout || 0], function(err){ self.emit('requeue', id); + if (fn) fn(err); --self.inFlight; }); }; @@ -379,13 +387,17 @@ Connection.prototype.requeue = function(id, timeout){ * @api private */ -Connection.prototype.touch = function(id){ +Connection.prototype.touch = function(id, fn){ assertValidMessageId(id); - var fn = this.onerror; var self = this; - if (!this._ready) return fn(new Error('cannot touch, connection not ready')); - this.command('TOUCH', [id], function(){ + if (!this._ready) { + var err = new Error('cannot touch, connection not ready'); + if (fn) fn(err); + return this.onerror(err); + } + this.command('TOUCH', [id], function(err){ self.emit('touch', id); + if (fn) fn(err); }); }; diff --git a/lib/message.js b/lib/message.js index 0b5a6e5..1045c80 100644 --- a/lib/message.js +++ b/lib/message.js @@ -41,12 +41,13 @@ function Message(body, conn) { /** * Mark the message as finished. * + * @param {Function} [fn] * @api public */ -Message.prototype.finish = function(){ +Message.prototype.finish = function(fn){ this.responded = true; - this.conn.finish(this.id); + this.conn.finish(this.id, fn); this.trace('message:finish', { msg: this }); }; @@ -77,25 +78,27 @@ Message.prototype.timedout = function(){ * Re-queue the message with optional `delay`. * * @param {Number|String} [delay] + * @param {Function} [fn] * @api public */ -Message.prototype.requeue = function(delay){ +Message.prototype.requeue = function(delay, fn){ if ('string' == typeof delay) delay = ms(delay); this.responded = true; - this.conn.requeue(this.id, delay); + this.conn.requeue(this.id, delay, fn); this.trace('message:requeue', { msg: this }); }; /** * Touch the message. * + * @param {Function} [fn] * @api public */ -Message.prototype.touch = function(){ +Message.prototype.touch = function(fn){ this.lastTouch = Date.now(); - this.conn.touch(this.id); + this.conn.touch(this.id, fn); this.trace('message:touch', { msg: this }); };