<commit_before>package utils

import (
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"net/http"
	"os"
)

// Downloader is parallel HTTP fetcher
type Downloader struct {
	queue   chan *downloadTask
	stop    chan bool
	stopped chan bool
	threads int
}

// downloadTask represents single item in queue
type downloadTask struct {
	url         string
	destination string
	result      chan<- error
}

// NewDownloader creates new instance of Downloader which specified number
// of threads
func NewDownloader(threads int) (downloader *Downloader) {
	downloader = &Downloader{
		queue:   make(chan *downloadTask, 1000),
		stop:    make(chan bool),
		stopped: make(chan bool),
		threads: threads,
	}

	for i := 0; i < downloader.threads; i++ {
		go downloader.process()
	}

	return
}

// Shutdown stops downloader after current tasks are finished,
// but doesn't process rest of queue
func (downloader *Downloader) Shutdown() {
	for i := 0; i < downloader.threads; i++ {
		downloader.stop <- true
	}

	for i := 0; i < downloader.threads; i++ {
		<-downloader.stopped
	}
}

// Download starts new download task
func (downloader *Downloader) Download(url string, destination string) <-chan error {
	ch := make(chan error, 1)

	downloader.queue <- &downloadTask{url: url, destination: destination, result: ch}

	return ch
}

// DownloadTemp starts new download to temporary file and returns File
//
// Temporary file would be already removed, so no need to cleanup
func (downloader *Downloader) DownloadTemp(url string) (*os.File, error) {
	ch := make(chan error, 1)

	tempfile, err := ioutil.TempFile(os.TempDir(), "aptly")
	if err != nil {
		return nil, err
	}

	defer os.Remove(tempfile.Name())

	downloader.queue <- &downloadTask{url: url, destination: tempfile.Name(), result: ch}

	err = <-ch
	if err != nil {
		tempfile.Close()
		return nil, err
	}

	return tempfile, nil
}

// handleTask processes single download task
func (downloader *Downloader) handleTask(task *downloadTask) {
	log.Printf("Downloading %s...\n", task.url)

	resp, err := http.Get(task.url)
	if err != nil {
		task.result <- err
		return
	}
	defer resp.Body.Close()

	if resp.StatusCode < 200 || resp.StatusCode > 299 {
		task.result <- fmt.Errorf("HTTP code %d while fetching %s", resp.StatusCode, task.url)
		return
	}

	outfile, err := os.Create(task.destination)
	if err != nil {
		task.result <- err
		return
	}
	defer outfile.Close()

	io.Copy(outfile, resp.Body)

	task.result <- nil
}

// process implements download thread in goroutine
func (downloader *Downloader) process() {
	for {
		select {
		case <-downloader.stop:
			downloader.stopped <- true
			return
		case task := <-downloader.queue:
			downloader.handleTask(task)
		}
	}
}<commit_msg>Rework downloader to be interface + implementation.<commit_after>package utils

import (
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"net/http"
	"os"
)

// Downloader is parallel HTTP fetcher
type Downloader interface {
	Download(url string, destination string) <-chan error
	DownloadTemp(url string) (*os.File, error)
	Shutdown()
}

// Check interface
var (
	_ Downloader = &downloaderImpl{}
)

// downloaderImpl is implementation of Downloader interface
type downloaderImpl struct {
	queue   chan *downloadTask
	stop    chan bool
	stopped chan bool
	threads int
}

// downloadTask represents single item in queue
type downloadTask struct {
	url         string
	destination string
	result      chan<- error
}

// NewDownloader creates new instance of Downloader which specified number
// of threads
func NewDownloader(threads int) Downloader {
	downloader := &downloaderImpl{
		queue:   make(chan *downloadTask, 1000),
		stop:    make(chan bool),
		stopped: make(chan bool),
		threads: threads,
	}

	for i := 0; i < downloader.threads; i++ {
		go downloader.process()
	}

	return downloader
}

// Shutdown stops downloader after current tasks are finished,
// but doesn't process rest of queue
func (downloader *downloaderImpl) Shutdown() {
	for i := 0; i < downloader.threads; i++ {
		downloader.stop <- true
	}

	for i := 0; i < downloader.threads; i++ {
		<-downloader.stopped
	}
}

// Download starts new download task
func (downloader *downloaderImpl) Download(url string, destination string) <-chan error {
	ch := make(chan error, 1)

	downloader.queue <- &downloadTask{url: url, destination: destination, result: ch}

	return ch
}

// DownloadTemp starts new download to temporary file and returns File
//
// Temporary file would be already removed, so no need to cleanup
func (downloader *downloaderImpl) DownloadTemp(url string) (*os.File, error) {
	ch := make(chan error, 1)

	tempfile, err := ioutil.TempFile(os.TempDir(), "aptly")
	if err != nil {
		return nil, err
	}

	defer os.Remove(tempfile.Name())

	downloader.queue <- &downloadTask{url: url, destination: tempfile.Name(), result: ch}

	err = <-ch
	if err != nil {
		tempfile.Close()
		return nil, err
	}

	return tempfile, nil
}

// handleTask processes single download task
func (downloader *downloaderImpl) handleTask(task *downloadTask) {
	log.Printf("Downloading %s...\n", task.url)

	resp, err := http.Get(task.url)
	if err != nil {
		task.result <- err
		return
	}
	defer resp.Body.Close()

	if resp.StatusCode < 200 || resp.StatusCode > 299 {
		task.result <- fmt.Errorf("HTTP code %d while fetching %s", resp.StatusCode, task.url)
		return
	}

	outfile, err := os.Create(task.destination)
	if err != nil {
		task.result <- err
		return
	}
	defer outfile.Close()

	io.Copy(outfile, resp.Body)

	task.result <- nil
}

// process implements download thread in goroutine
func (downloader *downloaderImpl) process() {
	for {
		select {
		case <-downloader.stop:
			downloader.stopped <- true
			return
		case task := <-downloader.queue:
			downloader.handleTask(task)
		}
	}
}
{'fraction_non_alphanumeric': 0.11926286509040333, 'fraction_numerical': 0.00521557719054242, 'mean_word_length': 3.6734362307067423, 'pattern_counts': {'":': 0, '<': 40, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 104, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6093424', 'n_tokens_mistral': 1989, 'n_tokens_neox': 1843, 'n_words': 592}