add retry options config

dara/core.go

@@ -34,7 +34,6 @@ type RuntimeOptions = util.RuntimeOptions
 type ExtendsParameters = util.ExtendsParameters
 
 var debugLog = debug.Init("dara")
-
 type HttpRequest interface {
 }
 
@@ -184,39 +183,6 @@ func Convert(in interface{}, out interface{}) error {
 	return err
 }
 
-// ConvertChan converts the source data to the target type and sends it to the specified channel.
-// @param src - source data
-// @param destChan - target channel
-// @return error - error during the conversion process
-func ConvertChan(src interface{}, destChan interface{}) error {
-	destChanValue := reflect.ValueOf(destChan)
-	if destChanValue.Kind() != reflect.Chan {
-		return fmt.Errorf("destChan must be a channel")
-	}
-
-	if destChanValue.Type().ChanDir() == reflect.SendDir {
-		return fmt.Errorf("destChan must be a receive or bidirectional channel")
-	}
-
-	elemType := destChanValue.Type().Elem()
-
-	destValue := reflect.New(elemType).Interface()
-
-	err := Convert(src, destValue)
-	if err != nil {
-		return err
-	}
-	destValueElem := reflect.ValueOf(destValue).Elem()
-
-	defer func() {
-		if r := recover(); r != nil {
-		}
-	}()
-
-	destChanValue.TrySend(destValueElem)
-	return nil
-}
-
 // Recover is used to format error
 func Recover(in interface{}) error {
 	if in == nil {
@@ -375,131 +341,6 @@ func DoRequest(request *Request, runtimeObject *RuntimeObject) (response *Respon
 	return
 }
 
-func DoRequestWithCtx(ctx context.Context, request *Request, runtimeObject *RuntimeObject) (response *Response, err error) {
-	if runtimeObject == nil {
-		runtimeObject = &RuntimeObject{}
-	}
-	fieldMap := make(map[string]string)
-	utils.InitLogMsg(fieldMap)
-	defer func() {
-		if runtimeObject.Logger != nil {
-			runtimeObject.Logger.PrintLog(fieldMap, err)
-		}
-	}()
-	if request.Method == nil {
-		request.Method = String("GET")
-	}
-
-	if request.Protocol == nil {
-		request.Protocol = String("http")
-	} else {
-		request.Protocol = String(strings.ToLower(StringValue(request.Protocol)))
-	}
-
-	requestURL := ""
-	request.Domain = request.Headers["host"]
-	if request.Port != nil {
-		request.Domain = String(fmt.Sprintf("%s:%d", StringValue(request.Domain), IntValue(request.Port)))
-	}
-	requestURL = fmt.Sprintf("%s://%s%s", StringValue(request.Protocol), StringValue(request.Domain), StringValue(request.Pathname))
-	queryParams := request.Query
-	// sort QueryParams by key
-	q := url.Values{}
-	for key, value := range queryParams {
-		q.Add(key, StringValue(value))
-	}
-	querystring := q.Encode()
-	if len(querystring) > 0 {
-		if strings.Contains(requestURL, "?") {
-			requestURL = fmt.Sprintf("%s&%s", requestURL, querystring)
-		} else {
-			requestURL = fmt.Sprintf("%s?%s", requestURL, querystring)
-		}
-	}
-	debugLog("> %s %s", StringValue(request.Method), requestURL)
-
-	httpRequest, err := http.NewRequestWithContext(ctx, StringValue(request.Method), requestURL, request.Body)
-	if err != nil {
-		return
-	}
-	httpRequest.Host = StringValue(request.Domain)
-
-	var client HttpClient
-	if runtimeObject.HttpClient == nil {
-		client = getDaraClient(runtimeObject.getClientTag(StringValue(request.Domain)))
-	} else {
-		client = runtimeObject.HttpClient
-	}
-
-	trans, err := getHttpTransport(request, runtimeObject)
-	if err != nil {
-		return
-	}
-	if defaultClient, ok := client.(*daraClient); ok {
-		defaultClient.Lock()
-		if !defaultClient.ifInit || defaultClient.httpClient.Transport == nil {
-			defaultClient.httpClient.Transport = trans
-		}
-		defaultClient.httpClient.Timeout = time.Duration(IntValue(runtimeObject.ReadTimeout)) * time.Millisecond
-		defaultClient.ifInit = true
-		defaultClient.Unlock()
-	}
-
-	for key, value := range request.Headers {
-		if value == nil || key == "content-length" {
-			continue
-		} else if key == "host" {
-			httpRequest.Header["Host"] = []string{*value}
-			delete(httpRequest.Header, "host")
-		} else if key == "user-agent" {
-			httpRequest.Header["User-Agent"] = []string{*value}
-			delete(httpRequest.Header, "user-agent")
-		} else {
-			httpRequest.Header[key] = []string{*value}
-		}
-		debugLog("> %s: %s", key, StringValue(value))
-	}
-	contentlength, _ := strconv.Atoi(StringValue(request.Headers["content-length"]))
-	event := utils.NewProgressEvent(utils.TransferStartedEvent, 0, int64(contentlength), 0)
-	utils.PublishProgress(runtimeObject.Listener, event)
-
-	putMsgToMap(fieldMap, httpRequest)
-	startTime := time.Now()
-	fieldMap["{start_time}"] = startTime.Format("2006-01-02 15:04:05")
-	res, err := hookDo(client.Call)(httpRequest, trans)
-	fieldMap["{cost}"] = time.Since(startTime).String()
-	completedBytes := int64(0)
-	if runtimeObject.Tracker != nil {
-		completedBytes = runtimeObject.Tracker.CompletedBytes
-	}
-	if err != nil {
-		select {
-		case <-ctx.Done():
-			err = TeaSDKError(ctx.Err())
-		default:
-		}
-
-		event = utils.NewProgressEvent(utils.TransferFailedEvent, completedBytes, int64(contentlength), 0)
-		utils.PublishProgress(runtimeObject.Listener, event)
-		return
-	}
-
-	event = utils.NewProgressEvent(utils.TransferCompletedEvent, completedBytes, int64(contentlength), 0)
-	utils.PublishProgress(runtimeObject.Listener, event)
-
-	response = NewResponse(res)
-	fieldMap["{code}"] = strconv.Itoa(res.StatusCode)
-	fieldMap["{res_headers}"] = Stringify(res.Header)
-	debugLog("< HTTP/1.1 %s", res.Status)
-	for key, value := range res.Header {
-		debugLog("< %s: %s", key, strings.Join(value, ""))
-		if len(value) != 0 {
-			response.Headers[strings.ToLower(key)] = String(value[0])
-		}
-	}
-	return
-}
-
 func getHttpTransport(req *Request, runtime *RuntimeObject) (*http.Transport, error) {
 	trans := new(http.Transport)
 	httpProxy, err := getHttpProxy(StringValue(req.Protocol), StringValue(req.Domain), runtime)
@@ -801,23 +642,8 @@ func isNil(a interface{}) bool {
 	return vi.IsNil()
 }
 
-func isNilOrZero(value interface{}) bool {
-	if value == nil {
-		return true
-	}
-
-	v := reflect.ValueOf(value)
-	switch v.Kind() {
-	case reflect.Chan, reflect.Func, reflect.Map, reflect.Ptr, reflect.Interface, reflect.Slice:
-		return v.IsNil()
-	default:
-		// Check for zero value
-		return reflect.DeepEqual(value, reflect.Zero(v.Type()).Interface())
-	}
-}
-
 func Default(inputValue interface{}, defaultValue interface{}) (_result interface{}) {
-	if isNilOrZero(inputValue) {
+	if IsNil(inputValue) {
 		_result = defaultValue
 		return _result
 	}
@@ -888,17 +714,10 @@ func ToMap(args ...interface{}) map[string]interface{} {
 			}
 		default:
 			val := reflect.ValueOf(obj)
-			if val.Kind().String() == "map" {
-				tmp := val.MapKeys()
-				for _, key := range tmp {
-					finalArg[key.String()] = val.MapIndex(key).Interface()
-				}
-			} else {
-				res := structToMap(val)
-				for key, value := range res {
-					if value != nil {
-						finalArg[key] = value
-					}
+			res := structToMap(val)
+			for key, value := range res {
+				if value != nil {
+					finalArg[key] = value
 				}
 			}
 		}

dara/json_parser.go

@@ -6,7 +6,6 @@ import (
 	jsoniter "github.com/json-iterator/go"
 	"github.com/modern-go/reflect2"
 	"io"
-	"io/ioutil"
 	"math"
 	"reflect"
 	"strconv"
@@ -333,28 +332,9 @@ func (decoder *nullableFuzzyFloat64Decoder) Decode(ptr unsafe.Pointer, iter *jso
 	}
 }
 
-func Stringify(a interface{}) string {
-	switch v := a.(type) {
-	case *string:
-		return StringValue(v)
-	case string:
-		return v
-	case []byte:
-		return string(v)
-	case io.Reader:
-		byt, err := ioutil.ReadAll(v)
-		if err != nil {
-			return ""
-		}
-		return string(byt)
-	}
-	byt := bytes.NewBuffer([]byte{})
-	jsonEncoder := json.NewEncoder(byt)
-	jsonEncoder.SetEscapeHTML(false)
-	if err := jsonEncoder.Encode(a); err != nil {
-		return ""
-	}
-	return string(bytes.TrimSpace(byt.Bytes()))
+func Stringify(m interface{}) string {
+	byt, _ := json.Marshal(m)
+	return string(byt)
 }
 
 func ParseJSON(a string) interface{} {

dara/json_parser_test.go

@@ -3,7 +3,6 @@ package dara
 import (
 	"reflect"
 	"testing"
-	"strings"
 
 	"github.com/alibabacloud-go/tea/utils"
 	jsoniter "github.com/json-iterator/go"
@@ -879,42 +878,3 @@ func TestUnmarshalWithDefaultDecoders(t *testing.T) {
 	err = jsoniter.Unmarshal(from, toUint64)
 	utils.AssertNotNil(t, err)
 }
-
-func Test_Stringify(t *testing.T) {
-	// interface
-	str := Stringify(map[string]interface{}{"test": "ok"})
-	utils.AssertEqual(t, `{"test":"ok"}`, str)
-	// string
-	str = Stringify("test")
-	utils.AssertEqual(t, "test", str)
-	// []byte
-	str = Stringify([]byte("test"))
-	utils.AssertEqual(t, "test", str)
-	// io.Reader
-	str = Stringify(strings.NewReader("test"))
-	utils.AssertEqual(t, "test", str)
-
-	str = Stringify("test")
-	utils.AssertEqual(t, "test", str)
-}
-
-
-func Test_ParseJSON(t *testing.T) {
-	obj := ParseJSON(`{"test":"ok"}`).(map[string]interface{})
-	utils.AssertEqual(t, "ok", obj["test"])
-
-	obj1 := ParseJSON(`["test1", "test2", "test3"]`).([]interface{})
-	utils.AssertEqual(t, "test2", obj1[1])
-
-	num := ParseJSON(`10`).(int)
-	utils.AssertEqual(t, 10, num)
-
-	boolVal := ParseJSON(`true`).(bool)
-	utils.AssertEqual(t, true, boolVal)
-
-	float64Val := ParseJSON(`1.00`).(float64)
-	utils.AssertEqual(t, 1.00, float64Val)
-
-	null := ParseJSON(`}}}}`)
-	utils.AssertEqual(t, nil, null)
-}

dara/stream.go

@@ -3,63 +3,44 @@ package dara
 import (
 	"bufio"
 	"bytes"
-	"context"
 	"encoding/json"
-	"fmt"
 	"io"
 	"io/ioutil"
 	"strings"
+	"fmt"
 )
 
+// 定义 Event 结构体
 type SSEEvent struct {
-	Id    *string
+	ID    *string
 	Event *string
 	Data  *string
 	Retry *int
 }
 
+// 解析单个事件
 func parseEvent(lines []string) *SSEEvent {
 	event := &SSEEvent{}
 	for _, line := range lines {
-		if strings.HasPrefix(line, "data:") {
-			var data string
-			if strings.HasPrefix(line, "data: ") {
-				data = strings.TrimPrefix(line, "data: ") + "\n"
-			} else {
-				data = strings.TrimPrefix(line, "data:") + "\n"
-			}
+		if strings.HasPrefix(line, "data: ") {
+			data := strings.TrimPrefix(line, "data: ") + "\n"
 			if event.Data == nil {
 				event.Data = new(string)
 			}
 			*event.Data += data
-		} else if strings.HasPrefix(line, "event:") {
-			var eventName string
-			if strings.HasPrefix(line, "event: ") {
-				eventName = strings.TrimPrefix(line, "event: ")
-			} else {
-				eventName = strings.TrimPrefix(line, "event:")
-			}
+		} else if strings.HasPrefix(line, "event: ") {
+			eventName := strings.TrimPrefix(line, "event: ")
 			event.Event = &eventName
-		} else if strings.HasPrefix(line, "id:") {
-			var id string
-			if strings.HasPrefix(line, "id: ") {
-				id = strings.TrimPrefix(line, "id: ")
-			} else {
-				id = strings.TrimPrefix(line, "id:")
-			}
-			event.Id = &id
-		} else if strings.HasPrefix(line, "retry:") {
-			var retryStr string
-			if strings.HasPrefix(line, "retry: ") {
-				retryStr = strings.TrimPrefix(line, "retry: ")
-			} else {
-				retryStr = strings.TrimPrefix(line, "retry:")
-			}
+		} else if strings.HasPrefix(line, "id: ") {
+			id := strings.TrimPrefix(line, "id: ")
+			event.ID = &id
+		} else if strings.HasPrefix(line, "retry: ") {
 			var retry int
-			fmt.Sscanf(retryStr, "%d", &retry)
+			fmt.Sscanf(strings.TrimPrefix(line, "retry: "), "%d", &retry)
 			event.Retry = &retry
 		}
 	}
+	// Remove last newline from data
 	if event.Data != nil {
 		data := strings.TrimRight(*event.Data, "\n")
 		event.Data = &data
@@ -151,65 +132,6 @@ func ReadAsSSE(body io.ReadCloser, eventChannel chan *SSEEvent, errorChannel cha
 			eventLines = append(eventLines, line)
 		}
 	}()
-}
-
-func ReadAsSSEWithContext(ctx context.Context, body io.ReadCloser, eventChannel chan *SSEEvent, errorChannel chan error) {
-
-	go func() {
-		defer func() {
-			body.Close()
-			close(eventChannel)
-		}()
-
-		reader := bufio.NewReader(body)
-		var eventLines []string
-
-		for {
-			select {
-			case <-ctx.Done():
-				errorChannel <- ctx.Err()
-				return
-			default:
-			}
-
-			line, err := reader.ReadString('\n')
-			if err != nil {
-				if err == io.EOF {
-					// Handle the end of the stream and possibly pending event
-					if len(eventLines) > 0 {
-						event := parseEvent(eventLines)
-						select {
-						case eventChannel <- event:
-						case <-ctx.Done():
-							errorChannel <- ctx.Err()
-							return
-						}
-					}
-					errorChannel <- nil
-					return
-				}
-				errorChannel <- err
-				return
-			}
-
-			line = strings.TrimRight(line, "\n")
-
-			if line == "" {
-				// End of an SSE event
-				if len(eventLines) > 0 {
-					event := parseEvent(eventLines)
-					select {
-					case eventChannel <- event:
-					case <-ctx.Done():
-						errorChannel <- ctx.Err()
-						return
-					}
-					eventLines = []string{} // Reset for the next event
-				}
-				continue
-			}
-
-			eventLines = append(eventLines, line)
-		}
-	}()
+
+	return
 }