initialize job source in one function

importer/fetch.go

@@ -39,13 +39,6 @@ type copyResult struct {
 func (job *Job) fetch() (interface{}, error) {
 	job.Progress.set(StepFetching, 0)
 
-	// wait until source is attached
-	select {
-	case <-job.ctx.Done():
-		return nil, job.ctx.Err()
-	case <-job.subC.sourceAttached:
-	}
-	// job.source is now initialized and points to a valid source
 	// make sure a potentially connected source gets notified in any case
 	defer job.source.Done(nil)
 

importer/job.go

@@ -82,10 +82,11 @@ func (job *Job) run() error {
 	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 {
+	if err := job.initializeSource(); err != nil {
 		job.im.store.UpdateFileImportState(job.Group, job.ID, store.ImportAborted)
 		return err
 	}
+	// job.source is now initialized and points to a valid source
 	loudness, err := job.fetch()
 	if err != nil {
 		job.im.store.UpdateFileImportState(job.Group, job.ID, store.ImportAborted)

importer/job_source.go

@@ -30,9 +30,15 @@ import (
 	"time"
 )
 
-func (job *Job) prepareSource() error {
+func (job *Job) initializeSource() error {
 	if job.Source.Scheme == SourceSchemeAttachment {
-		// the source will be attached using AttachSource()
+		// the source will be attached using job.AttachSource() so all we need to do
+		// is to wait for it to happen
+		select {
+		case <-job.ctx.Done():
+			return job.ctx.Err()
+		case <-job.subC.sourceAttached:
+		}
 		return nil
 	}
 
@@ -41,9 +47,6 @@ func (job *Job) prepareSource() error {
 	switch job.Source.Scheme {
 	case SourceSchemeFlowJS:
 		job.source = newJobSourceFlowJS(job.Source)
-
-		// TODO: implement other sources
-
 	case SourceSchemeFake:
 		job.source = newJobSourceFake(job.Source)
 	default: