llmongodb/llmongo.js

#!/usr/bin/env node

var Redis = require('redis');

var MongoClient = require('mongodb').MongoClient

var redis, mongodb, lastlogin, mongoUri, redisUri;

var args = process.argv.slice(2);

args.forEach(function(val, index, array){
  switch(val) {
    case '-mongo':
      if(args[index+1].search('mongodb://') >= 0) {
        mongoUri = args[index+1];
      } else {
        mongoUri = 'mongodb://'+args[index+1];
      }
      break;
    case '-redis':
      if(args[index+1].match(/\S+:\d+/)) {
        redisUri = args[index+1];
      } else {
        console.log("Wrong redis URI");
        process.exit(0);
      }
      break;
  }
});

mongoUri = mongoUri || 'mongodb://127.0.0.1:27018/dovecot';
redisUri = redisUri || 'redis-ll.mail.tiscali.sys:6379';

console.log("Connecting redis: ", redisUri);

redis = Redis.createClient(redisUri.split(':')[1], redisUri.split(':')[0]) //new Redis(redisUri);

redis.on("error", function (err) {
    console.log("Error redis" + err);
    process.exit(-99);
});

console.log("Connecting mongodb: ", mongoUri);

MongoClient.connect(mongoUri, function(err, db){
  if(err){
    console.log("Mongo connect error:", err);
    if (redis ) {
      redis.end();
    };
    process.exit(-99);
  }
  lastlogin = db.collection('lastlogin');

  var quit = function(){
    redis.end();
    db.close();
    process.exit(0);
  }

  // Legge uno user dal SET llindex e lo elimina; viene eseguita fino a quando ci sono users.
  // Calcola quante righe di lastlogin ci sono e chiama la funzione ricorsiva readLogs
  var readUsers = function() {
    console.log("readUsers");
    redis.spop('llindex', function(err, user){
      if( err ) {
        console.log("Error redis spop llindex: ", err);
        quit();
        process.exit(-1);
      }
      if( user ) {
        console.log("readLogs");
        redis.llen(user, function(err, llen){
          if (err) {
            console.log("Error redis llen "+user+": ", err)
            quit();
            process.exit(-2);
          } else {
            console.log("logslen: ", llen);

            // Legge una riga di log, la elimina da redis e la inserisce su mongo.
            // Viene eseguito n-1 volte, con n = numero di righe di log.
            var readLogs = function(lindex, user, len) {
              if (lindex == len ){
                // raggiunto il numero di righe da leggere, riparte con un altro user ( se esiste )
                console.log("Done: ", user, lindex);
                readUsers();
              } else {
                console.log("rpop log: ", user);
                redis.rpop(user, function(err, log){
                  if ( err ) {
                    console.log("Error redis rpop "+user+": ",err);
                    quit();
                    process.exit(-3);
                  } else {
                    console.log("log: ", log);
                    // imap:1432630259:10.39.75.60
                    var slog = log.split(':');
                    lastlogin.insert({
                      user: user,
                      protocol: slog[0],
                      date: new Date(slog[1]+"000"),
                      ip: slog[2],
                    }, function(err, r){
                      if(err){
                        console.log("insert error: ",err);
                        redis.rpush(user, log, function(err, result){
                          quit();
                          process.exit(-4);
                        });
                      } else {
                          // Salvato il log su mongo, legge il successivo.
                          console.log("Saved: ", log, lindex);
                          readLogs(lindex+1, user, len);
                      }
                    });
                  }
                })
              };
            };
          }

          // Legge il primo record per lo user "user" per llen-1 volte.
          readLogs(0, user, llen-1);
        });
      } else {
        // Se non ci sono piu' users, chiude i DB ed esce.
        console.log("quit readUsers");
        quit();
      }
    });
  };

  readUsers();
});