snappyless-nsqjs
Advanced tools
Without optional snappystream dependencies.
The official NodeJS client for the nsq client protocol. This implementation attempts to be
fully compliant and maintain feature parity with the official Go (go-nsq) and Python (pynsq) clients.
The topic and channel arguments are strings and must be specified. The options argument is optional. Below are the parameters that can be specified in the options object.
maxInFlight: 1 heartbeatInterval: 30 maxBackoffDuration: 128 maxAttempts: 0 maxAttempts > 0, then the message will be finished automatically when the number of attempts has been exhausted.requeueDelay: 90,000 (90secs) nsqdTCPAddresses ['localhost:4150']lookupdHTTPAddresses ['localhost:4161'], ['http://localhost/lookup'], ['http://localhost/path/lookup?extra_param=true']lookupdPollInterval: 60 lookupdPollJitter: 0.3 tls: false tlsVerification: true deflate: false deflateLevel: 6 snappy: false authSecret: nulloutputBufferSize: nulloutputBufferSize >= 64outputBufferTimeout: nulloutputBufferTimeout >= 1. A value of -1 disables timeouts.messageTimeout: nullsampleRate: null1 <= sampleRate <= 99clientId: nullidleTimeout: 0 Reader events are:
Reader.MESSAGE or messageReader.DISCARD or discardReader.ERROR or errorReader.NSQD_CONNECTED or nsqd_connectedReader.NSQD_CLOSED or nsqd_closedReader.MESSAGE and Reader.DISCARD both produce Message objects.
Reader.NSQD_CONNECTED and Reader.NSQD_CLOSED events both provide the host
and port of the nsqd to which the event pertains.
These methods are available on a Reader object:
connect() close() pause() unpause() isPaused() true if paused, false otherwise.The following properties and methods are available on Message objects produced by a Reader instance.
timestamp attempts id hasResponded body json() timeUntilTimeout(hard=false): finish() requeue(delay=null, backoff=true)
The delay is in milliseconds. This is how long nsqd will hold on the message
before attempting it again. The backoff parameter indicates that we should
treat this as an error within this process and we need to backoff to recover.touch() Allows messages to be sent to an nsqd.
Available Writer options:
tls: false tlsVerification: true deflate: false deflateLevel: 6 snappy: false clientId: nullWriter events are:
Writer.READY or readyWriter.CLOSED or closedWriter.ERROR or errorThese methods are available on a Writer object:
connect() close() publish(topic, msgs, [callback]) topic is a string. msgs is either a string, a Buffer, JSON serializable
object, a list of strings / Buffers / JSON serializable objects. callback takes a single error argument.deferPublish(topic, msg, timeMs, [callback]) topic is a string. msg is either a string, a Buffer, JSON serializable object. timeMs is the delay by which the message should be delivered. callback takes a single error argument.Start nsqd and nsqdlookupd
# nsqdLookupd Listens on 4161 for HTTP requests and 4160 for TCP requests
$ nsqlookupd &
$ nsqd --lookupd-tcp-address=127.0.0.1:4160 &
const nsq = require('nsqjs')
const reader = new nsq.Reader('sample_topic', 'test_channel', {
lookupdHTTPAddresses: '127.0.0.1:4161'
})
reader.connect()
reader.on('message', msg => {
console.log('Received message [%s]: %s', msg.id, msg.body.toString())
msg.finish()
})
Publish a message to nsqd to be consumed by the sample client:
$ curl -d "it really tied the room together" http://localhost:4151/pub?topic=sample_topic
This script simulates a message that takes a long time to process or at least longer than the default message timeout. To ensure that the message doesn't timeout while being processed, touch events are sent to keep it alive.
const nsq = require('nsqjs')
const reader = new nsq.Reader('sample_topic', 'test_channel', {
lookupdHTTPAddresses: '127.0.0.1:4161'
})
reader.connect()
reader.on('message', msg => {
console.log('Received message [%s]', msg.id)
const touch = () => {
if (!msg.hasResponded) {
console.log('Touch [%s]', msg.id)
msg.touch()
// Touch the message again a second before the next timeout.
setTimeout(touch, msg.timeUntilTimeout() - 1000)
}
}
const finish = () => {
console.log('Finished message [%s]: %s', msg.id, msg.body.toString())
msg.finish()
}
console.log('Message timeout is %f secs.', msg.timeUntilTimeout() / 1000)
setTimeout(touch, msg.timeUntilTimeout() - 1000)
// Finish the message after 2 timeout periods and 1 second.
setTimeout(finish, msg.timeUntilTimeout() * 2 + 1000)
})
nsqjs uses debug to log debug output.
To see all nsqjs events:
$ DEBUG=nsqjs:* node my_nsqjs_script.js
To see all reader events:
$ DEBUG=nsqjs:reader:* node my_nsqjs_script.js
To see a specific reader's events:
$ DEBUG=nsqjs:reader:<topic>/<channel>:* node my_nsqjs_script.js
Replace
<topic>and<channel>
To see all writer events:
$ DEBUG=nsqjs:writer:* node my_nsqjs_script.js
The writer sends a single message and then a list of messages.
const nsq = require('nsqjs')
const w = new nsq.Writer('127.0.0.1', 4150)
w.connect()
w.on('ready', () => {
w.publish('sample_topic', 'it really tied the room together')
w.deferPublish('sample_topic', ['This message gonna arrive 1 sec later.'], 1000)
w.publish('sample_topic', [
'Uh, excuse me. Mark it zero. Next frame.',
'Smokey, this is not \'Nam. This is bowling. There are rules.'
])
w.publish('sample_topic', 'Wu?', err => {
if (err) { return console.error(err.message) }
console.log('Message sent successfully')
w.close()
})
})
w.on('closed', () => {
console.log('Writer closed')
})
Reader.close to cleanup all intervals to allow node to exit cleanly.src to libmessage.finish is called after backoff event.maxAttempts is now by default 0. [ Breaking Change! ]MESSAGE listener if there's no
DISCARD listener.debug [ Breaking Change! ]close, pause, and unpause to Readerfinish, requeue, etc after nsqd disconnectReaderRdy, ConnectionRdy implementationReader implementationNSQDConnection
Readerconnect() now happens on next tick so that it can be called before event
handlers are registered.Message
TOUCH eventsNSQDConnection implementationwire implementationMessage implementationFAQs
NodeJS client for NSQ without snappy
The npm package snappyless-nsqjs receives a total of 0 weekly downloads. As such, snappyless-nsqjs popularity was classified as not popular.
We found that snappyless-nsqjs demonstrated a not healthy version release cadence and project activity because the last version was released a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Research
Security News
Socket’s threat research team has detected six malicious npm packages typosquatting popular libraries to insert SSH backdoors.
Security News
MITRE's 2024 CWE Top 25 highlights critical software vulnerabilities like XSS, SQL Injection, and CSRF, reflecting shifts due to a refined ranking methodology.
Security News
In this segment of the Risky Business podcast, Feross Aboukhadijeh and Patrick Gray discuss the challenges of tracking malware discovered in open source softare.