llmongodb/main.go

// llmongo.go
package main

import (
	"flag"
	"fmt"
	"log"
	"os"
	"regexp"
	"sync"
	"time"

	_ "github.com/nats-io/go-nats"
)

const (
	_Version  = "v5.0.0b4"
	_Producer = 0
	_Consumer = 1
	_Remover  = 2
)

var (
	loop bool

	done    chan bool
	consume chan loginsList
	remove  chan loginsList

	counter chan Counterchan

	wg sync.WaitGroup

	status int

	count *Counter
)

func main() {
	flag.Usage = usage
	flag.Parse()

	if opts.Version {
		fmt.Println(os.Args[0], _Version)
		os.Exit(0)
	}

	if dbs.MongoURI == "" || dbs.RedisURI == "" || dbs.Database == "" {
		flag.Usage()
		os.Exit(-1)
	}

	if opts.Influxdb != "" {
		var re = regexp.MustCompile(`(?m)(\w+)@([\w.]+):(\d+)`)
		//		re, _ := regexp.Compile(`(\w+)@(\d+.\d+.\d+.\d+:\d+)`)
		if re.MatchString(opts.Influxdb) {
			match := re.FindStringSubmatch(opts.Influxdb)
			if opts.Debug {
				fmt.Printf("Influxdb match: %+v\n", match)
			}
			infhost = match[1]
			infdb = match[2] + ":" + match[3]
		} else {
			opts.Influxdb = ""
		}
	}

	setTerm()

	fs, err := os.OpenFile(opts.LogFile, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0666)
	if err != nil {
		fmt.Println("Log file error: ", err.Error())
		os.Exit(-4)
	}
	defer fs.Close()

	log.SetOutput(fs)

	//	pid.PIDFile = opts.Pidfile
	//	pid.Write(true)
	//	defer pid.Remove()

	start := time.Now()
	fmt.Printf("Start:\t%+v\n\t%+v\n", opts, dbs)
	log.Printf("Start:\t%+v\n\t%+v\n", opts, dbs)

	opts.Month = start.Format("0601")

	dbs.poolRedis()
	defer dbs.rdb.Close()

	dbs.connectMongo()
	for k := range dbs.mdb {
		defer dbs.mdb[k].Close()
	}

	if opts.Timeout > 0 {
		time.AfterFunc(opts.Timeout, exit)
	}

	count = NewCounter()

	consume = make(chan loginsList, opts.Queue)
	remove = make(chan loginsList, opts.Queue)
	loop = true
	done = make(chan bool)
	counter = make(chan Counterchan)

	go count.Run()
	go producer()
	for i := 0; i < opts.Queue; i++ {
		for j := 0; j < len(dbs.mdb); j++ {
			go consumer()
		}
		go remover()
	}

	<-done
	fmt.Printf("Done\n")
	close(done)

	fmt.Println("Waiting WG")
	wg.Wait()

	count.SetTime(time.Since(start))

	fmt.Printf("Stop %v - user: %d - login: %d - insert: %d - errors: %d - rem: %d - duplicate: %d\n\r", count.GetTime(), count.GetUser(), count.GetLog(), count.GetInsert(), count.GetErr(), count.GetRem(), count.GetDup())
	log.Printf("Stop %v - user: %d - login: %d - insert: %d - errors: %d - rem: %d - duplicate: %d\n\r", count.GetTime(), count.GetUser(), count.GetLog(), count.GetInsert(), count.GetErr(), count.GetRem(), count.GetDup())

	if opts.Influxdb != "" {
		writeStats(start)
	}
}