pascalopitz/nodestalker

A Beanstalk client utilising node.js Tested for beanstalkd 1.4.6

INSTALL

npm install nodestalker

USAGE

Simple usage example:

var bs = require('nodestalker'),
    client = bs.Client('127.0.0.1:11300');

client.use('default').onSuccess(function(data) {
  console.log(data);

  client.put('my job').onSuccess(function(data) {
	console.log(data);
	client.disconnect();
  });
});

How do I reserve multiple items?

Each client basically represents one open socket to beanstalkd. Each command call just pumps one command into that socket, which then expects a corresponding return.

The server should maintain the state of the socket. However, reserve (or reserve with timeout) will only pull one job. You should then be able to reserve again on the same socket with the state of watch and ignore still preserved by the server.

Probably the most common usage scenario:

var bs = require('nodestalker'),
    client = bs.Client('127.0.0.1:11300'),
    tube = 'test_tube';

client.watch(tube).onSuccess(function(data) {
    function resJob() {
        client.reserve().onSuccess(function(job) {
            console.log('reserved', job);

            client.deleteJob(job.id).onSuccess(function(del_msg) {
                console.log('deleted', job);
                console.log('message', del_msg);
                resJob();
            });
        });
    }

    resJob();
});

If you want to do this fully in a fully asynchronous way, because there's a blocking process happening otherwise, you'll have to work with multiple sockets. This means that you'll have to repeat watch and ignore commands for each socket.

var bs = require('nodestalker'),
    tube = 'test_tube';

function processJob(job, callback) {
    // doing something really expensive
    console.log('processing...');
    setTimeout(function() {
        callback();
    }, 1000);
}

function resJob() {
    var client = bs.Client('127.0.0.1:11300');

    client.watch(tube).onSuccess(function(data) {
        client.reserve().onSuccess(function(job) {
            console.log('received job:', job);
            resJob();

            processJob(job, function() {
                client.deleteJob(job.id).onSuccess(function(del_msg) {
                    console.log('deleted', job);
                    console.log(del_msg);
                    client.disconnect();
                });
                console.log('processed', job);
            });
        });
    });
}

resJob();

DOCUMENTATION

npm run-script docs

Annotated source is now in the docs folder

TESTING

Also there are some tests now. Please make sure beanstalkd is running on the default settings.

To run all tests:

npm test

CREDIT

Depends on the yaml package by nodeca.

https://github.com/nodeca/js-yaml

Thanks to people that took time to fix some things.

aahoughton andho jney nmcquay tokudu justinwalsh yeldarby cincodenada

nodestalker's Issues

Issue with large jobs?

I am running into a problem where reserve just returns the initial line - id . Hence the job object is empty.

This only seems to happen when my jobs exceed a certain size. Have been able to reproduce with 100KB job size. I have up'd the max job size limit for bs.

Anybody else see such a behavior?

Loop didn't work?

hi mate,
I try to put new job to beanstalkd, it always only one job. why?

var b_client      = require('nodestalker').Client();

for (var i=0;i<5;i++){
  b_client.use('test_tube').onSuccess(function(data) {
    b_client.put("foo", 100, 0).onSuccess(function(data) {
      b_client.disconnect();
      console.log('put success');
    });
  });
} 

Feature request: Better chaining

Currently we end up with callback function after callback frunction.
See http://www.learnboost.com/mongoose/ for an example of how chaining should work.

client on error not firing

I have a simple tube client, where I do the following:

main_tube = nodestalk.Client();
try {
main_tube.use("main").onSuccess(function(data){
});
main_tube.addListener('error', function() {
console.log("main tube error");
});
catch (e){
throw e;
}

If I start this, with beanstalkd running, it works fine, code later can do a main_tube.put()...however, if I kill the beanstalkd process, I never get a "error" fired when I try to "put" here, I only get a exceptiono (uncaughtException)

Error: Socket is not writable
at Socket._writeOut (net.js:391:11)
at Socket.write (net.js:377:17)
at Array. (/usr/local/lib/node/.npm/nodestalker/0.1.0/package/lib/beanstalk_client.js:146:18)
at EventEmitter._tickCallback (node.js:126:26)

Is there a way to catch this without having to use a process.on("uncaughtException").

?

multiple put events, same client?

Should this be possible? Is hanging for me (pseudo code-->)

var client = new ns.Client();

client.use("foo").onSuccess(function(){
events.forEach(function(event){
client.put(event).onSuccess(function(){});
});

ie, put multiple events via the same client connection?

too fast gets corrupted jobs?

Write a put test that puts say 1000 jobs into a queue.

If you run this, you get assertion errors on undefined jobs (although using beanstalk telnet to checkout the queue all the jobs seem okay). If you put a setTimeout(function(){ watch()}, 500); around the call to the next watch(), I have yet to see an error. Without the timeout, get a corrupted job almost immediately

var ns = require('nodestalker');
var assert = require('assert');

var client = ns.Client();
var processed = 0;
var error = 0;

console.log("tube: " + process.argv[2]);
function watch(){
client.watch(process.argv[2]).onSuccess(function(){
client.reserve().onSuccess(function(data){
console.log(data);
console.log("processed: " + processed++);
assert.ok(data.id);
client.deleteJob(data.id).onSuccess(function(data){
console.log("deleted: " + data.id + " " + data);
watch();
});
});
});
}

watch();

Handling connection error to beanstalk

This is part of my code and it seems to raise an error when I try to run it when Beanstalk is not running.

var bs = require('nodestalker'),
        client = bs.Client(beanstalkdHost),
        tube = 'my-tube';

client.watch(tube).onSuccess(function (data) {
    // DO STUFF
)};

Error:

events.js:85
throw er; // Unhandled 'error' event
^
Error: connect ECONNREFUSED
at exports._errnoException (util.js:746:11)
at TCPConnectWrap.afterConnect as oncomplete

What I'd like to do is catch this error and retry in 5 seconds to connect and see if Beanstalk is responding. Right now I can't come up with a way to catch this error. And how can I see, if client is connected?

TypeError: Cannot read property 'write' of null at node_modules/nodestalker/lib/beanstalk_client.js:254:15

Hey guys!

I'm using [email protected] and from time to time I get this error. Error that crashes my Node.js process,...

TypeError: Cannot read property 'write' of null
  at <dir>/node_modules/nodestalker/lib/beanstalk_client.js:254:15
  at process._tickDomainCallback (node.js:381:11)

Any idea what is going on or how can I fix this?

This is how I'm using nodestalker:

bsClient = bs.Client(config.get('beanstalkd:host') + ':' + config.get('beanstalkd:port'))
.on 'error', (err)->
  log.error err
  process.exit(1)

putOnTube = (tubeName, payload, priority, delay, ttr, client, callback)->
  try
    client.use(tubeName).onError (err)->
      callback(err, null)
      return
    .onSuccess (pom)->
      pom = null
      client
      .put(payload, priority, delay, ttr)
      .onError (err)->
        callback(err, null)
        return
      .onSuccess (data)->
        client.disconnect()
        callback(null, data)
        return
      return
  catch error
    log.error(error)
    pmx.notify(error)
    callback(error, null)
  return

# Somewhere...
putOnTube 'my_tube', JSON.stringify(is_ok: "ok"), 0, 0, 60, bsClient, callback

Thank you for your comments and effort!

• Oto

Race conditions around "use" method

I've been trying to use nodestalker in a project of mine that routes work to different tubes in beanstalkd. The work gets routed as quickly as we find it, but it regularly was going to a tube other than the one I expected. When I started debugging I could not seem to find a reliable way to guarantee that work was going to go to the tube I expected it to go to.

var ns = require( 'nodestalker' );
ns.Debug.activate();
var Beanstalkd = ns.Client({address: '127.0.0.1', port: null});
Beanstalkd.use( 'useMe' ).onSuccess(function() {
    Beanstalkd.put( "useMeWork", 100, 0, 100 );
});
Beanstalkd.use( 'default' );
setTimeout( function() {
    Beanstalkd.put( "defaultWork", 100, 0, 100 );
}, 100);

The above snippet of code should recreate the problem. From the above snippet I would expect:
tube: useMe to contain work: "useMeWork"
-and-
tube: default to contain work: "defaultWork"

Instead what ends up happening is that the code connects to the 'default' tube first. Then shortly after it connects to the 'useMe' tube. The onSuccess never fires, and after 100ms elapses the "defaultWork" ends up in the 'useMe' tube.

If the setTimeout is removed no work gets created at all. nodestalker tries to process the "put" before the tube has been selected, throws it into the local queue. Then never seems to trigger any of the events that need to be triggered to process the local queue.

Running:
node: v0.10.22
nodestalker: v0.1.8

looks like there is a problem getting a connection after about 20000-30000 requests are processed.

I'm running a batch process and it hits this error consistently after 20000-30000 jobs.

version 0.1.21

events.js:141
throw er; // Unhandled 'error' event
^

Error: connect EADDRNOTAVAIL localhost:11300 - Local (localhost:0)
at Object.exports._errnoException (util.js:856:11)
at exports._exceptionWithHostPort (util.js:879:20)
at connect (net.js:843:14)
at net.js:938:9
at nextTickCallbackWith0Args (node.js:452:9)
at process._tickCallback (node.js:381:13)

Expected behavior when setting a watch on a tube

Hi Pascal

All is going fairly well but there remains one last thing I was not sure was entirely of my making.

I have successfully been putting and consuming jobs from tubes using nodestalker. One strange case exists, here's what I think I am seeing

• Client A puts jobs A,B,C into tube (nothing is watching)
• Client A doesn't disconnect, but is inactive
• Client B signs up to watch tube that is holding jobs A,B,C
• .. however strangely it doesn't immediate consume them.. ever even
• .. but it does consume new puts while connected (just not ones placed before connecting)

I notice in your tests that all the watchers are all connected and watching as you put. I was wondering whether you had explored the case where its a watch-less put followed by watch?

Cheers

Jerome

Pause example

Hello can you provide a test case when the need arises to pause?

Is nodestalker client stateful?

Hi there

I've been using the nodestalker and it's great. Overall works just as described, but I was a bif confused about something I was finding when integration testing the code we have:

This is what we define for queue/tube consumption

var queueClient = _self.queueClient;
queueClient.watch('queue').onSuccess(function(data) {

    queueClient.reserve_with_timeout(1).onSuccess(function(queueMessage) {
        var messageJson = JSON.parse(queueMessage.data);
        queueClient.deleteJob(queueMessage.id).onSuccess(function (obj) {
            _self.emit('queueMessage', messageJson.messageType, messageJson.params);
        });

    });
});

What I'm finding is that when a client comes and populates 10 messages into the tube, this client code only consumes the first message, rather than working through all tube messages.

Do i need to create a new client connection after each reserve & delete? Would seem odd to have to do so.. Your tests all use disconnect, which I don't think I would expect to do..

If you could shed some light on how we can change the implementation we have to just sit there consuming tube messages as fast as they come in, that would be great!

ls

How to use reserve_with_timeout?

I'm trying to use reserve_with_timeout, but I'm not successful with it.

I need to wait for 60 secs for a job. If no job is available in 60 sec, I would like to disconnect my client.

I tried below code, but if there is no job available in 60 sec, it is not disconnecting the client(ie, the script is not ending. it waits for new job. when new job is available it ends). I tried to read the test cases for reserve_with_timeout, be didn't understand what happens when there is no job available within the specified time.

(function(){
    var bs = require('nodestalker'),
        client = bs.Client(),
        tube = 'ban_urls';

    client.watch(tube).onSuccess(function(data) {
        function resJob() {
            client.reserve_with_timeout(60).onSuccess(function(job) {
                console.log(job);
                client.disconnect();
            });
            console.log('timeout');
        }
        resJob();
    });
}());