DOCSEARCHbyAlgolia
Submit Docs

nsqjs

repository·main·Indexed 20 days ago

https://github.com/dudleycarr/nsqjs

A NodeJS client for the NSQ protocol, version 0.13.0, designed to provide feature parity with official Go and Python clients. It includes a Reader for consuming messages from NSQ topics and channels, a Writer for publishing messages to nsqd instances, and a Message object for managing message lifecycles via methods like finish(), requeue(), and touch(). Supports TLS, Snappy and Deflate compression, and discovery via nsqlookupd.

Tokens
6.6K
Snippets
17
Records
31
Agent score
69%

What's inside nsqjs

  1. Work with Message instances

    main

    When a Reader receives a message, it is provided as a Message object. This object allows you to control the lifecycle of that specific message.

    Common operations on a Message include:

    • finish(): Acknowledges the message as successfully processed.
    • requeue(delay): Puts the message back into the queue with an optional delay (default requeueDelay is 90 seconds).
    • touch(): Resets the message timeout to prevent it from being marked as timed out.

    Messages also support maxAttempts to track how many times a message has been delivered. If maxAttempts is exceeded, the message is automatically finished (as of v0.7.0).

  2. Enable nsqjs debugging

    main

    nsqjs uses the debug library. You can enable logging by setting the DEBUG environment variable before running your script.

    • All nsqjs events: DEBUG=nsqjs:* node script.js
    • All reader events: DEBUG=nsqjs:reader:* node script.js
    • Specific reader events: DEBUG=nsqjs:reader:<topic>/<channel>:* node script.js
    • All writer events: DEBUG=nsqjs:writer:* node script.js

  3. Enable nsqjs debugging

    main
    Debugging support was previously a core dependency but has been moved to a devDependency as of version 0.13.0. You can use the debug module to inspect connection states, protocol logs, and internal events.

  4. Manage NSQ message lifecycle with the Message class

    main

    The Message class represents data received from an NSQ server. It provides methods to respond to nsqd to control the message lifecycle, such as finishing, requeueing, or touching the message to reset its timeout.

    Response Types:

    • Message.FINISH (0): Successfully processed the message.
    • Message.REQUEUE (1): Put the message back into the queue.
    • Message.TOUCH (2): Reset the message timeout on the nsqd side.

    Key Methods:

    • finish(): Tells nsqd the message is done.
    • requeue(delay, backoff): Requeues the message. delay is the number of milliseconds to wait before the message is available again. backoff (default true) determines if the reader should emit a backoff event.
    • touch(): Resets the message timer on the nsqd side. This can be called repeatedly to prevent a message from timing out, up to the max_msg_timeout configured on the server.
    • json(): Parses the message body as JSON. Throws an error if the body is not valid JSON.
    const Message = require('nsqjs').Message; // Assuming standard export

reader.on('message', (msg) => {
  try {
    const data = msg.json();
    // Process data...
    msg.finish();
  } catch (err) {
    // If processing fails, requeue with a delay
    msg.requeue(5000);
  }
});

  5. How Reader handles message discards

    main

    The Reader can automatically handle message exhaustion based on the maxAttempts configuration option.

    If maxAttempts is set to a value greater than 0, and a message's attempts count exceeds this value, the Reader will:

    1. Emit a Reader.DISCARD event (if you have listeners for it).
    2. Automatically call message.finish() to acknowledge the message to NSQ.
    3. Otherwise, it emits the standard Reader.MESSAGE event.

  6. Configure Reader options

    main

    The options object for new Reader(topic, channel, options) supports the following configuration keys:

    KeyTypeDefaultDescription
    maxInFlightnumber1Max messages to process at once (shared across connections).
    heartbeatIntervalnumber30Frequency in seconds for nsqd heartbeats.
    maxBackoffDurationnumber128Max seconds the Reader will backoff for a single event.
    maxAttemptsnumber0Attempts before handing to DISCARD handler. 0 means no limit.
    requeueDelaynumber90000Delay in ms for requeued messages.
    nsqdTCPAddressesstring or string[]nullHost/port pairs for nsqd instances (e.g., ['localhost:4150']).
    lookupdHTTPAddressesstring or string[]nullHost/port or full HTTP/HTTPS URIs for nsqlookupd.
    lookupdPollIntervalnumber60Seconds between querying lookupd.
    lookupdPollJitternumber0.3Jitter applied to lookupd polling.
    lowRdyTimeoutnumber50Timeout in ms for switching connections when maxInFlight is low.
    tlsbooleanfalseEnable TLS support.
    tlsVerificationbooleantrueRequire TLS cert verification (set to false for self-signed).
    deflatebooleanfalseUse zlib Deflate compression.
    deflateLevelnumber6zlib Deflate compression level.
    snappybooleanfalseUse Snappy compression.
    authSecretstringnullAuthentication secret.
    outputBufferSizenumbernullBuffer size in bytes for writing to client. -1 disables.
    outputBufferTimeoutnumbernullFlush timeout in ms. -1 disables.
    messageTimeoutnumbernullServer-side message timeout in ms.
    sampleRatenumbernullPercentage of messages to deliver (1 <= rate <= 99).
    clientIdstringnullIdentifier to disambiguate this client.
    idleTimeoutnumber0Socket timeout in seconds (0 is disabled).

  7. Initialize a Reader with new Reader(topic, channel, options)

    main

    Create a new Reader instance to consume messages from a specific NSQ topic and channel. The topic and channel arguments are required strings. The options object is optional and allows for fine-tuning connection and message handling behavior.

    const nsq = require('nsqjs')

const reader = new nsq.Reader('sample_topic', 'test_channel', {
  lookupdHTTPAddresses: '127.0.0.1:4161'
})

reader.connect()

  8. Manage Reader lifecycle and flow

    main

    The following methods are available on a Reader object:

    • connect(): Connect to specified nsqds or those discovered via lookupd.
    • close(): Disconnect from all nsqds (does not wait for in-flight messages).
    • pause(): Stop message flow (does not affect in-flight messages).
    • unpause(): Resume normal message flow.
    • isPaused(): Returns true if the reader is currently paused.

  9. Interact with Message objects

    main

    When a Reader emits a message event, it provides a Message object. Use these properties and methods to manage the message lifecycle:

    Properties:

    • timestamp: Numeric timestamp from nsqd.
    • attempts: Number of processing attempts made.
    • id: Opaque string ID.
    • hasResponded: Boolean indicating if a response has been sent.
    • body: The message payload as a Buffer.

    Methods:

    • json(): Parses the body as JSON and caches the result.
    • timeUntilTimeout(hard=false): Returns time remaining until timeout. If hard is true, calculates time until the hard timeout (when nsqd requeues regardless of touches).
    • finish(): Marks the message as successfully processed.
    • requeue(delay=null, backoff=true): Requeues the message. delay is in ms. backoff indicates if this should trigger a process backoff.
    • touch(): Extends the soft timeout by the normal timeout amount to prevent premature requeueing during long processing tasks.

  10. Handle Reader events

    main

    A Reader instance emits several events that you can listen to using .on():

    • ready / READY: Reader is ready to receive messages.
    • not_ready / NOT_READY: Reader is not ready.
    • message / MESSAGE: A new Message object is available.
    • discard / DISCARD: A message was discarded.
    • error / ERROR: An error occurred.
    • nsqd_connected / NSQD_CONNECTED: Connected to an nsqd (provides host and port).
    • nsqd_closed / NSQD_CLOSED: Disconnected from an nsqd (provides host and port).

  11. Publish messages with a Writer

    main

    Use the following methods on a Writer instance to send data:

    • publish(topic, msgs, [callback]):
      • topic: string.
      • msgs: A string, Buffer, JSON serializable object, or an array of these types.
      • callback: Function receiving a single error argument.
    • deferPublish(topic, msg, timeMs, [callback]):
      • topic: string.
      • msg: string, Buffer, or JSON serializable object.
      • timeMs: Delay in milliseconds before delivery.
      • callback: Function receiving a single error argument.

  12. Initialize a Writer with new Writer(nsqdHost, nsqdPort, options)

    main

    Create a new Writer instance to publish messages to a specific nsqd instance.

    Options:

    • tls: boolean (default false)
    • tlsVerification: boolean (default true)
    • deflate: boolean (default false)
    • deflateLevel: number (default 6)
    • snappy: boolean (default false)
    • clientId: string (default null)
    const nsq = require('nsqjs')

const w = new nsq.Writer('127.0.0.1', 4150)

w.connect()