attaching source works now

api/v1/api_imports.go

@@ -62,7 +62,26 @@ func (api *API) ReadImportOfFile() http.Handler {
 
 func (api *API) UploadFile() http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		sendWebResponse(w, http.StatusNotImplemented, ErrorResponse{Error: "uploading files not yet implemented"})
+		vars := mux.Vars(r)
+		id, err := idFromString(vars["file-id"])
+		if err != nil {
+			sendWebResponse(w, http.StatusBadRequest, ErrorResponse{Error: "invalid file-id: " + err.Error()})
+			return
+		}
+		job, err := api.importer.GetJob(vars["group-id"], id)
+		if err != nil {
+			sendError(w, err)
+			return
+		}
+
+		if err = job.AttachSource(uint64(r.ContentLength), r.Body); err != nil {
+			sendError(w, err)
+			return
+		}
+		<-job.Done()
+
+		// TODO: get result from job
+		sendWebResponse(w, http.StatusOK, nil)
 	})
 }
 

importer/fetch.go

@@ -25,6 +25,8 @@
 package importer
 
 import (
+	"crypto/sha256"
+	"encoding/hex"
 	"io"
 	"io/ioutil"
 	"sync/atomic"
@@ -47,18 +49,20 @@ func (d *devNull) Read(p []byte) (n int, err error) {
 }
 
 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.subC.sourceAttached)
+	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 {
@@ -85,14 +89,17 @@ func (job *Job) fetch() (err error) {
 		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() {
-		// TODO: wrap job.source using io.TeeReader to compute source file hash
-		_, err := io.Copy(&progressWriter{job, 0, conv}, job.source.r)
+		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
 	}()
 

importer/importer.go

@@ -52,7 +52,7 @@ func (im *Importer) ListJobs(group string) (Jobs, error) {
 
 func (im *Importer) CreateJob(group string, id uint64, src url.URL, user, refID string) (*Job, error) {
 	// TODO: update this list once we implemented other sources
-	if src.Scheme != "attachment" {
+	if src.Scheme != SourceSchemeAttachment {
 		return nil, ErrSourceNotSupported
 	}
 

importer/job.go

@@ -54,7 +54,7 @@ type Job struct {
 		sourceAttached chan struct{}
 		running        chan struct{}
 		done           chan struct{}
-	} `json:"-"`
+	}
 }
 
 type Jobs []*Job

importer/types.go

@@ -36,6 +36,8 @@ import (
 
 const (
 	DefaultBacklog = 100
+
+	SourceSchemeAttachment = "attachment"
 )
 
 //******* Errors