added notification channels for running and done

importer/fetch.go

@@ -58,7 +58,7 @@ func (job *Job) prepareSource() {
 	l := uint64(10 * 1024 * 1024)
 	r := devNull(l)
 	atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(&job.source)), unsafe.Pointer(&JobSource{l, &r}))
-	close(job.sourceAttached)
+	close(job.subC.sourceAttached)
 }
 
 type progressWriter struct {
@@ -83,7 +83,7 @@ func (job *Job) fetch() (err error) {
 	select {
 	case <-job.ctx.Done():
 		return job.ctx.Err()
-	case <-job.sourceAttached:
+	case <-job.subC.sourceAttached:
 	}
 
 	// TODO: use an actual converter here

importer/job.go

@@ -37,20 +37,24 @@ import (
 
 // Job represents a import job for the file identified by Group and ID
 type Job struct {
-	im             *Importer
-	ctx            context.Context
-	cancel         context.CancelFunc
-	State          JobState    `json:"state"`
-	CreatedAt      Timestamp   `json:"created"`
-	StartedAt      Timestamp   `json:"started,omitempty"`
-	ID             uint64      `json:"id"`
-	Group          string      `json:"group"`
-	User           string      `json:"user"`
-	Source         SourceURL   `json:"source"`
-	RefID          string      `json:"ref-id,omitempty"`
-	Progress       JobProgress `json:"progress"`
-	source         *JobSource
-	sourceAttached chan struct{}
+	im        *Importer
+	ctx       context.Context
+	cancel    context.CancelFunc
+	State     JobState    `json:"state"`
+	CreatedAt Timestamp   `json:"created"`
+	StartedAt Timestamp   `json:"started,omitempty"`
+	ID        uint64      `json:"id"`
+	Group     string      `json:"group"`
+	User      string      `json:"user"`
+	Source    SourceURL   `json:"source"`
+	RefID     string      `json:"ref-id,omitempty"`
+	Progress  JobProgress `json:"progress"`
+	source    *JobSource
+	subC      struct {
+		sourceAttached chan struct{}
+		running        chan struct{}
+		done           chan struct{}
+	} `json:"-"`
 }
 
 type Jobs []*Job
@@ -62,6 +66,7 @@ func (job *Job) run() (err error) {
 	// either way job.cancel() needs to be called and the job needs to be removed from the inventory
 	defer job.cancel()
 	defer job.cleanup()
+	defer close(job.subC.done)
 
 	job.prepareSource()
 
@@ -71,7 +76,7 @@ func (job *Job) run() (err error) {
 		// job.Cancel() only set the state to JobCanceled... so we simulate a canceled context here
 		return context.Canceled
 	}
-
+	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())
 
@@ -149,7 +154,7 @@ func (job *Job) AttachSource(length uint64, r io.Reader) error {
 	if ok := atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&job.source)), unsafe.Pointer(nil), unsafe.Pointer(src)); !ok {
 		return ErrSourceAlreadyAttached
 	}
-	close(job.sourceAttached)
+	close(job.subC.sourceAttached)
 	return nil
 }
 
@@ -164,7 +169,13 @@ func (job *Job) Cancel() {
 	job.cancel()
 }
 
-// TODO: handle subscriptions to done
+func (job *Job) Running() <-chan struct{} {
+	return job.subC.running
+}
+
+func (job *Job) Done() <-chan struct{} {
+	return job.subC.done
+}
 
 func (job *Job) cleanup() {
 	atomic.StoreUint32((*uint32)(&job.State), uint32(JobDestroying))
@@ -178,6 +189,8 @@ func newJob(im *Importer, group string, id uint64, src url.URL, user, refID stri
 	job := &Job{im: im, Group: group, ID: id, Source: SourceURL(src), User: user, RefID: refID}
 	job.State = JobNew
 	job.CreatedAt.set(time.Now())
-	job.sourceAttached = make(chan struct{})
+	job.subC.sourceAttached = make(chan struct{})
+	job.subC.running = make(chan struct{})
+	job.subC.done = make(chan struct{})
 	return job
 }