fetch.go

//
//  tank
//
//  Import and Playlist Daemon for autoradio project
//
//
//  Copyright (C) 2017-2018 Christian Pointner <equinox@helsinki.at>
//
//  This file is part of tank.
//
//  tank is free software: you can redistribute it and/or modify
//  it under the terms of the GNU General Public License as published by
//  the Free Software Foundation, either version 3 of the License, or
//  any later version.
//
//  tank is distributed in the hope that it will be useful,
//  but WITHOUT ANY WARRANTY; without even the implied warranty of
//  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
//  GNU General Public License for more details.
//
//  You should have received a copy of the GNU General Public License
//  along with tank. If not, see <http://www.gnu.org/licenses/>.
//

package importer

import (
	"io"
	"io/ioutil"
	"sync/atomic"
	"time"
	"unsafe"
)

type devNull uint64

func (d *devNull) Read(p []byte) (n int, err error) {
	l := len(p)
	if *d > devNull(l) {
		*d = *d - devNull(l)
		time.Sleep(20 * time.Millisecond)
		return l, nil
	}
	l = int(*d)
	*d = 0
	return l, io.EOF
}

func (job *Job) prepareSource() {
	// TODO: enable this as soon as attachment source is implemented
	//	if job.Source.Scheme == "attachment" {
	// the source will be attached using AttachSource()
	//		return
	//	}
	// TODO: implement other sources

	// simulate a 10 MB file...
	l := uint64(10 * 1024 * 1024)
	r := devNull(l)
	atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&job.source)), unsafe.Pointer(&JobSource{l, &r}))
	close(job.sourceAttached)
}

type progressWriter struct {
	job     *Job
	written uint64
	w       io.Writer
}

func (pw *progressWriter) Write(p []byte) (n int, err error) {
	n, err = pw.w.Write(p)
	if n > 0 {
		pw.written += uint64(n)
	}
	pw.job.Progress.set(StepFetching, float32(pw.written)/float32(pw.job.source.len))
	return
}

func (job *Job) fetch() (err error) {
	job.Progress.set(StepFetching, 0)

	// wait until source is attached
	select {
	case <-job.ctx.Done():
		return job.ctx.Err()
	case <-job.sourceAttached:
	}

	// TODO: use an actual converter here
	conv := ioutil.Discard

	done := make(chan error)
	go func() {
		// TODO: wrap job.source using io.TeeReader to compute source file hash
		_, err := io.Copy(&progressWriter{job, 0, conv}, job.source.r)
		done <- err
	}()

	select {
	case <-job.ctx.Done():
		return job.ctx.Err()
	case err = <-done:
	}

	// TODO: wait for converter to finish up and return the result

	job.Progress.set(StepFetching, 1)
	return err
}