segmentio/nsq.js

NSQ client for nodejs

nsq.js's Issues

don't emit finish till server recieves FIN

I think in some cases, like when gracefully closing a process, since we emit finish at the same time as sending FIN and we may be exiting the process before the FIN actually got sent over the wire.

Doesn't sound like there is a response back from the FIN though, so not sure how we can easily know the writes got flushed. Perhaps setImmediate?

https://github.com/segmentio/nsq.js/blob/master/lib/connection.js#L325-L326

CC: @calvinfo @visionmedia

add connection instance to related events

to log the offending remote

reconnect event

Priority / QOS

Is there any sense of priority queues ? Not sure from look at code of NSQ...

why close does not exit?

var nsq = require('nsq.js');

var writer = nsq.writer('10.10.133.80:4150');
writer.publish('urlitems', "www.sina.com.cn");
console.log("ok.");
writer.close();

when i run it, the command do not exit, why?

add ms() sugar for .requeue()

.requeue('5m') etc

errors for touching msgs in flight

How does one catch errors for calls to touch on messages that are no longer in-flight?

tls support

when -tls-required

methods for other queue operations such as emptying

mainly for tests

add msg.inspect()

[object Object] isn't very useful ;p

queue writes

eventually give up on reconnects

just bloats the logs

move into separate repos

framer etc

Run nsq.js on React.js

Shitty question,

Can I run this nsq client library with react.js? I use react with ES6 standard.
I already try but no luck.

Thanks.

reinvention of the wheel

Hey TJ -

You may have already seen https://github.com/brianc/node-nsqueue but wanted to point it out to you incase you didn't. It's actually written in JS, extremely well tested & been used by us heavily in production for about 50m messages so far. It's intended to be low level, handling none of the reconnection, back-off, dropping, or anything else logic. It's purpose is to facilitate writing higher level clients (with nsqlookup, reconnection, etc) on top. It contains a component similar to your lib/connection.js that basically maps 1-1 with the nsq binary protocol. It supports MPUB with optional callbacks for all the operations nsq will actually give you an ack for. The only thing I know which is currently missing is compression. Our messages are tiny so I didn't have a need for that yet.

I originally was going to use the coffee-script library, but had an issue with messages going "missing" and it was a bit easier for me to write a new client than to debug that one for a few reasons -- and I wanted something more 'low level.' I'm not looking to win any sort of award or popularity, I simply wrote it because I needed something working in production as soon as possible. I figure you were probably in a similar situation. If you see any way we could merge my work into this repo or platform this one on top of node-nsqueue - I'd be happy to collapse 2 modules into one / save the re-work rather than have yet another "900 modules that do the same thing, all of them missing 1 or 2 features." At the very least maybe you can find some parts reusable & save yourself some time.

Please lemme know whatcha think.

Cheers,
b

add jstrace probes

npm install nsq.js error

$ npm install nsq.js --save
npm WARN package.json [email protected] No description
npm WARN package.json [email protected] No repository field.
npm WARN package.json [email protected] No README data

> [email protected] install E:\zan\coding\panli\curlnode\node_modules\nsq.js\node_modules\bignum
> node-gyp configure build


E:\zan\coding\panli\curlnode\node_modules\nsq.js\node_modules\bignum>if not defined npm_config_node_gyp (node "D:\nodejs\node_modules\npm\bin\node-gyp-bin\\..\..\node_modules\node-gyp\bin\node-gyp.js" configure build )  else (node  configure build )
▒ڴ˽▒▒▒▒▒▒▒▒һ▒▒▒▒▒▒һ▒▒▒▒Ŀ▒▒▒▒Ҫ▒▒▒ò▒▒▒▒▒▒ɣ▒▒▒▒▒ӡ▒/m▒▒▒▒▒ء▒
  bignum.cc
..\bignum.cc(413): warning C4244: ▒▒=▒▒: ▒ӡ▒double▒▒ת▒▒▒▒▒▒uint64_t▒▒▒▒▒▒▒ܶ▒ʧ▒▒▒▒ [E:\zan\coding\panli\curlnode\node_modules\nsq.js\node_modules\bignum\build\bignum.vcxproj]
  win_delay_load_hook.c
LINK : fatal error LNK1181: ▒޷▒▒▒▒▒▒▒▒ļ▒▒▒C:\OpenSSL-Win64\lib\libeay32.lib▒▒ [E:\zan\coding\panli\curlnode\node_modules\nsq.js\node_modules\bignum\build\bignum.vcxproj]
gyp ERR! build error
gyp ERR! stack Error: `C:\Program Files (x86)\MSBuild\12.0\bin\msbuild.exe` failed with exit code: 1
gyp ERR! stack     at ChildProcess.onExit (D:\nodejs\node_modules\npm\node_modules\node-gyp\lib\build.js:270:23)
gyp ERR! stack     at emitTwo (events.js:87:13)
gyp ERR! stack     at ChildProcess.emit (events.js:172:7)
gyp ERR! stack     at Process.ChildProcess._handle.onexit (internal/child_process.js:200:12)
gyp ERR! System Windows_NT 6.1.7601
gyp ERR! command "D:\\nodejs\\node.exe" "D:\\nodejs\\node_modules\\npm\\node_modules\\node-gyp\\bin\\node-gyp.js" "configure" "build"
gyp ERR! cwd E:\zan\coding\panli\curlnode\node_modules\nsq.js\node_modules\bignum
gyp ERR! node -v v4.2.1
gyp ERR! node-gyp -v v3.0.3
gyp ERR! not ok
npm ERR! Windows_NT 6.1.7601
npm ERR! argv "D:\\nodejs\\node.exe" "D:\\nodejs\\node_modules\\npm\\bin\\npm-cli.js" "install" "nsq.js" "--save"
npm ERR! node v4.2.1
npm ERR! npm  v2.14.7
npm ERR! code ELIFECYCLE

npm ERR! [email protected] install: `node-gyp configure build`
npm ERR! Exit status 1
npm ERR!
npm ERR! Failed at the [email protected] install script 'node-gyp configure build'.
npm ERR! This is most likely a problem with the bignum package,
npm ERR! not with npm itself.
npm ERR! Tell the author that this fails on your system:
npm ERR!     node-gyp configure build
npm ERR! You can get their info via:
npm ERR!     npm owner ls bignum
npm ERR! There is likely additional logging output above.

npm ERR! Please include the following file with any support request:
npm ERR!     E:\zan\coding\panli\curlnode\npm-debug.log

writes and closes

Try to play nicer.

Error: invalid state
    at Connection.onframe (/usr/local/lib/node_modules/integration-worker/node_modules/nsq.js/lib/connection.js:226:17)
    at Framer.emit (events.js:95:17)
    at Framer.write (/usr/local/lib/node_modules/integration-worker/node_modules/nsq.js/lib/framer.js:56:10)
    at Socket.<anonymous> (/usr/local/lib/node_modules/integration-worker/node_modules/nsq.js/lib/connection.js:90:49)
    at Socket.emit (events.js:95:17)
    at Socket.<anonymous> (_stream_readable.js:764:14)
    at Socket.emit (events.js:92:17)
    at emitReadable_ (_stream_readable.js:426:10)
    at emitReadable (_stream_readable.js:422:5)
    at readableAddChunk (_stream_readable.js:165:9)

Error: write after end
    at writeAfterEnd (_stream_writable.js:132:12)
    at Socket.Writable.write (_stream_writable.js:180:5)
    at Socket.write (net.js:615:40)
    at Connection.command (/usr/local/lib/node_modules/integration-worker/node_modules/nsq.js/lib/connection.js:183:8)
    at Connection.finish (/usr/local/lib/node_modules/integration-worker/node_modules/nsq.js/lib/connection.js:349:8)
    at Message.finish (/usr/local/lib/node_modules/integration-worker/node_modules/nsq.js/lib/message.js:49:13)
    at /usr/local/lib/node_modules/integration-worker/lib/worker.js:178:34
    at Worker.process (/usr/local/lib/node_modules/integration-worker/lib/worker.js:88:14)
    at Worker.onmessage (/usr/local/lib/node_modules/integration-worker/lib/worker.js:175:18)
    at Reader.emit (events.js:106:17)

lookup backoff / jitter

refactor

getting a little messayyy

support Ready(count)

postpone reconnection attempt until lookup is performed

otherwise we can't remove topics, the consumer just immediately reconnects re-creating the topic

nsqio/nsq#343 (comment)

Reconnection to dead nsqd nodes

It's more a curiosity about how this happens, shouldn't the nsqd nodes reconnect once network is okay, since they heartbeat every X seconds the nsqlookupd hosts?
I would like to know if this is related to nsqio/nsq#851

fix Infinity maxAttempts default in Reader

in connect we default to Infinity, but default to 5 in reader haha.. my bad. Infinity is the sane one IMO

MPUB support

proper nsqd termination

from what I've read nsqd does not send any "I'm doing down" message so we can't really disambiguate between graceful and ungraceful exits

increase throughput

more than enough for now but could use improvement, haven't profiled anything yet

tests are wonky

Sometimes the pass and sometimes the timeout/fail. It's defiantly an issue with just the tests as our apps run just fine but it would be nice to fix them :-)

tests ;D

little tricky to test without launching a ton of real instances, but still lots that can use unit tests

Deferred publish

Hi,
Any plan to add the deferred publish DPUB available from nsq 0.3.6 ?
Thank you

identify options

queue writes until first connection is established

mostly for testing etc, annoying to add a ton of callbacks for initialization

Writer does not work with using nsqlookupd

Writer.lookup does not pass opts to nsq-lookup. However, nsq.lookup throws an error if opts.topic is not set. Ergo, Writer never connects to publish.

remove `msg.finish` callback?

got confused about why msg.finish() wasn't calling back, until I saw nsqio/nsq#320

any thoughts on removing it from the msg.finish signature? seems like you could still listen on error and handle that however you'd like

multiple msg.finish() as noop

Calling msg.finish() on a finished message should be a noop, otherwise what would it be good for?

better starvation handling

had a case where one producer had no messages

add callback to .close()

That doesn't call back till everything in-flight has drained.

don't reconnect if identify fails

If you identify incorrectly, the nsqd server drops your connection. I think rather than reconnecting with the same bad values over and over, we should just bubble the error up.

14/09/05 10:22:05 CLIENT(192.168.1.3:59946): desired protocol magic '  V2'
2014/09/05 10:22:05 [192.168.1.3:59946] IDENTIFY: {ShortId:Garretts-MacBook-Pro LongId:Garretts-MacBook-Pro.local ClientID:example Hostname: HeartbeatInterval:0 OutputBufferSize:0 OutputBufferTimeout:0 FeatureNegotiation:true TLSv1:false Deflate:false DeflateLevel:0 Snappy:false SampleRate:0 UserAgent:nsq.js/0.11.3 MsgTimeout:1800000}
2014/09/05 10:22:05 ERROR: [192.168.1.3:59946] - E_BAD_BODY IDENTIFY msg timeout (1800000) is invalid - msg timeout (1800000) is invalid
2014/09/05 10:22:05 PROTOCOL(V2): [192.168.1.3:59946] exiting ioloop
2014/09/05 10:22:05 PROTOCOL(V2): [192.168.1.3:59946] exiting messagePump
2014/09/05 10:22:05 TCP: new client(192.168.1.3:59947)
2014/09/05 10:22:05 CLIENT(192.168.1.3:59947): desired protocol magic '  V2'
2014/09/05 10:22:05 [192.168.1.3:59947] IDENTIFY: {ShortId:Garretts-MacBook-Pro LongId:Garretts-MacBook-Pro.local ClientID:example Hostname: HeartbeatInterval:0 OutputBufferSize:0 OutputBufferTimeout:0 FeatureNegotiation:true TLSv1:false Deflate:false DeflateLevel:0 Snappy:false SampleRate:0 UserAgent:nsq.js/0.11.3 MsgTimeout:1800000}
2014/09/05 10:22:05 ERROR: [192.168.1.3:59947] - E_BAD_BODY IDENTIFY msg timeout (1800000) is invalid - msg timeout (1800000) is invalid
2014/09/05 10:22:05 PROTOCOL(V2): [192.168.1.3:59947] exiting ioloop
2014/09/05 10:22:05 PROTOCOL(V2): [192.168.1.3:59947] exiting messagePump
2014/09/05 10:22:06 TCP: new client(192.168.1.3:59948)
2014/09/05 10:22:06 CLIENT(192.168.1.3:59948): desired protocol magic '  V2'
2014/09/05 10:22:06 [192.168.1.3:59948] IDENTIFY: {ShortId:Garretts-MacBook-Pro LongId:Garretts-MacBook-Pro.local ClientID:example Hostname: HeartbeatInterval:0 OutputBufferSize:0 OutputBufferTimeout:0 FeatureNegotiation:true TLSv1:false Deflate:false DeflateLevel:0 Snappy:false SampleRate:0 UserAgent:nsq.js/0.11.3 MsgTimeout:1800000}
2014/09/05 10:22:06 ERROR: [192.168.1.3:59948] - E_BAD_BODY IDENTIFY msg timeout (1800000) is invalid - msg timeout (1800000) is invalid
2014/09/05 10:22:06 PROTOCOL(V2): [192.168.1.3:59948] exiting ioloop
2014/09/05 10:22:06 PROTOCOL(V2): [192.168.1.3:59948] exiting messagePump
2014/09/05 10:22:06 TCP: new client(192.168.1.3:59949)
2014/09/05 10:22:06 CLIENT(192.168.1.3:59949): desired protocol magic '  V2'
2014/09/05 10:22:06 [192.168.1.3:59949] IDENTIFY: {ShortId:Garretts-MacBook-Pro LongId:Garretts-MacBook-Pro.local ClientID:example Hostname: HeartbeatInterval:0 OutputBufferSize:0 OutputBufferTimeout:0 FeatureNegotiation:true TLSv1:false Deflate:false DeflateLevel:0 Snappy:false SampleRate:0 UserAgent:nsq.js/0.11.3 MsgTimeout:1800000}
2014/09/05 10:22:06 ERROR: [192.168.1.3:59949] - E_BAD_BODY IDENTIFY msg timeout (1800000) is invalid - msg timeout (1800000) is invalid
2014/09/05 10:22:06 PROTOCOL(V2): [192.168.1.3:59949] exiting ioloop
2014/09/05 10:22:06 PROTOCOL(V2): [192.168.1.3:59949] exiting messagePump
2014/09/05 10:22:06 TCP: new client(192.168.1.3:59950)
2014/09/05 10:22:06 CLIENT(192.168.1.3:59950): desired protocol magic '  V2'
2014/09/05 10:22:06 [192.168.1.3:59950] IDENTIFY: {ShortId:Garretts-MacBook-Pro LongId:Garretts-MacBook-Pro.local ClientID:example Hostname: HeartbeatInterval:0 OutputBufferSize:0 OutputBufferTimeout:0 FeatureNegotiation:true TLSv1:false Deflate:false DeflateLevel:0 Snappy:false SampleRate:0 UserAgent:nsq.js/0.11.3 MsgTimeout:1800000}
2014/09/05 10:22:06 ERROR: [192.168.1.3:59950] - E_BAD_BODY IDENTIFY msg timeout (1800000) is invalid - msg timeout (1800000) is invalid
2014/09/05 10:22:06 PROTOCOL(V2): [192.168.1.3:59950] exiting ioloop
2014/09/05 10:22:06 PROTOCOL(V2): [192.168.1.3:59950] exiting messagePump
2014/09/05 10:22:06 LOOKUPD(0.0.0.0:4160): sending heartbeat
2014/09/05 10:22:06 TCP: new client(192.168.1.3:59952)
2014/09/05 10:22:06 CLIENT(192.168.1.3:59952): desired protocol magic '  V2'
2014/09/05 10:22:06 [192.168.1.3:59952] IDENTIFY: {ShortId:Garretts-MacBook-Pro LongId:Garretts-MacBook-Pro.local ClientID:example Hostname: HeartbeatInterval:0 OutputBufferSize:0 OutputBufferTimeout:0 FeatureNegotiation:true TLSv1:false Deflate:false DeflateLevel:0 Snappy:false SampleRate:0 UserAgent:nsq.js/0.11.3 MsgTimeout:1800000}
2014/09/05 10:22:06 ERROR: [192.168.1.3:59952] - E_BAD_BODY IDENTIFY msg timeout (1800000) is invalid - msg timeout (1800000) is invalid
2014/09/05 10:22:06 PROTOCOL(V2): [192.168.1.3:59952] exiting ioloop
2014/09/05 10:22:06 PROTOCOL(V2): [192.168.1.3:59952] exiting messagePump
2014/09/05 10:22:21 LOOKUPD(0.0.0.0:4160): sending heartbeat
2014/09/05 10:22:36 LOOKUPD(0.0.0.0:4160): sending heartbeat

give up after N reconnection attempts

gets annoying restarting shit and flooding logs

uncatchable error

Although I have an error listener in place, there's an unhandled error event. I'll investigate.

var nsq = require('nsq.js');

var r = nsq.reader({
  topic: 't',
  channel: 'c',
  nsqd: ['n']
});

r.on('error', function(err){
  console.error(err.stack);
});

 ∴  node t
Error: n:4150: getaddrinfo ENOTFOUND n n:4150
    at errnoException (dns.js:25:10)
    at GetAddrInfoReqWrap.onlookup [as oncomplete] (dns.js:75:26)
events.js:141
      throw er; // Unhandled 'error' event
            ^
Error: n:4150: getaddrinfo ENOTFOUND n n:4150
    at errnoException (dns.js:25:10)
    at GetAddrInfoReqWrap.onlookup [as oncomplete] (dns.js:75:26)
 ∴  echo $?
1

Pause/Resume is broken

Pause/Resume is just broken.

Tried node 4.4.5 and 6.4.0 with nsq.js@latest and with the version that the pause example first showed up in the library.

/cc @dweinstein

add lookupd polling for writer connection pool

Memory leak caused by setInterval

It seems hardly to resolve this problem in normal way, but node-sleep + child_process may be an alternative solution.

// timer.js
var sleep = require('sleep').sleep;
while (1) {
  sleep(1);
  process.send('tick');
}

// reader.js
this.timer = child_process.fork('./timer');
this.timer.on('message', function (msg) {
  // poll
});

add tests for writer pooling

add Reader#close() support for set of connections