add context

dara/core.go

@@ -34,6 +34,7 @@ type RuntimeOptions = util.RuntimeOptions
 type ExtendsParameters = util.ExtendsParameters
 
 var debugLog = debug.Init("dara")
+
 type HttpRequest interface {
 }
 
@@ -150,6 +151,9 @@ func NewRuntimeObject(runtime map[string]interface{}) *RuntimeObject {
 	if runtime["httpClient"] != nil {
 		runtimeObject.HttpClient = runtime["httpClient"].(HttpClient)
 	}
+	if runtime["retryOptions"] != nil {
+		runtimeObject.RetryOptions = runtime["retryOptions"].(*RetryOptions)
+	}
 	return runtimeObject
 }
 
@@ -338,6 +342,131 @@ 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)
@@ -639,8 +768,23 @@ 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 IsNil(inputValue) {
+	if isNilOrZero(inputValue) {
 		_result = defaultValue
 		return _result
 	}
@@ -711,10 +855,17 @@ func ToMap(args ...interface{}) map[string]interface{} {
 			}
 		default:
 			val := reflect.ValueOf(obj)
-			res := structToMap(val)
-			for key, value := range res {
-				if value != nil {
-					finalArg[key] = value
+			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
+					}
 				}
 			}
 		}

dara/core_test.go

@@ -30,6 +30,12 @@ var runtimeObj = map[string]interface{}{
 	"listener":      &Progresstest{},
 	"tracker":       &utils.ReaderTracker{CompletedBytes: int64(10)},
 	"logger":        utils.NewLogger("info", "", &bytes.Buffer{}, "{time}"),
+	"retryOptions": &RetryOptions{
+		Retryable: true,
+		RetryCondition: []*RetryCondition{
+			{MaxAttempts: 3, Exception: []string{"AErr"}, ErrorCode: []string{"A1Err"}},
+		},
+	},
 }
 
 var key = `-----BEGIN RSA PRIVATE KEY-----
@@ -193,6 +199,10 @@ func TestRuntimeObject(t *testing.T) {
 
 	runtimeobject = NewRuntimeObject(runtimeObj)
 	utils.AssertEqual(t, false, BoolValue(runtimeobject.IgnoreSSL))
+
+	utils.AssertEqual(t, true, runtimeobject.RetryOptions.Retryable)
+
+	utils.AssertEqual(t, 1, len(runtimeobject.RetryOptions.RetryCondition))
 }
 
 func TestSDKError(t *testing.T) {
@@ -351,11 +361,27 @@ type Test struct {
 }
 
 func TestToMap(t *testing.T) {
+	inStr := map[string]string{
+		"tea":  "test",
+		"test": "test2",
+	}
+	result := ToMap(inStr)
+	utils.AssertEqual(t, "test", result["tea"])
+	utils.AssertEqual(t, "test2", result["test"])
+
+	inInt := map[string]int{
+		"tea":  12,
+		"test": 13,
+	}
+	result = ToMap(inInt)
+	utils.AssertEqual(t, 12, result["tea"])
+	utils.AssertEqual(t, 13, result["test"])
+
 	in := map[string]*string{
 		"tea": String("test"),
 		"nil": nil,
 	}
-	result := ToMap(in)
+	result = ToMap(in)
 	utils.AssertEqual(t, "test", result["tea"])
 	utils.AssertNil(t, result["nil"])
 
@@ -488,6 +514,7 @@ func Test_GetBackoffTime(t *testing.T) {
 	ms = GetBackoffTime(backoff, Int(1))
 	utils.AssertEqual(t, true, IntValue(ms) <= 3)
 }
+
 type httpClient struct {
 	HttpClient
 	httpClient *http.Client
@@ -627,6 +654,41 @@ func Test_DoRequest(t *testing.T) {
 	utils.AssertEqual(t, `Internal error`, err.Error())
 }
 
+func Test_DoRequestWithContextCancellation(t *testing.T) {
+	origTestHookDo := hookDo
+	defer func() { hookDo = origTestHookDo }()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Millisecond)
+	defer cancel()
+
+	hookDo = func(fn func(req *http.Request, transport *http.Transport) (*http.Response, error)) func(req *http.Request, transport *http.Transport) (*http.Response, error) {
+		return func(req *http.Request, transport *http.Transport) (*http.Response, error) {
+			select {
+			case <-time.After(2 * time.Second):
+				return &http.Response{StatusCode: 200, Header: http.Header{}, Body: http.NoBody}, nil
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			}
+		}
+	}
+
+	request := &Request{
+		Method:   String("GET"),
+		Protocol: String("http"),
+		Headers:  map[string]*string{"host": String("ecs.cn-hangzhou.aliyuncs.com")},
+	}
+
+	runtimeObject := &RuntimeObject{}
+
+	resp, err := DoRequestWithCtx(ctx, request, runtimeObject)
+
+	utils.AssertNil(t, resp)
+	if err == nil {
+		t.Fatal("Expected an error due to context timeout, got nil")
+	}
+	utils.AssertContains(t, err.Error(), "context deadline exceeded")
+}
+
 func Test_DoRequestWithConcurrent(t *testing.T) {
 	origTestHookDo := hookDo
 	defer func() { hookDo = origTestHookDo }()
@@ -658,6 +720,53 @@ func Test_DoRequestWithConcurrent(t *testing.T) {
 	wg.Wait()
 }
 
+func Test_DoRequestWithConcurrentAndContext(t *testing.T) {
+	origTestHookDo := hookDo
+	defer func() { hookDo = origTestHookDo }()
+	hookDo = func(fn func(req *http.Request, transport *http.Transport) (*http.Response, error)) func(req *http.Request, transport *http.Transport) (*http.Response, error) {
+		return func(req *http.Request, transport *http.Transport) (*http.Response, error) {
+			time.Sleep(100 * time.Millisecond) // 模拟请求延迟
+			return mockResponse(200, ``, nil)
+		}
+	}
+
+	var wg sync.WaitGroup
+	for i := 0; i < 50; i++ {
+		wg.Add(1)
+		go func(readTimeout int) {
+			defer wg.Done()
+
+			// 每个 goroutine 有自己的上下文，设定超时控制在 500 毫秒
+			ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+			defer cancel()
+
+			runtime := NewRuntimeObject(map[string]interface{}{
+				"readTimeout": readTimeout,
+				"ctx":         ctx, // 将上下文集成到运行时对象中
+			})
+
+			for j := 0; j < 50; j++ {
+				wg.Add(1)
+				go func() {
+					defer wg.Done()
+
+					request := NewRequest()
+					resp, err := DoRequest(request, runtime)
+
+					if err != nil {
+						// 检查是否是由于上下文超时导致的错误
+						utils.AssertContains(t, err.Error(), "context deadline exceeded")
+					} else {
+						utils.AssertNil(t, err)
+						utils.AssertNotNil(t, resp)
+					}
+				}()
+			}
+		}(i)
+	}
+	wg.Wait()
+}
+
 func Test_getHttpProxy(t *testing.T) {
 	originHttpProxy := os.Getenv("HTTP_PROXY")
 	originHttpsProxy := os.Getenv("HTTPS_PROXY")
@@ -985,6 +1094,33 @@ func Test_Default(t *testing.T) {
 
 	b := ToInt32(a)
 	utils.AssertEqual(t, Int32Value(b), int32(10))
+
+	// Testing Default with nil values
+	if result := Default(nil, "default"); result != "default" {
+		t.Errorf("expected 'default', got '%v'", result)
+	}
+
+	// Testing Default with zero values
+	if result := Default("", "default"); result != "default" {
+		t.Errorf("expected 'default', got '%v'", result)
+	}
+
+	if result := Default(0, 42); result != 42 {
+		t.Errorf("expected 42, got %v", result)
+	}
+
+	if result := Default(false, true); result != true {
+		t.Errorf("expected true, got %v", result)
+	}
+
+	// Testing Default with non-zero values
+	if result := Default("value", "default"); result != "value" {
+		t.Errorf("expected 'value', got '%v'", result)
+	}
+
+	if result := Default([]int{1, 2, 3}, []int{}); !reflect.DeepEqual(result, []int{1, 2, 3}) {
+		t.Errorf("expected [1 2 3], got '%v'", result)
+	}
 }
 
 func TestToBytes(t *testing.T) {

dara/json_parser.go

@@ -6,6 +6,7 @@ import (
 	jsoniter "github.com/json-iterator/go"
 	"github.com/modern-go/reflect2"
 	"io"
+	"io/ioutil"
 	"math"
 	"reflect"
 	"strconv"
@@ -332,9 +333,28 @@ func (decoder *nullableFuzzyFloat64Decoder) Decode(ptr unsafe.Pointer, iter *jso
 	}
 }
 
-func Stringify(m interface{}) string {
-	byt, _ := json.Marshal(m)
-	return string(byt)
+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 ParseJSON(a string) interface{} {

dara/json_parser_test.go

@@ -3,6 +3,7 @@ package dara
 import (
 	"reflect"
 	"testing"
+	"strings"
 
 	"github.com/alibabacloud-go/tea/utils"
 	jsoniter "github.com/json-iterator/go"
@@ -878,3 +879,42 @@ 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)
+}