refactoring submit operation

1 files changed, 51 insertions, 38 deletions

diff --git a/go/operation/submit.go b/go/operation/submit.go
index b429ca02a..b157fc052 100644
--- a/go/operation/submit.go
+++ b/go/operation/submit.go
@@ -6,11 +6,18 @@ import (
 	"mime"
 	"os"
 	"path"
-	"path/filepath"
 	"strconv"
 	"strings"
 )
 
+type FilePart struct {
+	ReadCloser io.ReadCloser //required, all rest are optional
+	FileName   string
+	IsGzipped  bool
+	MimeType   string
+	ModTime    int64 //in seconds
+}
+
 type SubmitResult struct {
 	FileName string `json:"fileName"`
 	FileUrl  string `json:"fileUrl"`
@@ -19,28 +26,10 @@ type SubmitResult struct {
 	Error    string `json:"error"`
 }
 
-func Submit(master string, reader io.Reader, replication string) (result SubmitResult, err error) {
-	assignResult, assignError := Assign(master, 1, replication)
-	if assignError != nil {
-		result.Error = assignError.Error()
-		return
-	}
-	url := "http://" + assignResult.PublicUrl + "/" + assignResult.Fid
-	uploadResult, uploadError := Upload(url, "", reader, false, "")
-	if uploadError != nil {
-		result.Error = uploadError.Error()
-		return
-	}
-	result.Size = uploadResult.Size
-	result.FileUrl = url
-	result.Fid = assignResult.Fid
-	return result, nil
-}
-
-func SubmitFiles(master string, files []string, replication string) ([]SubmitResult, error) {
+func SubmitFiles(master string, files []FilePart, replication string) ([]SubmitResult, error) {
 	results := make([]SubmitResult, len(files))
 	for index, file := range files {
-		results[index].FileName = file
+		results[index].FileName = file.FileName
 	}
 	ret, err := Assign(master, len(files), replication)
 	if err != nil {
@@ -54,7 +43,7 @@ func SubmitFiles(master string, files []string, replication string) ([]SubmitRes
 		if index > 0 {
 			fid = fid + "_" + strconv.Itoa(index)
 		}
-		results[index].Size, err = upload(file, ret.PublicUrl, fid)
+		results[index].Size, err = file.Upload(ret.PublicUrl, fid)
 		if err != nil {
 			fid = ""
 			results[index].Error = err.Error()
@@ -65,25 +54,49 @@ func SubmitFiles(master string, files []string, replication string) ([]SubmitRes
 	return results, nil
 }
 
-func upload(filename string, server string, fid string) (int, error) {
-	glog.V(2).Info("Start uploading file:", filename)
-	fh, err := os.Open(filename)
-	if err != nil {
-		glog.V(0).Info("Failed to open file: ", filename)
-		return 0, err
+func NewFileParts(fullPathFilenames []string) (ret []FilePart, err error) {
+	ret = make([]FilePart, len(fullPathFilenames))
+	for index, file := range fullPathFilenames {
+		if ret[index], err = NewFilePart(file); err != nil {
+			return
+		}
 	}
-	fi, fiErr := fh.Stat()
-	if fiErr != nil {
-		glog.V(0).Info("Failed to stat file:", filename)
-		return 0, fiErr
+	return
+}
+func NewFilePart(fullPathFilename string) (ret FilePart, err error) {
+	fh, openErr := os.Open(fullPathFilename)
+	if openErr != nil {
+		glog.V(0).Info("Failed to open file: ", fullPathFilename)
+		return ret, openErr
+	}
+	ret.ReadCloser = fh
+
+	if fi, fiErr := fh.Stat(); fiErr != nil {
+		glog.V(0).Info("Failed to stat file:", fullPathFilename)
+		return ret, fiErr
+	} else {
+		ret.ModTime = fi.ModTime().UTC().Unix()
+	}
+	ext := strings.ToLower(path.Ext(fullPathFilename))
+	ret.IsGzipped = ext == ".gz"
+	if ret.IsGzipped {
+		ret.FileName = fullPathFilename[0 : len(fullPathFilename)-3]
 	}
-	filename = path.Base(filename)
-	isGzipped := path.Ext(filename) == ".gz"
-	if isGzipped {
-		filename = filename[0 : len(filename)-3]
+	ret.FileName = fullPathFilename
+	if ext != "" {
+		ret.MimeType = mime.TypeByExtension(ext)
+	}
+
+	return ret, nil
+}
+
+func (fi FilePart) Upload(server string, fid string) (int, error) {
+	fileUrl := "http://" + server + "/" + fid
+	if fi.ModTime != 0 {
+		fileUrl += "?ts=" + strconv.Itoa(int(fi.ModTime))
 	}
-	mtype := mime.TypeByExtension(strings.ToLower(filepath.Ext(filename)))
-	ret, e := Upload("http://"+server+"/"+fid+"?ts="+strconv.Itoa(int(fi.ModTime().Unix())), filename, fh, isGzipped, mtype)
+	defer fi.ReadCloser.Close()
+	ret, e := Upload(fileUrl, fi.FileName, fi.ReadCloser, fi.IsGzipped, fi.MimeType)
 	if e != nil {
 		return 0, e
 	}