thunk-disque's Introduction

thunk-disque

A thunk/promise-based disque client, support all disque features.

Features

Auto meet nodes to cluster
Auto connect available nodes on demand
Auto select connection by jobId

中文教程 http://disquebook.com/

https://github.com/antirez/disque

https://github.com/thunks/thunks

Installation

Node.js:

npm install thunk-disque

Demo

var disque = require('thunk-disque')
var client = disque.createClient([7711, 7712, 7713])
var clientP = disque.createClient([7711, 7712, 7713], {usePromise: true})

// thunk API
client.info()(function (err, info) {
  console.log(err, info)

  return this.addjob('queueA', 'Hello', 0)
})(function (err, res) {
  console.log(err, res)
  // null
  // 'DI81250b3ccbac68e6625e79c8e7c5b286b1dcd2ac05a0SQ'
  return this.show(res)

})(function (err, res) {
  console.log(err, res)
  // null
  // {
  //   id: 'DI81250b3ccbac68e6625e79c8e7c5b286b1dcd2ac05a0SQ',
  //   queue: 'queueA',
  //   state: 'queued',
  //   repl: 3,
  //   ttl: 86400,
  //   ctime: 1430579357544000000,
  //   delay: 0,
  //   retry: 8640,
  //   'nodes-delivered':
  //    [ 'f0e652056250c887ed294a53fa9386ea05abb0be',
  //      '2067c69f914c619ed9f348f5ce6e7532ec26e9a8',
  //      '81250b3c4318f0b6463da3742c7cf7069a46b6f6' ],
  //   'nodes-confirmed': [],
  //   'next-requeue-within': 8639835,
  //   'next-awake-within': 8639335,
  //   body: 'Hello'
  // }
  return this.clientEnd()
});

// promise API
clientP.info()
  .then(function (info) {
    console.log(info)

    return clientP.addjob('queueA', 'Hello', 0)
  })
  .then(function (res) {
    console.log(res)
    // 'DI81250b3ccbac68e6625e79c8e7c5b286b1dcd2ac05a0SQ'
    return clientP.show(res)

  })
  .then(function (res) {
    console.log(res)
    // {
    //   id: 'DI81250b3ccbac68e6625e79c8e7c5b286b1dcd2ac05a0SQ',
    //   queue: 'queueA',
    //   state: 'queued',
    //   repl: 3,
    //   ttl: 86400,
    //   ctime: 1430579357544000000,
    //   delay: 0,
    //   retry: 8640,
    //   'nodes-delivered':
    //    [ 'f0e652056250c887ed294a53fa9386ea05abb0be',
    //      '2067c69f914c619ed9f348f5ce6e7532ec26e9a8',
    //      '81250b3c4318f0b6463da3742c7cf7069a46b6f6' ],
    //   'nodes-confirmed': [],
    //   'next-requeue-within': 8639835,
    //   'next-awake-within': 8639335,
    //   body: 'Hello'
    // }
  })
  .catch(function (err) {
    console.error(err)
  })

API

var disque = require('thunk-disque')

disque.createClient([port], [host], [options])

disque.createClient([addressArray], [options])

port: {Number}, default: 7711;
host: {String}, default: '127.0.0.1';
options: {Object}, default: {};
- handleError: {Boolean}, Optional, Handle client error event. Default: true.
- authPass: {String}, Optional, Default: ''.
- autoMeet: {Boolean}, Optional, Default: false.
- returnBuffers: {Boolean}, Optional, Default: false.
- usePromise: {Boolean|Promise}, Optional, Default: false.
  
  Use default Promise:
```
var disque = require('thunk-disque')
var client = disque.createClient({
  usePromise: true
})
```
  Use bluebird:
```
var disque = require('thunk-disque')
var Bluebird = require('bluebird')
var client = disque.createClient({
  usePromise: Bluebird
})
```
- noDelay: {Boolean}, Optional, Default: true. Disables the Nagle algorithm. By default TCP connections use the Nagle algorithm, they buffer data before sending it off. Setting true for noDelay will immediately fire off data each time socket.write() is called.
- retryMaxDelay: {Number}, Optional, Default: Infinity. By default every time the client tries to connect and fails time before reconnection (delay) almost multiply by 1.2. This delay normally grows infinitely, but setting retryMaxDelay limits delay to maximum value, provided in milliseconds.
- maxAttempts: {Number}, Optional, Default: 10. By default client will try reconnecting until connected. Setting maxAttempts limits total amount of reconnects.

Create a disque client, return the client.

// connect to 127.0.0.1:7711
var client1 = disque.createClient()
var client2 = disque.createClient(7711, '127.0.0.1')

// connect to 127.0.0.1:7711, 127.0.0.1:7712
// and auto meet them into cluster
var client3 = redis.createClient([7711, 7712], {autoMeet: true})
var client4 = redis.createClient(['127.0.0.1:7711', '127.0.0.1:7712'], {autoMeet: true}) // IPv4
var client5 = redis.createClient(['[::1]:7711', '[::1]:7712'], {autoMeet: true}) // IPv6

disque.log([...])

var client = disque.createClient()
client.info()(redis.log)

Events

client.on('close', function () {})

client.on('connect', function () {})

client.on('connection', function (connection) {})

client.on('warn', function (error) {})

client.on('error', function (error) {})

client.on('reconnecting', function (message) {})

client.on('monitor', function (message) {})

Others

client.clientCommands

client.clientEnd()

client.clientUnref()

Disque Commands

client.ackjob

client.addjob

client.auth

client.bgrewriteaof

client.client

client.cluster

client.command

client.config

client.debug

client.deljob

client.dequeue

client.enqueue

client.fastack

client.getjob

client.hello

client.info

client.latency

client.loadjob

client.monitor

client.ping

client.qlen

client.qpeek

client.show

client.shutdown

client.slowlog

client.time

thunk-disque's People

Contributors

Stargazers

Watchers

thunk-disque's Issues

Client is not attempting reconnect

The disque client looks to have reconnect functionality built in. However, I am having trouble getting the client to reconnect to the disque server.

My test scenario is that I first create a disque client and shortly afterwards shutdown the disque server. I expected that when I brought the server back up, the client would eventually reconnect. From my test, the client remains in the ended state and will not reconnect. If I trace through what the client is doing, it looks to be skipping the reconnect attempts because it returns due to ended being true.

Between 2690766...05146ea the ended property is now set to true when the socket's close event is fired due to the call to tryRemove. Any pointers for how to get this working would be very much appreciated.

Crash when disconnected from Disque

Hi zensh!

I've been testing Disque using your client lib and during a simulated disconnect, thunk-disque crashed when trying to recover pending commands. Going to take a look at it and see what's the problem. 👍

Here's the debug output:

[nodemon] starting
  disque create socket, 192.168.99.100:7711 +0ms
  disque add command +15ms
 Command {
  data: <Buffer ...>,
  name: 'hello',
  callback: [Function],
  mayBeJobId: undefined }
  disque socket write, node 192.168.99.100:7711, length 15 +17ms
 <Buffer ...>
  disque socket connected, 192.168.99.100:7711 +0ms
  disque resp receive, node 192.168.99.100:7711 +3ms
 [ 1,
  'ac9a5c99eef25116f4f8b8a88d02d493f4745ff3',
  [ 'ac9a5c99eef25116f4f8b8a88d02d493f4745ff3', '', '7711', '1' ] ] Command {
  data: <Buffer ...>,
  name: 'hello',
  callback: [Function],
  mayBeJobId: undefined }

=====> Killed VM to simulate disconnect <=====
=====> Added job to a queue             <=====

  disque add command +4m
 Command {
  data: <Buffer ...>,
  name: 'addjob',
  callback: [Function: callback],
  mayBeJobId: 'testqueue' }
  disque socket write, node 192.168.99.100:7711, length 187 +1ms
 <Buffer ...>
thunk-disque Error: read EHOSTUNREACH
    at exports._errnoException (util.js:855:11)
    at TCP.onread (net.js:544:26)
  disque rescue pending commands, 192.168.99.100:7711 +55s
  disque socket reconnecting, 192.168.99.100:7711 +4s
  disque destroy socket, 192.168.99.100:7711 +1ms
  disque create socket, 192.168.99.100:7711 +1ms
thunk-disque Error: connect ETIMEDOUT 192.168.99.100:7711
    at Object.exports._errnoException (util.js:855:11)
    at exports._exceptionWithHostPort (util.js:878:20)
    at TCPConnectWrap.afterConnect [as oncomplete] (net.js:1063:14)
  disque rescue pending commands, 192.168.99.100:7711 +26s
/node_modules/thunk-disque/lib/connection.js:61
  while (this.pendingQueue.length) {
                          ^

TypeError: Cannot read property 'length' of null
    at Connection.rescuePending (/node_modules/thunk-disque/lib/connection.js:61:27)
    at Connection.reconnecting (/node_modules/thunk-disque/lib/connection.js:145:10)
    at Socket.<anonymous> (/node_modules/thunk-disque/lib/connection.js:119:11)
    at Socket.g (events.js:260:16)
    at emitOne (events.js:77:13)
    at Socket.emit (events.js:169:7)
    at TCP._onclose (net.js:469:12)
[nodemon] app crashed

Configuration:
Mac OS X El Cap - 10.11.2
node.js - 5.3.0
thunks-disque - master @ 3393055

QSTAT only returns data once in a while

When I run the QSTAT command it mostly returns null. However, every so often it will come back with the stats for the queue. I have found that a number of attempts in quick succession often result in the stats showing up eventually.

Perhaps this has something to do with the client not waiting for the result of the QSTAT command, then receiving them on subsequent calls because the data becomes available later.

I am have configured the client with usePromise set to true. I am connected to a 3-node cluster, with each host on a separate IP.

Please let me know if there is any additional detail I can provide.

Recommend Projects

thunks / thunk-disque Goto Github PK