pipeline.v

// Copyright (c) 2019 Mike 'Fuzzy' Partin
// Copyright (c) 2019 Neal Zerby
// Usage of this source code is governed by the 3 clause BSD License
// that can be found in the LICENSE.md file

module main

import sync
import time

// Pipe object and new() method

struct Pipe {
mut:
	// necessary for thread safety
	lock sync.Mutex
	// exit signal for all threads
	exit bool
	// map of named collectors
	collectors map[string]Collecter
	// metric queue, implimented as a FIFO
	queue []Metric
}

fn new_pipe(s int) Pipe {
	mut retv := Pipe{exit: false, collectors: [Collecter{} ; 50], queue: [Metric{} ; s]}
	retv.collectors.delete(0)
	retv.queue.delete(0)
	return retv
}

// Pipe object utility methods

pub fn (p mut Pipe) acquire() {
	p.lock.lock()
}

pub fn (p mut Pipe) release() {
	p.lock.unlock()
}

pub fn (p mut Pipe) exit() {
	p.exit = true
}

// Pipe object queue methods

pub fn (p mut Pipe) put(m Metric) {
	p.acquire()
	p.queue << m
	p.release()
}

pub fn (p mut Pipe) get() ?Metric {
	p.acquire()
	retv := p.queue.first() or {
		return error('No elements in queue')
	}
	p.queue.delete(0)
	p.release()
	return retv
}

// Pipe object collector methods

pub fn (p mut Pipe) register(n string, c Collecter) bool {
	p.acquire()
	if n in p.collectors.keys() {
		return false
	}
	p.collectors[n] = c
	return true
	p.release()
}

// Pipe object output

pub fn (p mut Pipe) exporter() {
	for {
		if p.exit {
			return
		}
		data := p.get() or {
			continue
		}
		
	}
}

// Pipe object scheduler

pub fn (p mut Pipe) collector() {
	for {
		if p.exit {
			return
		}
		for key in p.collectors.keys() {
			go p.collect(key)
		}
		time.sleep(10)
	}
}

fn (p mut Pipe) collect(c string) {
	p.collectors[c].collect(mut p)
}