Workflow de node - Ubicando números por proveedor

apibluekite.js

//Recordar que los módulos van en config.json y allí se crean las variables para poder usarlos.
var restify = require('restify');
var http = require('http');

var workflow = module.exports = {
    name: 'Looking for cellphone company',
    chain: [
    {
        name: 'Create an account in TIGO',
        timeout: 20,
        retry: 1,
        body: function (job, cb) {

            //Este número es el que vamos a obtener de la base de datos
            var readableStream = qs.stringify({number:'55252125'});

            var options = {
              payload: readableStream,
              headers: { 'Content-Type':'application/x-www-form-urlencoded',
              'Content-Length': readableStream.length }
            }

            //Si el post no funciona hay que revisar el token que viene de module2.js en un 'var'
            Wreck.post(job.params.urlapi + '/v1/services/gt/tigo-phone-postpay/accounts?access_token=' + job.params.token,options,function(error, response, payload){
            var parsed = parse(response.headers.link);
            var es = new EventSource(parsed.monitor.url);
              es.on('status', function(e) {
                //Acá se imprime el streaming del monitor mientras siga respondiendo
                console.log(e.data);
              }).on('result', function(e) {

                //Por acá se va cuando se termina el streaming
                var jsonData = JSON.parse(e.data);
                WreckAccount.get(jsonData.url.replace('accounts.gt','api'), function (err, res, payload) {

                  //Acá hice el get del account y es donde debo de revisar el status
                  console.log('Visité el link del account: ', payload);

                  //Esta es sólo una manera arbitraria de saber si continuamos
                  //buscando con los demás proveedores
                  if(Math.floor(Math.random() * (10 - 0 + 1) + 0) > 5){
                    job.params.continueFlag = 1;
                  }else{
                      job.params.continueFlag = 0;
                  }

                  return cb(null,'Finalizado')
                });
              }).on('error', function() {
                console.log('ERROR!');
                es.close()
                return cb(null,'Finalizado');
                //Uno de los dos retuns cb(null, finalizado) no debe de ir, aún no sé cuál
              });
            });
        }
    },
    {
        name: 'Create an account in CLARO',
        timeout: 20,
        retry: 1,
        body: function (job, cb) {

            if(jobs.params.continueFlag != 0){
                //Este número es el que vamos a obtener de la base de datos
                //podemos testearlo también usando el job.params.password que es
                //de donde viene el número random generado desde module2.js
                var readableStream = qs.stringify({number:'55252125'});

                var options = {
                  payload: readableStream,
                  headers: { 'Content-Type':'application/x-www-form-urlencoded',
                  'Content-Length': readableStream.length }
                }

                //Si el post no funciona hay que revisar el token que viene de module2.js en un 'var'
                Wreck.post(job.params.urlapi + '/v1/services/gt/claro-phone-postpay/accounts?access_token=' + job.params.token,options,function(error, response, payload){
                var parsed = parse(response.headers.link);
                var es = new EventSource(parsed.monitor.url);
                  es.on('status', function(e) {
                    //Acá se imprime el streaming del monitor mientras siga respondiendo
                    console.log(e.data);
                  }).on('result', function(e) {

                    //Por acá se va cuando se termina el streaming
                    var jsonData = JSON.parse(e.data);
                    WreckAccount.get(jsonData.url.replace('accounts.gt','api'), function (err, res, payload) {

                      //Acá hice el get del account y es donde debo de revisar el status
                      console.log('Visité el link del account: ', payload);

                      //Esta es sólo una manera arbitraria de saber si continuamos
                      //buscando con los demás proveedores
                      if(Math.floor(Math.random() * (10 - 0 + 1) + 0) > 5){
                        job.params.continueFlag = 1;
                      }else{
                          job.params.continueFlag = 0;
                      }

                      return cb(null,'Finalizado')
                    });
                  }).on('error', function() {
                    console.log('ERROR!');
                    es.close()
                    return cb(null,'Finalizado');
                    //Uno de los dos retuns cb(null, finalizado) no debe de ir, aún no sé cuál
                  });
                });
            }else{
              return
            }
        }
    },
    {

        name: 'Create an account in MOVISTAR',
        timeout: 20,
        retry: 1,
        body: function (job, cb) {

            if(jobs.params.continueFlag != 0){
                //Este número es el que vamos a obtener de la base de datos
                //podemos testearlo también usando el job.params.password que es
                //de donde viene el número random generado desde module2.js
                var readableStream = qs.stringify({number:'55252125'});

                var options = {
                  payload: readableStream,
                  headers: { 'Content-Type':'application/x-www-form-urlencoded',
                  'Content-Length': readableStream.length }
                }

                //Si el post no funciona hay que revisar el token que viene de module2.js en un 'var'
                Wreck.post(job.params.urlapi + '/v1/services/gt/movistar-phone-postpay/accounts?access_token=' + job.params.token,options,function(error, response, payload){
                var parsed = parse(response.headers.link);
                var es = new EventSource(parsed.monitor.url);
                  es.on('status', function(e) {
                    //Acá se imprime el streaming del monitor mientras siga respondiendo
                    console.log(e.data);
                  }).on('result', function(e) {

                    //Por acá se va cuando se termina el streaming
                    var jsonData = JSON.parse(e.data);
                    WreckAccount.get(jsonData.url.replace('accounts.gt','api'), function (err, res, payload) {

                      //Acá hice el get del account y es donde debo de revisar el status
                      console.log('Visité el link del account: ', payload);

                      //Esta es sólo una manera arbitraria de saber si continuamos
                      //buscando con los demás proveedores
                      if(Math.floor(Math.random() * (10 - 0 + 1) + 0) > 5){
                        job.params.continueFlag = 1;
                      }else{
                          job.params.continueFlag = 0;
                      }

                      return cb(null,'Finalizado')
                    });
                  }).on('error', function() {
                    console.log('ERROR!');
                    es.close()
                    return cb(null,'Finalizado');
                    //Uno de los dos retuns cb(null, finalizado) no debe de ir, aún no sé cuál
                  });
                });
            }else{
              return
            }
        }
    }
    ],
    timeout: 500,
    onerror: [ {
        name: 'Error general',
        body: function (job, cb){
            console.log('Error general');
            return cb('Error general');
        }
    }
    ]
};

config.json

{
  "backend": {
    "module": "wf-redis-backend",
    "opts": {
      "port": 6379,
      "host": "127.0.0.1",
      "db": 14
    }
  },
  "api": {
    "port": 8080
  },
  "runner": {
    "identifier": "cd925eef-93fb-4bfe-a820-2aaedf9fc006",
    "forks": 10,
    "do_fork": false,
    "run_interval": 250,
    "activity_interval": 30000,
    "sandbox": {
        "modules": {
            "restify": "restify",
            "uuid": "node-uuid",
            "EventSource": "eventsource",
            "http": "http",
            "Wreck": "wreck",
            "WreckAccount": "wreck"
        }
    }
  }
}

module2.js

if (process.argv.length < 4) {
    console.error('Username and password/token required as arguments');
    process.exit(1);
}

var $login = process.argv[2];
var $password = Math.floor(Math.random() * (50059999 - 50050000 + 1) + 50050000);
var $token = 'NjAyNTQ2NjYxN2EyZTM4ODEzNjExNWU5MTJmYjhhYTEzNjQwMzdkY3x8MDVmMjI2NjkzMzJhNTE0MjkwNDE0ZGFkMTdkY2YzNDVmMzI3NGMxYg==';
var $linkMonitor = 'Empty';
var $urlapi = 'http://api.qa.bluekite.com'//'http://10.9.32.158:3100';//'http://api.sandbox.bluekite.com:80';
var $contador = 1;
var $continueFlag = 1;

var util = require('util');
var assert = require('assert');
var path = require('path');
var fs = require('fs');
var Factory = require('wf').Factory;
var aWorkflow = require('./apibluekite');

aWorkflow.name = $login;

var config_file = path.normalize(__dirname + '/config.json');
fs.readFile(config_file, 'utf8', function (err, data) {
    if (err) {
        throw err;
    }

    var config = JSON.parse(data),
        Backend = require(config.backend.module),
        backend = new Backend(config.backend.opts),
        factory;

    backend.init(function () {
        factory = Factory(backend);
        factory.workflow(aWorkflow, function (err, wf) {
            assert.ifError(err);
            assert.ok(wf);
            var aJob = {
                target: '/gists',
                workflow: wf.uuid,
                params: {
                    login: $login,
                    password: $password,
                    token: $token,
                    linkMonitor: $linkMonitor,
                    urlapi: $urlapi,
                    contador: $contador,
                    continueFlag: $continueFlag
                }
            };
            factory.job(aJob, function (err, job) {
                assert.ifError(err);
                assert.ok(job);
                assert.equal(job.execution, 'queued');
                assert.ok(job.uuid);

                var intervalId = setInterval(function () {
                    backend.getJob(job.uuid, function (err, obj) {
                        assert.ifError(err);
                        if (obj.execution === 'queued') {
                            console.log('Job waiting to be processed');
                        } else if (obj.execution === 'running') {
                            console.log('Job in progress ...');
                        } else {
                            console.log('Job finished. Here come the results:');
                            console.log(util.inspect(obj, false, 8));
                            // Only one workflow with the same name, need to
                            // delete it to allow creating it again:
                            backend.deleteWorkflow(wf, function (err, res) {
                                assert.ifError(err);
                                assert.ok(res);
                                clearInterval(intervalId);
                                process.exit(0);
                            });
                        }
                    });
                }, 3000);
            });
    });
  });
});

package.json

{
  "author": "Pedro Palazón Candel <kusorbox@gmail.com> (http://www.joyent.com)",
  "name": "wf-redis-example",
  "description": "Node Workflow usage examples",
  "version": "0.10.0",
  "private": true,
  "homepage": "https://github.com/kusor/node-workflow-example",
  "repository": {
    "type": "git",
    "url": "git://github.com/kusor/node-workflow-example.git"
  },
  "engines": {
    "node": ">=0.8"
  },
  "dependencies": {
    "bunyan": "0.23.1",
    "eventsource": "^0.1.3",
    "node-uuid": "1.4.0",
    "pooling": "0.4.6",
    "restify": "2.6.1",
    "wf": "~0.10",
    "wf-redis-backend": "~0.10"
  },
  "devDependencies": {}
}