main.go

// This services listens on log queue and saves log to database.
package main

import (
	"log"
	"time"
)

func main() {
	jobChannel := make(chan (int))
	go func() {
		for {
			jobChannel <- 1
			time.Sleep(1 * time.Second)
		}
	}()

	dispatcher := NewDispatcher(jobChannel, 2)
	dispatcher.Dispatch(func(i int) {
	})
	dispatcher.Wait()
}

// Dispatcher is
type Dispatcher struct {
	Concurrency int
	jobChannel  chan (int)
	doneChannel chan (bool)
	killChannel chan (bool)
}

// WorkFn is a function executed by a worker.
type WorkFn func(int)

// NewDispatcher takes
func NewDispatcher(jobChannel chan (int), concurrency int) *Dispatcher {
	return &Dispatcher{
		Concurrency: concurrency,
		jobChannel:  jobChannel,
	}
}

// Dispatch takes
func (dispatcher *Dispatcher) Dispatch(fn WorkFn) {
	for i := 0; i < dispatcher.Concurrency; i++ {
		go wrapWorkFn(dispatcher, fn, i)
	}
}

// Wait watkes for all workers to finish their jobs.
func (dispatcher *Dispatcher) Wait() {
	for i := 0; i < dispatcher.Concurrency; i++ {
		<-dispatcher.doneChannel
	}
}

// Kill send die message to all workers.
func (dispatcher *Dispatcher) Kill() {
	for i := 0; i < dispatcher.Concurrency; i++ {
		dispatcher.killChannel <- true
	}
}

func wrapWorkFn(dispatcher *Dispatcher, fn WorkFn, id int) {
	for {
		select {
		case job := <-dispatcher.jobChannel:
			log.Printf("%d: %d", id, job)
			// fn(job)
		case <-dispatcher.killChannel:
			dispatcher.doneChannel <- true
		}
	}
}