Merge pull request #155 from snickers/move-setupjob

pipeline: move SetupJob from downloaders pkg (close #151)

downloaders/downloader.go

@@ -1,15 +1,11 @@
 package downloaders
 
 import (
-	"net/url"
-	"path"
 	"strings"
 
 	"code.cloudfoundry.org/lager"
 	"github.com/flavioribeiro/gonfig"
 	"github.com/snickers/snickers/db"
-	"github.com/snickers/snickers/helpers"
-	"github.com/snickers/snickers/types"
 )
 
 // DownloadFunc is a function type for the multiple
@@ -27,44 +23,3 @@ func GetDownloadFunc(jobSource string) DownloadFunc {
 
 	return HTTPDownload
 }
-
-// SetupJob is responsible for set the initial state for a given
-// job before starting. It sets local source and destination
-// paths and the final destination as well.
-func SetupJob(jobID string, dbInstance db.Storage, config gonfig.Gonfig) (types.Job, error) {
-	job, err := dbInstance.RetrieveJob(jobID)
-	if err != nil {
-		return types.Job{}, err
-	}
-
-	localSource, err := helpers.GetLocalSourcePath(config, job.ID)
-	if err != nil {
-		return types.Job{}, err
-	}
-	job.LocalSource = localSource + path.Base(job.Source)
-
-	job.LocalDestination, err = helpers.GetLocalDestination(config, dbInstance, jobID)
-	if err != nil {
-		return types.Job{}, err
-	}
-
-	u, err := url.Parse(job.Destination)
-	if err != nil {
-		return types.Job{}, err
-	}
-	outputFilename, err := helpers.GetOutputFilename(dbInstance, jobID)
-	if err != nil {
-		return types.Job{}, err
-	}
-	u.Path = path.Join(u.Path, outputFilename)
-	job.Destination = u.String()
-
-	job.Status = types.JobDownloading
-	job.Details = "0%"
-	job, err = dbInstance.UpdateJob(job.ID, job)
-	if err != nil {
-		return types.Job{}, err
-	}
-
-	return job, nil
-}

downloaders/downloaders_test.go

@@ -62,19 +62,6 @@ var _ = Describe("Downloaders", func() {
 				ContainSubstring("No filename could be determined"),
 				ContainSubstring("The AWS Access Key Id you provided does not exist in our records")))
 		})
-
-		It("Should set the local source and local destination on Job", func() {
-			dbInstance.StoreJob(exampleJob)
-			downloader(logger, cfg, dbInstance, exampleJob.ID)
-			changedJob, _ := dbInstance.RetrieveJob("123")
-			swapDir, _ := cfg.GetString("SWAP_DIRECTORY", "")
-
-			sourceExpected := swapDir + "123/src/source_here.mp4"
-			Expect(changedJob.LocalSource).To(Equal(sourceExpected))
-
-			destinationExpected := swapDir + "123/dst/source_here_240p.mp4"
-			Expect(changedJob.LocalDestination).To(Equal(destinationExpected))
-		})
 	}
 
 	Context("HTTP Downloader", func() {
@@ -113,12 +100,13 @@ var _ = Describe("Downloaders", func() {
 		BeforeEach(func() {
 			downloader = S3Download
 			exampleJob = types.Job{
-				ID:          "123",
-				Source:      "http://AWSKEY:AWSSECRET@BUCKET.s3.amazonaws.com/source_here.mp4",
-				Destination: "s3://user@pass:/bucket/",
-				Preset:      types.Preset{Name: "240p", Container: "mp4"},
-				Status:      types.JobCreated,
-				Details:     "",
+				ID:               "123",
+				Source:           "http://AWSKEY:AWSSECRET@BUCKET.s3.amazonaws.com/source_here.mp4",
+				Destination:      "s3://user@pass:/bucket/",
+				Preset:           types.Preset{Name: "240p", Container: "mp4"},
+				Status:           types.JobCreated,
+				Details:          "",
+				LocalDestination: "/tmp/output_here.mp4",
 			}
 		})
 

downloaders/ftp.go

@@ -19,9 +19,8 @@ func FTPDownload(logger lager.Logger, config gonfig.Gonfig, dbInstance db.Storag
 	log.Info("start", lager.Data{"job": jobID})
 	defer log.Info("finished")
 
-	job, err := SetupJob(jobID, dbInstance, config)
+	job, err := dbInstance.RetrieveJob(jobID)
 	if err != nil {
-		log.Error("setting-up-job", err)
 		return err
 	}
 

downloaders/http.go

@@ -17,9 +17,8 @@ func HTTPDownload(logger lager.Logger, config gonfig.Gonfig, dbInstance db.Stora
 	log.Info("start", lager.Data{"job": jobID})
 	defer log.Info("finished")
 
-	job, err := SetupJob(jobID, dbInstance, config)
+	job, err := dbInstance.RetrieveJob(jobID)
 	if err != nil {
-		log.Error("setting-up-job", err)
 		return err
 	}
 

downloaders/s3.go

@@ -21,9 +21,8 @@ func S3Download(logger lager.Logger, config gonfig.Gonfig, dbInstance db.Storage
 	log.Info("start", lager.Data{"job": jobID})
 	defer log.Info("finished")
 
-	job, err := SetupJob(jobID, dbInstance, config)
+	job, err := dbInstance.RetrieveJob(jobID)
 	if err != nil {
-		log.Error("setting-up-job", err)
 		return err
 	}
 

pipeline/pipeline.go

@@ -1,13 +1,16 @@
 package pipeline
 
 import (
+	"net/url"
 	"os"
+	"path"
 
 	"code.cloudfoundry.org/lager"
 	"github.com/flavioribeiro/gonfig"
 	"github.com/snickers/snickers/db"
 	"github.com/snickers/snickers/downloaders"
 	"github.com/snickers/snickers/encoders"
+	"github.com/snickers/snickers/helpers"
 	"github.com/snickers/snickers/types"
 	"github.com/snickers/snickers/uploaders"
 )
@@ -22,6 +25,14 @@ func StartJob(logger lager.Logger, config gonfig.Gonfig, dbInstance db.Storage,
 	})
 	defer log.Info("finished")
 
+	log.Info("setup")
+	newJob, err := SetupJob(job.ID, dbInstance, config)
+	job = *newJob
+	if err != nil {
+		log.Error("setup-job failed", err)
+		return
+	}
+
 	log.Info("downloading")
 	downloadFunc := downloaders.GetDownloadFunc(job.Source)
 	if err := downloadFunc(log, config, dbInstance, job.ID); err != nil {
@@ -77,3 +88,44 @@ func CleanSwap(dbInstance db.Storage, jobID string) error {
 	err = os.RemoveAll(job.LocalDestination)
 	return err
 }
+
+// SetupJob is responsible for set the initial state for a given
+// job before starting. It sets local source and destination
+// paths and the final destination as well.
+func SetupJob(jobID string, dbInstance db.Storage, config gonfig.Gonfig) (*types.Job, error) {
+	job, err := dbInstance.RetrieveJob(jobID)
+	if err != nil {
+		return nil, err
+	}
+
+	localSource, err := helpers.GetLocalSourcePath(config, job.ID)
+	if err != nil {
+		return nil, err
+	}
+	job.LocalSource = localSource + path.Base(job.Source)
+
+	job.LocalDestination, err = helpers.GetLocalDestination(config, dbInstance, jobID)
+	if err != nil {
+		return nil, err
+	}
+
+	u, err := url.Parse(job.Destination)
+	if err != nil {
+		return nil, err
+	}
+	outputFilename, err := helpers.GetOutputFilename(dbInstance, jobID)
+	if err != nil {
+		return nil, err
+	}
+	u.Path = path.Join(u.Path, outputFilename)
+	job.Destination = u.String()
+
+	job.Status = types.JobDownloading
+	job.Details = "0%"
+	job, err = dbInstance.UpdateJob(job.ID, job)
+	if err != nil {
+		return nil, err
+	}
+
+	return &job, nil
+}

pipeline/pipeline_test.go

@@ -43,6 +43,31 @@ var _ = Describe("Pipeline", func() {
 		dbInstance.ClearDatabase()
 	})
 
+	Context("SetupJob function", func() {
+		It("Should set the local source and local destination on Job", func() {
+			exampleJob := types.Job{
+				ID:          "123",
+				Source:      "http://flv.io/source_here.mp4",
+				Destination: "s3://user@pass:/bucket/",
+				Preset:      types.Preset{Name: "240p", Container: "mp4"},
+				Status:      types.JobCreated,
+				Details:     "",
+			}
+
+			dbInstance.StoreJob(exampleJob)
+			SetupJob(exampleJob.ID, dbInstance, cfg)
+			changedJob, _ := dbInstance.RetrieveJob("123")
+
+			swapDir, _ := cfg.GetString("SWAP_DIRECTORY", "")
+
+			sourceExpected := swapDir + "123/src/source_here.mp4"
+			Expect(changedJob.LocalSource).To(Equal(sourceExpected))
+
+			destinationExpected := swapDir + "123/dst/source_here_240p.mp4"
+			Expect(changedJob.LocalDestination).To(Equal(destinationExpected))
+		})
+	})
+
 	Context("Pipeline", func() {
 		It("Should get the HTTPDownload function if source is HTTP", func() {
 			jobSource := "http://flv.io/KailuaBeach.mp4"