// // 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 ( "crypto/sha256" "encoding/hex" "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() { switch job.Source.Scheme { case SourceSchemeAttachment: // the source will be attached using AttachSource() return // TODO: implement other sources default: // 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.subC.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.subC.sourceAttached: } // job.source is now initialized and points to a valid source // TODO: use an actual converter here conv := ioutil.Discard done := make(chan error) go func() { hash := sha256.New() src := io.TeeReader(job.source.r, hash) written, err := io.Copy(&progressWriter{job, 0, conv}, src) job.im.dbgLog.Printf("fetch(): done copying %d bytes from source (SHA256: %s)", written, hex.EncodeToString(hash.Sum(nil))) 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 }