llmongodb/consumer.go

// consumer
package main

import (
	"fmt"
	"github.com/garyburd/redigo/redis"
	"log"
	"strconv"
	"strings"
	"time"
)

func consumer(user string) {
	var date int64
	var lastval string
	var conn = dbs.rdb.Get()
	defer conn.Close()

	start := time.Now()
	for {
		// Estrae l'ultimo login dell'utente 'user'
		val, err := redis.String(conn.Do("LINDEX", user, "-1"))
		if err != nil {
			if opts.Debug {
				log.Printf("LINDEX error: %+v - %s\n\r", err, val)
				fmt.Printf("LINDEX error: %+v - %s\n\r", err, val)
			}
			if strings.Contains(err.Error(), "nil returned") {
				// se ha trovato user e righe di login
				if lastval != "" {
					// reinserisce l'ultimo login e imposta il ttl su Redis
					retval, errr := conn.Do("lpush", user, lastval)
					ttl, errt := conn.Do("expire", user, int(opts.RedisTTL.Seconds()))
					if opts.Debug {
						log.Printf("LPUSH retval: %+v %+v %+v %+v %s %s %d\n\r", retval, errr, ttl, errt, user, lastval, int(opts.RedisTTL.Seconds()))
						fmt.Printf("LPUSH retval: %+v %+v %+v %+v %s %s %d\n\r", retval, errr, ttl, errt, user, lastval, int(opts.RedisTTL.Seconds()))
					}
				}
			} else {
				if opts.Debug {
					fmt.Printf("SADD error: %+v - %s\n\r", err, val)
				}
				val, err := conn.Do("sadd", "llindex", user)
				if err != nil {
					log.Printf("SADD error: %+v - %s\n\r", err, val)
				}
			}
			break
		}
		// se la riga di login e' vuota
		if val == "" {
			log.Println("Login empty: ", user)
			retval, _ := conn.Do("lrem", user, "-1", val)
			log.Println("LREM retval: ", user, val, retval)
			//return
			continue
		}
		sval := strings.Split(val, ":")
		// se il formato della riga di login non e' corretto
		if sval[1] == "" {
			log.Println("Login format error: ", val, user)
			retval, _ := conn.Do("lrem", user, "-1", val)
			log.Println("LREM retval: ", user, val, retval)
			//return
			continue
		}
		// se il timestamp della riga di login non e' corretto
		date, err = strconv.ParseInt(sval[1], 10, 64)
		if err != nil {
			log.Printf("Date Error: %+v - %s\n", err, user)
			//return
			continue
		}
		ml := MongoLogin{
			User:     user,
			Protocol: sval[0],
			Ip:       sval[2],
			Date:     time.Unix(date, 0),
		}
		ind := Index{
			User: user,
			Date: time.Unix(date, 0),
		}
		// inserisce il login su Mongodb
		count++
		_, err = dbs.ll.Upsert(ind, ml)
		if err != nil {
			log.Printf("Insert error: %+v\n", err)
			// se l'errore non e' "duplicate key error" salta al prossimo senza cancellare niente
			if !strings.Contains(err.Error(), "E11000") {
				errCount += 1
				//return
				continue
			}
		}
		// cancella da Redis la riga di login inserita
		retval, err := conn.Do("lrem", user, "-1", val)
		if opts.Debug {
			log.Println("LREM retval: ", retval, user, val)
			fmt.Println("LREM retval: ", retval, user, val)
		}
		lastval = val
	}
	fmt.Printf("CONS: user=%s in %v - active=%d\n\r", user, time.Since(start), dbs.rdb.ActiveCount())
	log.Printf("CONS: user=%s in %v - active=%d\n\r", user, time.Since(start), dbs.rdb.ActiveCount())
}