//Delegates the files to its workers.
func (s *NagiosSpoolfileCollector) run() {
	promServer := statistics.GetPrometheusServer()
	for {
		select {
		case <-s.quit:
			s.quit <- true
			return
		case <-time.After(IntervalToCheckDirectory):
			pause := config.PauseNagflux.Load().(bool)
			if pause {
				logging.GetLogger().Debugln("NagiosSpoolfileCollector in pause")
				continue
			}

			logging.GetLogger().Debug("Reading Directory: ", s.spoolDirectory)
			files, _ := ioutil.ReadDir(s.spoolDirectory)
			promServer.SpoolFilesOnDisk.Set(float64(len(files)))
			for _, currentFile := range files {
				select {
				case <-s.quit:
					s.quit <- true
					return
				case s.jobs <- path.Join(s.spoolDirectory, currentFile.Name()):
				case <-time.After(time.Duration(1) * time.Minute):
					logging.GetLogger().Warn("NagiosSpoolfileCollector: Could not write to buffer")
				}
			}
		}
	}
}

//Checks if the files are old enough, if so they will be added in the queue
func (nfc FileCollector) run() {
	for {
		select {
		case <-nfc.quit:
			nfc.quit <- true
			return
		case <-time.After(spoolfile.IntervalToCheckDirectory):
			pause := config.PauseNagflux.Load().(bool)
			if pause {
				logging.GetLogger().Debugln("NagfluxFileCollector in pause")
				continue
			}
			for _, currentFile := range spoolfile.FilesInDirectoryOlderThanX(nfc.folder, spoolfile.MinFileAge) {
				for _, p := range nfc.parseFile(currentFile) {
					for _, r := range nfc.results {
						select {
						case <-nfc.quit:
							nfc.quit <- true
							return
						case r <- p:
						case <-time.After(time.Duration(1) * time.Minute):
							nfc.log.Warn("NagfluxFileCollector: Could not write to buffer")
						}
					}
				}
				err := os.Remove(currentFile)
				if err != nil {
					logging.GetLogger().Warn(err)
				}
			}
		}
	}
}

//NewPrometheusServer creates a new PrometheusServer
func NewPrometheusServer(address string) PrometheusServer {
	pMutex.Lock()
	server = initServerConfig()
	pMutex.Unlock()
	if address != "" {
		go func() {
			http.Handle("/metrics", prometheus.Handler())
			if err := http.ListenAndServe(address, nil); err != nil {
				logging.GetLogger().Warn(err.Error())
			}
		}()
		logging.GetLogger().Infof("serving prometheus metrics at %s/metrics", address)
	}
	return server
}

//NewGearmanWorker generates a new GearmanWorker.
//leave the key empty to disable encryption, otherwise the gearmanpacketes are expected to be encrpyten with AES-ECB 128Bit and a 32 Byte Key.
func NewGearmanWorker(address, queue, key string, results map[data.Datatype]chan collector.Printable, livestatusCacheBuilder *livestatus.CacheBuilder) *GearmanWorker {
	var decrypter *crypto.AESECBDecrypter
	if key != "" {
		byteKey := ShapeKey(key, 32)
		var err error
		decrypter, err = crypto.NewAESECBDecrypter(byteKey)
		if err != nil {
			panic(err)
		}
	}
	worker := &GearmanWorker{
		quit:                  make(chan bool),
		results:               results,
		nagiosSpoolfileWorker: spoolfile.NewNagiosSpoolfileWorker(-1, make(chan string), make(map[data.Datatype]chan collector.Printable), livestatusCacheBuilder),
		aesECBDecrypter:       decrypter,
		worker:                createGearmanWorker(address),
		log:                   logging.GetLogger(),
		jobQueue:              queue,
	}
	go worker.run()
	go worker.handleLoad()
	go worker.handlePause()

	return worker
}

func TestConnectToLivestatus(t *testing.T) {
	//Create Livestatus mock
	livestatus := MockLivestatus{"localhost:6557", "tcp", map[string]string{"test\n\n": "foo;bar\n"}, true}

	go livestatus.StartMockLivestatus()
	connector := LivestatusConnector{logging.GetLogger(), livestatus.LivestatusAddress, livestatus.ConnectionType}

	csv := make(chan []string)
	finished := make(chan bool)
	go connector.connectToLivestatus("test\n\n", csv, finished)

	expected := []string{"foo", "bar"}

	waitingForTheEnd := true
	for waitingForTheEnd {
		select {
		case line := <-csv:
			if !reflect.DeepEqual(line, expected) {
				t.Errorf("Expected:%s result:%s", expected, line)
			}
		case result := <-finished:
			if !result {
				t.Error("Connector exited with error")
			}
			waitingForTheEnd = false
		case <-time.After(time.Duration(3) * time.Second):
			t.Error("Livestatus connection timed out")
		}
	}
	livestatus.StopMockLivestatus()
}

//Checks if the files are old enough, if so they will be added in the queue
func (nfc NagfluxFileCollector) run() {
	for {
		select {
		case <-nfc.quit:
			nfc.quit <- true
			return
		case <-time.After(spoolfile.IntervalToCheckDirectory):
			for _, currentFile := range spoolfile.FilesInDirectoryOlderThanX(nfc.folder, spoolfile.MinFileAgeInSeconds) {
				data, err := ioutil.ReadFile(currentFile)
				if err != nil {
					break
				}
				for _, line := range strings.SplitAfter(string(data), "\n") {
					line = strings.TrimSpace(line)
					if line == "" {
						continue
					}
					select {
					case <-nfc.quit:
						nfc.quit <- true
						return
					case nfc.results <- line:
					case <-time.After(time.Duration(1) * time.Minute):
						nfc.log.Warn("NagfluxFileCollector: Could not write to buffer")
					}
				}
				err = os.Remove(currentFile)
				if err != nil {
					logging.GetLogger().Warn(err)
				}
			}
		}
	}
}

//Stops his workers and itself.
func (s *NagiosSpoolfileCollector) Stop() {
	s.quit <- true
	<-s.quit
	for _, worker := range s.workers {
		worker.Stop()
	}
	logging.GetLogger().Debug("SpoolfileCollector stopped")
}

func TestNewCacheBuilder(t *testing.T) {
	logging.InitTestLogger()
	connector := &Connector{logging.GetLogger(), "localhost:6558", "tcp"}
	builder := NewLivestatusCacheBuilder(connector)
	if builder == nil {
		t.Error("Constructor returned null pointer")
	}
}

//Generates a worker and starts it.
func NagiosSpoolfileWorkerGenerator(jobs chan string, results chan interface{}, fieldseperator string, livestatusCacheBuilder *livestatus.LivestatusCacheBuilder) func() *NagiosSpoolfileWorker {
	workerId := 0
	regexPerformancelable, err := regexp.Compile(`([^=]+)=(U|[\d\.\-]+)([\w\/%]*);?([\d\.\-:[email protected]]+)?;?([\d\.\-:[email protected]]+)?;?([\d\.\-]+)?;?([\d\.\-]+)?;?\s*`)
	if err != nil {
		logging.GetLogger().Error("Regex creation failed:", err)
	}
	regexAltCommand, err := regexp.Compile(`.*\[(.*)\]\s?$`)
	if err != nil {
		logging.GetLogger().Error("Regex creation failed:", err)
	}
	return func() *NagiosSpoolfileWorker {
		s := &NagiosSpoolfileWorker{workerId, make(chan bool), jobs, results, statistics.NewCmdStatisticReceiver(), fieldseperator, livestatusCacheBuilder, regexPerformancelable, regexAltCommand}
		workerId++
		go s.run()
		return s
	}
}

//GetYearMonthFromStringTimeMs returns the year and the month of a string which is in ms.
func GetYearMonthFromStringTimeMs(timeString string) (int, int) {
	i, err := strconv.ParseInt(timeString[:len(timeString)-3], 10, 64)
	if err != nil {
		logging.GetLogger().Warn(err.Error())
	}
	date := time.Unix(i, 0)
	return date.Year(), int(date.Month())
}

//Prints the data in influxdb lineformat
func (notification LivestatusNotificationData) Print(version float32) string {
	notification.sanitizeValues()
	if version >= 0.9 {
		var tags string
		if notification.notification_type == "HOST\\ NOTIFICATION" {
			tags = ",type=host_notification"
		} else if notification.notification_type == "SERVICE\\ NOTIFICATION" {
			tags = ",type=service_notification"
		} else {
			logging.GetLogger().Warn("This notification type is not supported:" + notification.notification_type)
		}
		value := fmt.Sprintf("%s:<br> %s", strings.TrimSpace(notification.notification_level), notification.comment)
		return notification.genInfluxLineWithValue(tags, value)
	} else {
		logging.GetLogger().Fatalf("This influxversion [%f] given in the config is not supportet", version)
		return ""
	}
}

//PrintForElasticsearch prints in the elasticsearch json format
func (notification NotificationData) PrintForElasticsearch(version, index string) string {
	if helper.VersionOrdinal(version) >= helper.VersionOrdinal("2.0") {
		text := notificationToText(notification.notificationType)
		value := fmt.Sprintf("%s:<br> %s", strings.TrimSpace(notification.notificationLevel), notification.comment)
		return notification.genElasticLineWithValue(index, text, value, notification.entryTime)
	}
	logging.GetLogger().Criticalf("This elasticsearchversion [%f] given in the config is not supported", version)
	panic("")
}

func TestNewLivestatusCollector(t *testing.T) {
	livestatus := &MockLivestatus{"localhost:6559", "tcp", map[string]string{}, true}
	go livestatus.StartMockLivestatus()
	connector := &LivestatusConnector{logging.GetLogger(), "localhost:6559", "tcp"}
	collector := NewLivestatusCollector(make(chan interface{}), connector, "&")
	if collector == nil {
		t.Error("Constructor returned null pointer")
	}
	collector.Stop()
}

//PrintForElasticsearch prints in the elasticsearch json format
func (downtime DowntimeData) PrintForElasticsearch(version, index string) string {
	if helper.VersionOrdinal(version) >= helper.VersionOrdinal("2.0") {
		typ := `downtime`
		start := downtime.genElasticLineWithValue(index, typ, strings.TrimSpace("Downtime start: <br>"+downtime.comment), downtime.entryTime)
		end := downtime.genElasticLineWithValue(index, typ, strings.TrimSpace("Downtime end: <br>"+downtime.comment), downtime.endTime)
		return start + "\n" + end
	}
	logging.GetLogger().Criticalf("This elasticsearchversion [%f] given in the config is not supported", version)
	panic("")
}

func TestNewLivestatusCollector(t *testing.T) {
	livestatus := &MockLivestatus{"localhost:6559", "tcp", map[string]string{}, true}
	go livestatus.StartMockLivestatus()
	connector := &Connector{logging.GetLogger(), "localhost:6559", "tcp"}
	collector := NewLivestatusCollector(make(map[data.Datatype]chan collector.Printable), connector, false)
	if collector == nil {
		t.Error("Constructor returned null pointer")
	}
	collector.Stop()
}

//Starts the webserver.
func StartMonitoringServer(port string) *MonitoringServer {
	mutex.Lock()
	if singleMonitoringServer == nil && port != "" {
		singleMonitoringServer = &MonitoringServer{port, make(chan bool), logging.GetLogger(), statistics.NewSimpleStatisticsUser(), make(map[string][]int)}
		singleMonitoringServer.statisticUser.SetDataReceiver(statistics.NewCmdStatisticReceiver())
		go singleMonitoringServer.run()
	}
	mutex.Unlock()
	return singleMonitoringServer
}

//PrintForInfluxDB prints the data in influxdb lineformat
func (downtime DowntimeData) PrintForInfluxDB(version string) string {
	downtime.sanitizeValues()
	if helper.VersionOrdinal(version) >= helper.VersionOrdinal("0.9") {
		tags := ",type=downtime,author=" + downtime.author
		start := fmt.Sprintf("%s%s message=\"%s\" %s", downtime.getTablename(), tags, strings.TrimSpace("Downtime start: <br>"+downtime.comment), helper.CastStringTimeFromSToMs(downtime.entryTime))
		end := fmt.Sprintf("%s%s message=\"%s\" %s", downtime.getTablename(), tags, strings.TrimSpace("Downtime end: <br>"+downtime.comment), helper.CastStringTimeFromSToMs(downtime.endTime))
		return start + "\n" + end
	}
	logging.GetLogger().Criticalf("This influxversion [%f] given in the config is not supported", version)
	panic("")
}

func TestConnectToLivestatus(t *testing.T) {
	//Create Livestatus mock
	livestatus := MockLivestatus{"localhost:6560", "tcp", map[string]string{"test\n\n": "foo;bar\n"}, true}

	go livestatus.StartMockLivestatus()
	connector := Connector{logging.GetLogger(), livestatus.LivestatusAddress, livestatus.ConnectionType}
	if err := helper.WaitForPort("tcp", "localhost:6560", time.Duration(2)*time.Second); err != nil {
		panic(err)
	}
	csv := make(chan []string)
	finished := make(chan bool)
	go connector.connectToLivestatus("test\n\n", csv, finished)

	expected := []string{"foo", "bar"}

	waitingForTheEnd := true
	for waitingForTheEnd {
		select {
		case line := <-csv:
			if !reflect.DeepEqual(line, expected) {
				t.Errorf("Expected:%s result:%s", expected, line)
			}
		case result := <-finished:
			if !result {
				t.Error("Connector exited with error")
			}
			waitingForTheEnd = false
		case <-time.After(time.Duration(3) * time.Second):
			t.Error("Livestatus connection timed out")
		}
	}
	livestatus.StopMockLivestatus()

	connector2 := Connector{logging.GetLogger(), "/live", "file"}
	csv2 := make(chan []string)
	finished2 := make(chan bool)
	go connector2.connectToLivestatus("test\n\n", csv2, finished2)
	if result := <-finished2; result {
		t.Error("Expected an error with unknown connection type")
	}
}

//Waits for files to parse and sends the data to the main queue.
func (w *NagiosSpoolfileWorker) run() {
	promServer := statistics.GetPrometheusServer()
	var file string
	for {
		select {
		case <-w.quit:
			w.quit <- true
			return
		case file = <-w.jobs:
			promServer.SpoolFilesInQueue.Set(float64(len(w.jobs)))
			startTime := time.Now()
			logging.GetLogger().Debug("Reading file: ", file)
			filehandle, err := os.OpenFile(file, os.O_RDONLY, os.ModePerm)
			if err != nil {
				logging.GetLogger().Warn("NagiosSpoolfileWorker: Opening file error: ", err)
				break
			}
			reader := bufio.NewReaderSize(filehandle, 4096*10)
			queries := 0
			line, isPrefix, err := reader.ReadLine()
			for err == nil && !isPrefix {
				splittedPerformanceData := helper.StringToMap(string(line), "\t", "::")
				for singlePerfdata := range w.PerformanceDataIterator(splittedPerformanceData) {
					for _, r := range w.results {
						select {
						case <-w.quit:
							w.quit <- true
							return
						case r <- singlePerfdata:
							queries++
						case <-time.After(time.Duration(1) * time.Minute):
							logging.GetLogger().Warn("NagiosSpoolfileWorker: Could not write to buffer")
						}
					}
				}
				line, isPrefix, err = reader.ReadLine()
			}
			if err != nil && err != io.EOF {
				logging.GetLogger().Warn(err)
			}
			if isPrefix {
				logging.GetLogger().Warn("NagiosSpoolfileWorker: filebuffer is too small")
			}
			filehandle.Close()
			err = os.Remove(file)
			if err != nil {
				logging.GetLogger().Warn(err)
			}
			promServer.SpoolFilesParsedDuration.Add(float64(time.Since(startTime).Nanoseconds() / 1000000))
			promServer.SpoolFilesLines.Add(float64(queries))
		case <-time.After(time.Duration(5) * time.Minute):
			logging.GetLogger().Debug("NagiosSpoolfileWorker: Got nothing to do")
		}
	}
}

//Waits for files to parse and sends the data to the main queue.
func (w *NagiosSpoolfileWorker) run() {
	var file string
	for {
		select {
		case <-w.quit:
			w.quit <- true
			return
		case file = <-w.jobs:
			startTime := time.Now()
			data, err := ioutil.ReadFile(file)
			if err != nil {
				break
			}
			lines := strings.SplitAfter(string(data), "\n")
			queries := 0
			for _, line := range lines {
				splittedPerformanceData := helper.StringToMap(line, "\t", "::")
				for singlePerfdata := range w.performanceDataIterator(splittedPerformanceData) {
					select {
					case <-w.quit:
						w.quit <- true
						return
					case w.results <- singlePerfdata:
						queries++
					case <-time.After(time.Duration(1) * time.Minute):
						logging.GetLogger().Warn("NagiosSpoolfileWorker: Could not write to buffer")
					}
				}
			}
			err = os.Remove(file)
			if err != nil {
				logging.GetLogger().Warn(err)
			}
			w.statistics.ReceiveQueries("read/parsed", statistics.QueriesPerTime{queries, time.Since(startTime)})
		case <-time.After(time.Duration(5) * time.Minute):
			logging.GetLogger().Debug("NagiosSpoolfileWorker: Got nothing to do")
		}
	}
}