added some more job sources

importer/fetch.go

@@ -27,27 +27,10 @@ package importer
 import (
 	"encoding/base64"
 	"io"
-	"sync/atomic"
 
 	"golang.org/x/crypto/blake2b"
 )
 
-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...
-		atomic.StoreUint32(&job.sourceSet, 1)
-		job.source = newJobSourceNull(10 * 1024 * 1024)
-		close(job.subC.sourceAttached)
-	}
-}
-
 type copyResult struct {
 	err  error
 	hash string

importer/importer.go

@@ -54,7 +54,8 @@ func (im *Importer) CreateJob(group string, id uint64, src url.URL, user, refID
 	// TODO: update this list once we implemented other sources
 	switch src.Scheme {
 	case SourceSchemeAttachment:
-		//	case SourceSchemeFlowJS:
+	case SourceSchemeFlowJS:
+	case SourceSchemeFake:
 	default:
 		return nil, ErrSourceNotSupported
 	}

importer/job.go

@@ -71,8 +71,6 @@ func (job *Job) run() error {
 	defer job.cleanup()
 	defer close(job.subC.done)
 
-	job.prepareSource()
-
 	if !atomic.CompareAndSwapUint32((*uint32)(&job.State), uint32(JobPending), uint32(JobRunning)) {
 		job.im.store.UpdateFileImportState(job.Group, job.ID, store.ImportAborted)
 		// the job was canceled before job.Start() could initialize the context and because of this
@@ -82,21 +80,22 @@ func (job *Job) run() error {
 	close(job.subC.running)
 	job.StartedAt.set(time.Now())
 	job.im.dbgLog.Printf("running import for %s/%d from: %s", job.Group, job.ID, job.Source.String())
-
 	job.im.store.UpdateFileImportState(job.Group, job.ID, store.ImportRunning)
+
+	if err := job.prepareSource(); err != nil {
+		job.im.store.UpdateFileImportState(job.Group, job.ID, store.ImportAborted)
+		return err
+	}
 	loudness, err := job.fetch()
 	if err != nil {
 		job.im.store.UpdateFileImportState(job.Group, job.ID, store.ImportAborted)
-		// send result to all done subscriptions
 		return err
 	}
 	if err = job.normalize(loudness); err != nil {
 		job.im.store.UpdateFileImportState(job.Group, job.ID, store.ImportAborted)
-		// send result to all done subscriptions
 		return err
 	}
 	job.im.store.UpdateFileImportState(job.Group, job.ID, store.ImportDone)
-	// send result to all done subscriptions
 	return err
 }
 
@@ -159,6 +158,10 @@ func (job *Job) AttachSource(length uint64, r io.Reader) (<-chan *JobSourceResul
 		return nil, ErrImportNotRunning
 	}
 
+	// only allow to attach external sources if the job's source was in fact an attachment URL
+	if job.Source.Scheme != SourceSchemeAttachment {
+		return nil, ErrSourceAlreadyAttached
+	}
 	if ok := atomic.CompareAndSwapUint32(&job.sourceSet, 0, 1); !ok {
 		return nil, ErrSourceAlreadyAttached
 	}

importer/job_source.go

@@ -26,27 +26,53 @@ package importer
 
 import (
 	"io"
+	"sync/atomic"
 	"time"
 )
 
-//******* null
+func (job *Job) prepareSource() error {
+	if job.Source.Scheme == SourceSchemeAttachment {
+		// the source will be attached using AttachSource()
+		return nil
+	}
+
+	atomic.StoreUint32(&job.sourceSet, 1)
+
+	switch job.Source.Scheme {
+	case SourceSchemeFlowJS:
+		job.source = newJobSourceFlowJS(job.Source)
 
-type JobSourceNull uint64
+		// TODO: implement other sources
 
-func newJobSourceNull(len uint64) *JobSourceNull {
-	src := JobSourceNull(len)
+	case SourceSchemeFake:
+		job.source = newJobSourceFake(job.Source)
+	default:
+		return ErrSourceNotSupported
+	}
+
+	close(job.subC.sourceAttached)
+	return nil
+}
+
+//******* fake
+
+type JobSourceFake uint64
+
+func newJobSourceFake(srcURL SourceURL) *JobSourceFake {
+	// TODO: parse size from srcURL
+	src := JobSourceFake(10 * 1024 * 1024) // simulate a 10 MB file...
 	return &src
 }
 
-func (src *JobSourceNull) Len() uint64 {
+func (src *JobSourceFake) Len() uint64 {
 	return uint64(*src)
 }
 
-func (src *JobSourceNull) Read(p []byte) (n int, err error) {
+func (src *JobSourceFake) Read(p []byte) (n int, err error) {
 	l := len(p)
-	if *src > JobSourceNull(l) {
-		*src = *src - JobSourceNull(l)
-		time.Sleep(20 * time.Millisecond)
+	if *src > JobSourceFake(l) {
+		*src = *src - JobSourceFake(l)
+		time.Sleep(200 * time.Millisecond)
 		return l, nil
 	}
 	l = int(*src)
@@ -54,7 +80,7 @@ func (src *JobSourceNull) Read(p []byte) (n int, err error) {
 	return l, io.EOF
 }
 
-func (src *JobSourceNull) Done(result *JobSourceResult) {
+func (src *JobSourceFake) Done(result *JobSourceResult) {
 	return
 }
 
@@ -87,3 +113,26 @@ func (src *JobSourceAttachment) Done(result *JobSourceResult) {
 	}
 	src.done <- result
 }
+
+//******* flow.js
+
+type JobSourceFlowJS struct {
+	// TODO: implement this
+}
+
+func newJobSourceFlowJS(srcURL SourceURL) *JobSourceFlowJS {
+	src := &JobSourceFlowJS{}
+	return src
+}
+
+func (src *JobSourceFlowJS) Len() uint64 {
+	return 0
+}
+
+func (src *JobSourceFlowJS) Read(p []byte) (n int, err error) {
+	return 0, ErrNotImplemented
+}
+
+func (src *JobSourceFlowJS) Done(result *JobSourceResult) {
+	return
+}

importer/types.go

@@ -40,6 +40,7 @@ const (
 
 	SourceSchemeAttachment = "attachment"
 	SourceSchemeFlowJS     = "flowjs"
+	SourceSchemeFake       = "fake"
 )
 
 //******* Errors