llmongodb/llmongo.js

#!/usr/bin/env node

var Redis = require('ioredis');

var MongoClient = require('mongodb').MongoClient

var redis, mongodb, lastlogin;
var mongoReplicaSet = 'mongodb://mongo-1.mail.tiscali.sys:27018,mongo-3.mail.tiscali.sys:27018,mongo-5.mail.tiscali.sys:27018/dovecot';

var quit = function() {
};

redis = new Redis('redis://redis-ll.mail.tiscali.sys:6379/0');

MongoClient.connect(mongoReplicaSet, function(err, db){
  if(err){
    console.log("err");
    if (redis ) {
      redis.quit();
    };
    process.exit(-1);
  }
  lastlogin = db.collection('lastlogin');

  var quit = function(){
    redis.quit();
    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(err);
        quit();
        process.exit(-1);
      }
      if( user ) {
        console.log("readLogs");
        redis.llen(user, function(err, llen){

          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(err);
                  quit();
                  process.exit(-1);
                }

                console.log("log: ", log);
                // imap:1432630259:10.39.75.60
                var slog = log.split(':');
                lastlogin.insert({
                  user: user,
                  protocol: slog[0],
                  date: slog[1],
                  ip: slog[2],
                }, function(err, r){
                  if(err){
                    console.log("insert error: ",err);
                  } 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();
});