<commit_before>package kumo

import (
	"fmt"
	"io/ioutil"
	"path/filepath"
	"sync"
	"time"

	"github.com/sww/dumblog"
)

type Download struct {
	Stop           chan bool
	Queue          chan *Segment
	ConnectionPool ConnectionPool
	DecodeQueue    chan string
	Logger         *dumblog.DumbLog
	Progress       *Progress
	Wait           *sync.WaitGroup
	TempPath       string
}

func InitDownload(host, username, password string, port, connections int, w *sync.WaitGroup) (*Download, error) {
	connectionPool, err := InitConnectionPool(host, username, password, port, connections)
	if err != nil {
		return nil, err
	}

	return &Download{
		ConnectionPool: *connectionPool,
		Queue:          make(chan *Segment),
		Stop:           make(chan bool, 1),
		Wait:           w,
	}, nil
}

func (d *Download) Run() {
	for {
		select {
		case <-d.Stop:
			d.Logger.Print("Download Stopping")
			return
		case segment := <-d.Queue:
			go func() {
				defer d.Progress.Add(segment.Bytes)
				d.Logger.Print("[DOWNLOAD] Download.Run() got segment ", segment.Segment)

				connection := <-d.ConnectionPool.connections
				segmentName, err := d.download(segment, &connection)

				if err != nil {
					d.Wait.Done()
					d.Logger.Print("[DOWNLOAD] d.Wait.Done() because of err ", err)
					return
				}

				d.DecodeQueue <- segmentName
			}()
		default:
			time.Sleep(100 * time.Millisecond)
		}
	}
}

func group(group string, connection *Connection) error {
	_, err := connection.client.Group(group)
	if err != nil {
		return err
	}

	connection.group = group

	return nil
}

func (d *Download) download(segment *Segment, connection *Connection) (string, error) {
	defer func() { d.ConnectionPool.connections <- *connection }()

	d.Logger.Print("[DOWNLOAD] Group: ", segment.Group)
	if segment.Group != connection.group {
		d.Logger.Printf("[DOWNLOAD] Switching from group %v to %v", connection.group, segment.Group)
		// TODO: Handle error.<commit_msg>Set isBroken on download error.<commit_after>package kumo

import (
	"fmt"
	"io/ioutil"
	"path/filepath"
	"sync"
	"time"

	"github.com/sww/dumblog"
)

type Download struct {
	Stop           chan bool
	Queue          chan *Segment
	ConnectionPool ConnectionPool
	DecodeQueue    chan string
	Logger         *dumblog.DumbLog
	Progress       *Progress
	Wait           *sync.WaitGroup
	TempPath       string
}

func InitDownload(host, username, password string, port, connections int, w *sync.WaitGroup) (*Download, error) {
	connectionPool, err := InitConnectionPool(host, username, password, port, connections)
	if err != nil {
		return nil, err
	}

	return &Download{
		ConnectionPool: *connectionPool,
		Queue:          make(chan *Segment),
		Stop:           make(chan bool, 1),
		Wait:           w,
	}, nil
}

func (d *Download) Run() {
	for {
		select {
		case <-d.Stop:
			d.Logger.Print("Download Stopping")
			return
		case segment := <-d.Queue:
			go func() {
				defer d.Progress.Add(segment.Bytes)
				d.Logger.Print("[DOWNLOAD] Download.Run() got segment ", segment.Segment)

				connection := <-d.ConnectionPool.connections
				segmentName, err := d.download(segment, &connection)

				if err != nil {
					d.Progress.isBroken = true
					d.Wait.Done()
					d.Logger.Print("[DOWNLOAD] d.Wait.Done() because of err ", err)
					return
				}

				d.DecodeQueue <- segmentName
			}()
		default:
			time.Sleep(100 * time.Millisecond)
		}
	}
}

func group(group string, connection *Connection) error {
	_, err := connection.client.Group(group)
	if err != nil {
		return err
	}

	connection.group = group

	return nil
}

func (d *Download) download(segment *Segment, connection *Connection) (string, error) {
	defer func() { d.ConnectionPool.connections <- *connection }()

	d.Logger.Print("[DOWNLOAD] Group: ", segment.Group)
	if segment.Group != connection.group {
		d.Logger.Printf("[DOWNLOAD] Switching from group %v to %v", connection.group, segment.Group)
		// TODO: Handle error.
{'fraction_non_alphanumeric': 0.12780040733197556, 'fraction_numerical': 0.002036659877800407, 'mean_word_length': 3.1271008403361344, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 21, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5956354', 'n_tokens_mistral': 1403, 'n_tokens_neox': 1252, 'n_words': 331}