Pipeline

import "sync"

type pipeline struct {
	client    *aPIClient
	next      chan bool
	downloads []*download
	whenDone  func()
	wg        *sync.WaitGroup
}

func newPipeline(cl *aPIClient, whenDone func()) *pipeline {
	return &pipeline{
		client:   cl,
		next:     make(chan bool, maxConcDLs),
		wg:       new(sync.WaitGroup),
		whenDone: whenDone,
	}
}

func (p *pipeline) start() {
	debugf("beginning to work on %d jobs", len(p.downloads))

	p.wg.Add(len(p.downloads))

	for i := 0; i < maxConcDLs; i++ {
		p.next <- true
	}

	go p.waitDLsFinished()
	go p.processDownloads()
}

func (p *pipeline) waitDLsFinished() {
	p.wg.Wait()

	debugf("all downloads finished. executing callback")
	close(p.next)
	go p.whenDone()
}

func (p *pipeline) processDownloads() {
	debugf("pipeline begins to wait")

	for range p.next {
		if len(p.downloads) == 0 {
			break
		}

		dl := p.downloads[0]
		p.downloads = p.downloads[1:]

		debugf("executing next download")
		go func() {
			p.client.download(dl.link, dl.imgFile)
			p.next <- true
			p.wg.Done()
		}()
	}

	debugf("all downloads started")
}

func (p *pipeline) addDownload(link string, imgFile *imageFile) {
	p.downloads = append(p.downloads, &download{link, imgFile})
}