refactor(rest): convert to byte collector

internal/io/http/client.go

@@ -42,7 +42,6 @@
 type ClientConf struct {
 	config       *RawConf
 	client       *http.Client
-	compressor   message.Compressor   // compressor used to payload compression when specifies compressAlgorithm
 	decompressor message.Decompressor // decompressor used to payload decompression when specifies compressAlgorithm
 
 	// auth related
@@ -91,7 +90,7 @@
 	Code int `json:"code"`
 }
 
-var bodyTypeMap = map[string]string{"none": "", "text": "text/plain", "json": "application/json", "html": "text/html", "xml": "application/xml", "javascript": "application/javascript", "form": ""}
+var bodyTypeMap = map[string]string{"none": "", "text": "text/plain", "json": "application/json", "html": "text/html", "xml": "application/xml", "javascript": "application/javascript", "form": "", "binary": "application/octet-stream"}
 
 // newTransport allows EdgeX Foundry, protected by OpenZiti to override and obtain a transport
 // protected by OpenZiti's zero trust connectivity. See client_edgex.go where this function is
@@ -198,11 +197,6 @@
 	cc.config = c
 	// that means payload need compression and decompression, so we need initialize compressor and decompressor
 	if c.Compression != "" {
-		cc.compressor, err = compressor.GetCompressor(c.Compression)
-		if err != nil {
-			return fmt.Errorf("init payload compressor failed, %w", err)
-		}
-
 		cc.decompressor, err = compressor.GetDecompressor(c.Compression)
 		if err != nil {
 			return fmt.Errorf("init payload decompressor failed, %w", err)
@@ -222,11 +216,11 @@
 
 // initialize the oAuth access token
 func (cc *ClientConf) auth(ctx api.StreamContext) error {
-	resp, err := httpx.Send(conf.Log, cc.client, "json", http.MethodPost, cc.accessConf.Url, nil, true, cc.accessConf.Body)
+	resp, err := httpx.Send(conf.Log, cc.client, "json", http.MethodPost, cc.accessConf.Url, nil, cc.accessConf.Body)
 	if err != nil {
 		return err
 	}
-	tokens, _, err := cc.parseResponse(ctx, resp, "")
+	tokens, _, err := cc.parseResponse(ctx, resp, "", true, true)
 	if err != nil {
 		return err
 	}
@@ -274,11 +268,11 @@
 			}
 		}
 
-		resp, err := httpx.Send(conf.Log, cc.client, "json", http.MethodPost, cc.refreshConf.Url, headers, true, cc.refreshConf.Body)
+		resp, err := httpx.Send(conf.Log, cc.client, "json", http.MethodPost, cc.refreshConf.Url, headers, cc.refreshConf.Body)
 		if err != nil {
 			return fmt.Errorf("fail to get refresh token: %v", err)
 		}
-		nt, _, err := cc.parseResponse(ctx, resp, "")
+		nt, _, err := cc.parseResponse(ctx, resp, "", true, true)
 		if err != nil {
 			return fmt.Errorf("Cannot parse refresh token response to json: %v", err)
 		}
@@ -317,9 +311,11 @@
 }
 
 // parse the response status. For rest sink, it will not return the body by default if not need to debug
-func (cc *ClientConf) parseResponse(ctx api.StreamContext, resp *http.Response, lastMD5 string) ([]map[string]interface{}, string, error) {
+func (cc *ClientConf) parseResponse(ctx api.StreamContext, resp *http.Response, lastMD5 string, returnBody bool, skipDecompression bool) ([]map[string]interface{}, string, error) {
 	if resp.StatusCode < 200 || resp.StatusCode > 299 {
 		return nil, "", fmt.Errorf("%s: %d", CODE_ERR, resp.StatusCode)
+	} else if !returnBody { // For rest sink who only need to know if the request is successful
+		return nil, "", nil
 	}
 
 	c, err := io.ReadAll(resp.Body)
@@ -332,7 +328,7 @@
 	}()
 
 	newMD5 := ""
-	if cc.config.Incremental {
+	if returnBody && cc.config.Incremental {
 		newMD5 = getMD5Hash(c)
 		if newMD5 == lastMD5 {
 			return nil, newMD5, nil
@@ -341,18 +337,21 @@
 
 	switch cc.config.ResponseType {
 	case "code":
-		if cc.config.Compression != "" {
-			if c, err = cc.responseBodyDecompress(ctx, resp, c); err != nil {
-				return nil, "", fmt.Errorf("try to decompress payload failed, %w", err)
+		if returnBody {
+			if cc.config.Compression != "" && !skipDecompression {
+				if c, err = cc.responseBodyDecompress(ctx, resp, c); err != nil {
+					return nil, "", fmt.Errorf("try to decompress payload failed, %w", err)
+				}
 			}
+			m, e := decode(c)
+			if e != nil {
+				return nil, "", fmt.Errorf("%s: decode fail for %v", BODY_ERR, e)
+			}
+			return m, newMD5, e
 		}
-		m, e := decode(c)
-		if e != nil {
-			return nil, "", fmt.Errorf("%s: decode fail for %v", BODY_ERR, e)
-		}
-		return m, newMD5, e
+		return nil, "", nil
 	case "body":
-		if cc.config.Compression != "" {
+		if cc.config.Compression != "" && !skipDecompression {
 			if c, err = cc.responseBodyDecompress(ctx, resp, c); err != nil {
 				return nil, "", fmt.Errorf("try to decompress payload failed, %w", err)
 			}
@@ -371,7 +370,10 @@
 				return nil, "", fmt.Errorf("%s: %d", CODE_ERR, ro.Code)
 			}
 		}
-		return payloads, newMD5, nil
+		if returnBody {
+			return payloads, newMD5, nil
+		}
+		return nil, "", nil
 	default:
 		return nil, "", fmt.Errorf("%s: unsupported response type %s", BODY_ERR, cc.config.ResponseType)
 	}

internal/io/http/httppull_source.go

@@ -75,11 +75,11 @@ func doPull(ctx api.StreamContext, c *ClientConf, lastMD5 string) ([]map[string]
 	if err != nil {
 		return nil, "", err
 	}
-	resp, err := httpx.Send(ctx.GetLogger(), c.client, c.config.BodyType, c.config.Method, c.config.Url, headers, true, []byte(c.config.Body))
+	resp, err := httpx.Send(ctx.GetLogger(), c.client, c.config.BodyType, c.config.Method, c.config.Url, headers, []byte(c.config.Body))
 	if err != nil {
 		return nil, "", err
 	}
-	results, newMD5, err := c.parseResponse(ctx, resp, lastMD5)
+	results, newMD5, err := c.parseResponse(ctx, resp, lastMD5, true, false)
 	if err != nil {
 		return nil, "", err
 	}

internal/io/http/rest_sink.go

@@ -44,20 +44,13 @@
 	return nil
 }
 
-func (r *RestSink) Collect(ctx api.StreamContext, item api.MessageTuple) error {
-	return r.collect(ctx, item, item.ToMap())
-}
-
-func (r *RestSink) CollectList(ctx api.StreamContext, items api.MessageTupleList) error {
-	return r.collect(ctx, items, items.ToMaps())
-}
-
-func (r *RestSink) collect(ctx api.StreamContext, item any, data any) error {
+func (r *RestSink) Collect(ctx api.StreamContext, item api.RawTuple) error {
 	logger := ctx.GetLogger()
 	headers := r.config.Headers
 	bodyType := r.config.BodyType
 	method := r.config.Method
 	u := r.config.Url
+
 	if dp, ok := item.(api.HasDynamicProps); ok {
 		for k := range headers {
 			nv, ok := dp.DynamicProps(k)
@@ -78,7 +71,21 @@
 			u = nu
 		}
 	}
-	resp, err := httpx.Send(ctx.GetLogger(), r.client, bodyType, method, u, headers, r.config.SendSingle, data)
+
+	switch r.config.Compression {
+	case "zstd":
+		if headers == nil {
+			headers = make(map[string]string)
+		}
+		headers["Content-Encoding"] = "zstd"
+	case "gzip":
+		if headers == nil {
+			headers = make(map[string]string)
+		}
+		headers["Content-Encoding"] = "gzip"
+	}
+
+	resp, err := httpx.Send(ctx.GetLogger(), r.client, bodyType, method, u, headers, item.Raw())
 	failpoint.Inject("recoverAbleErr", func() {
 		err = errors.New("connection reset by peer")
 	})
@@ -91,15 +98,15 @@
 				originErr.Error(),
 				recoverAble,
 				method,
-				u, data))
+				u, string(item.Raw())))
 		}
 		return fmt.Errorf(`rest sink fails to send out the data:err=%s recoverAble=%v method=%s path="%s" request_body="%s"`,
 			originErr.Error(),
 			recoverAble,
-			method, u, data)
+			method, u, string(item.Raw()))
 	} else {
 		logger.Debugf("rest sink got response %v", resp)
-		_, b, err := r.parseResponse(ctx, resp, "")
+		_, b, err := r.parseResponse(ctx, resp, "", r.config.DebugResp, false)
 		// do not record response body error as it is not an error in the sink action.
 		if err != nil && !strings.HasPrefix(err.Error(), BODY_ERR) {
 			if strings.HasPrefix(err.Error(), BODY_ERR) {
@@ -125,4 +132,4 @@
 	return &RestSink{}
 }
 
-var _ api.TupleCollector = &RestSink{}
+var _ api.BytesCollector = &RestSink{}