Fix the function download command issue (#275)

* Fix the function download command issue
---

Fixes #270

*Motivation*

Fix the function download command doesn't write the response file
data into the target file.

**Modification**

- Fix download command
- Add upload command
- Add integration test for uploading and download the file and check the file sha256 value

pkg/cli/client.go

@@ -98,6 +98,11 @@ func (c *Client) Get(endpoint string, obj interface{}) error {
 
 func (c *Client) GetWithQueryParams(endpoint string, obj interface{}, params map[string]string,
 	decode bool) ([]byte, error) {
+	return c.GetWithOptions(endpoint, obj, params, decode, nil)
+}
+
+func (c *Client) GetWithOptions(endpoint string, obj interface{}, params map[string]string,
+	decode bool, file io.Writer) ([]byte, error) {
 
 	req, err := c.newRequest(http.MethodGet, endpoint)
 	if err != nil {
@@ -126,11 +131,18 @@ func (c *Client) GetWithQueryParams(endpoint string, obj interface{}, params map
 			return nil, err
 		}
 	} else if !decode {
-		body, err := ioutil.ReadAll(resp.Body)
-		if err != nil {
-			return nil, err
+		if file != nil {
+			_, err := io.Copy(file, resp.Body)
+			if err != nil {
+				return nil, err
+			}
+		} else {
+			body, err := ioutil.ReadAll(resp.Body)
+			if err != nil {
+				return nil, err
+			}
+			return body, err
 		}
-		return body, err
 	}
 
 	return nil, err

pkg/pulsar/functions.go

@@ -128,6 +128,9 @@ type Functions interface {
 	// File: file:/dir/fileName.jar
 	// Http: http://www.repo.com/fileName.jar
 	UpdateFunctionWithURL(functionConfig *utils.FunctionConfig, pkgURL string, updateOptions *utils.UpdateOptions) error
+
+	// Upload function to Pulsar
+	Upload(sourceFile, path string) error
 }
 
 type functions struct {
@@ -282,35 +285,41 @@ func (f *functions) DownloadFunction(path, destinationFile string) error {
 	endpoint := f.pulsar.endpoint(f.basePath, "download")
 	_, err := os.Open(destinationFile)
 	if err != nil {
-		_, err = os.Create(destinationFile)
-		if err != nil {
-			return err
+		if !os.IsNotExist(err) {
+			return fmt.Errorf("file %s already exists, please delete "+
+				"the file first or change the file name", destinationFile)
 		}
 	}
+	file, err := os.Create(destinationFile)
+	if err != nil {
+		return err
+	}
 
 	tmpMap := make(map[string]string)
 	tmpMap["path"] = path
 
-	_, err = f.pulsar.Client.GetWithQueryParams(endpoint, nil, tmpMap, false)
-
+	_, err = f.pulsar.Client.GetWithOptions(endpoint, nil, tmpMap, false, file)
 	if err != nil {
 		return err
 	}
-
 	return nil
 }
 
 func (f *functions) DownloadFunctionByNs(destinationFile, tenant, namespace, function string) error {
 	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, function, "download")
 	_, err := os.Open(destinationFile)
 	if err != nil {
-		_, err = os.Create(destinationFile)
-		if err != nil {
-			return err
+		if !os.IsNotExist(err) {
+			return fmt.Errorf("file %s already exists, please delete "+
+				"the file first or change the file name", destinationFile)
 		}
 	}
+	file, err := os.Create(destinationFile)
+	if err != nil {
+		return err
+	}
 
-	err = f.pulsar.Client.Get(endpoint, nil)
+	_, err = f.pulsar.Client.GetWithOptions(endpoint, nil, nil, false, file)
 	if err != nil {
 		return err
 	}
@@ -645,3 +654,30 @@ func (f *functions) TriggerFunction(tenant, namespace, name, topic, triggerValue
 
 	return str, nil
 }
+
+func (f *functions) Upload(sourceFile, path string) error {
+	if strings.TrimSpace(sourceFile) == "" && strings.TrimSpace(path) == "" {
+		return fmt.Errorf("source file or path is empty")
+	}
+	file, err := os.Open(sourceFile)
+	if err != nil {
+		return err
+	}
+	endpoint := f.pulsar.endpoint(f.basePath, "upload")
+	var b bytes.Buffer
+	w := multipart.NewWriter(&b)
+	writer, err := w.CreateFormFile("data", file.Name())
+	if err != nil {
+		return err
+	}
+	_, err = io.Copy(writer, file)
+	if err != nil {
+		return err
+	}
+	w.WriteField("path", path)
+	err = w.Close()
+	if err != nil {
+		return err
+	}
+	return f.pulsar.Client.PostWithMultiPart(endpoint, nil, &b, w.FormDataContentType())
+}