fix validate

dara/core.go

@@ -34,6 +34,7 @@ type RuntimeOptions = util.RuntimeOptions
 type ExtendsParameters = util.ExtendsParameters
 
 var debugLog = debug.Init("dara")
+
 type HttpRequest interface {
 }
 
@@ -183,6 +184,39 @@ 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 {
@@ -341,6 +375,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)

dara/core_test.go

@@ -193,6 +193,150 @@ func TestConvertType(t *testing.T) {
 	utils.AssertEqual(t, "test", string(out.Body))
 }
 
+// TestConvertChan tests the ConvertChan function
+func TestConvertChan(t *testing.T) {
+	// Test case 1: Successful conversion and sending to channel
+	t.Run("SuccessfulConversion", func(t *testing.T) {
+		// Create a channel to receive the converted data
+		ch := make(chan test, 1)
+
+		// Source data to convert
+		src := map[string]interface{}{
+			"key":  123,
+			"body": []byte("test"),
+		}
+
+		// Perform conversion
+		err := ConvertChan(src, ch)
+
+		// Check for errors
+		if err != nil {
+			t.Errorf("Expected no error, got %v", err)
+		}
+		utils.AssertNil(t, err)
+
+		// Check if data was sent to channel
+		select {
+		case result := <-ch:
+			utils.AssertEqual(t, "123", result.Key)
+			utils.AssertEqual(t, "test", string(result.Body))
+		}
+	})
+
+	// Test case 2: Invalid destination channel (not a channel)
+	t.Run("InvalidDestChan", func(t *testing.T) {
+		src := map[string]interface{}{
+			"test": "data",
+		}
+
+		// Pass a non-channel type
+		err := ConvertChan(src, "not_a_channel")
+
+		if err == nil {
+			t.Error("Expected error for invalid channel, got nil")
+		}
+
+		expected := "destChan must be a channel"
+		if err.Error() != expected {
+			t.Errorf("Expected error message '%s', got '%s'", expected, err.Error())
+		}
+	})
+
+	// Test case 3: Send-only channel
+	t.Run("SendOnlyChannel", func(t *testing.T) {
+		// Create a send-only channel
+		ch := make(chan<- map[string]interface{}, 1)
+
+		src := map[string]interface{}{
+			"test": "data",
+		}
+
+		err := ConvertChan(src, ch)
+
+		if err == nil {
+			t.Error("Expected error for send-only channel, got nil")
+		}
+
+		expected := "destChan must be a receive or bidirectional channel"
+		if err.Error() != expected {
+			t.Errorf("Expected error message '%s', got '%s'", expected, err.Error())
+		}
+	})
+
+	// Test case 4: Conversion with struct
+	t.Run("StructConversion", func(t *testing.T) {
+		type TestStruct struct {
+			Name string `json:"name"`
+			Age  int    `json:"age"`
+		}
+
+		ch := make(chan TestStruct, 1)
+
+		// Source data matching struct fields
+		src := map[string]interface{}{
+			"name": "Alice",
+			"age":  30,
+		}
+
+		err := ConvertChan(src, ch)
+
+		if err != nil {
+			t.Errorf("Expected no error, got %v", err)
+		}
+
+		select {
+		case result := <-ch:
+			if result.Name != "Alice" || result.Age != 30 {
+				t.Errorf("Expected {Name: Alice, Age: 30}, got %v", result)
+			}
+		default:
+			t.Error("Expected data in channel, but channel is empty")
+		}
+	})
+
+	// Test case 5: Nil source data
+	t.Run("NilSource", func(t *testing.T) {
+		ch := make(chan map[string]interface{}, 1)
+
+		err := ConvertChan(nil, ch)
+
+		// Depending on implementation, this might error or succeed
+		// Here we're testing it doesn't panic
+		if err != nil {
+			// If there's an error, that's fine, just make sure it's handled
+			t.Logf("Nil source resulted in error (acceptable): %v", err)
+		} else {
+			// If no error, check what was sent
+			select {
+			case result := <-ch:
+				if result == nil {
+					t.Log("Nil source correctly handled")
+				} else {
+					t.Errorf("Expected nil or error, got %v", result)
+				}
+			default:
+				t.Log("Channel empty after nil source, which is acceptable")
+			}
+		}
+	})
+}
+
+// BenchmarkConvertChan benchmarks the ConvertChan function
+func BenchmarkConvertChan(b *testing.B) {
+	ch := make(chan map[string]interface{}, 1)
+	src := map[string]interface{}{
+		"name":   "benchmark_test",
+		"value":  12345,
+		"active": true,
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_ = ConvertChan(src, ch)
+		<-ch // Clear the channel
+	}
+}
+
 func TestRuntimeObject(t *testing.T) {
 	runtimeobject := NewRuntimeObject(nil)
 	utils.AssertNil(t, runtimeobject.IgnoreSSL)
@@ -362,7 +506,7 @@ type Test struct {
 
 func TestToMap(t *testing.T) {
 	inStr := map[string]string{
-		"tea": "test",
+		"tea":  "test",
 		"test": "test2",
 	}
 	result := ToMap(inStr)
@@ -370,7 +514,7 @@ func TestToMap(t *testing.T) {
 	utils.AssertEqual(t, "test2", result["test"])
 
 	inInt := map[string]int{
-		"tea": 12,
+		"tea":  12,
 		"test": 13,
 	}
 	result = ToMap(inInt)
@@ -514,6 +658,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
@@ -653,6 +798,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 }()
@@ -684,6 +864,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")

dara/stream.go

@@ -3,44 +3,63 @@ 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: ") {
-			data := strings.TrimPrefix(line, "data: ") + "\n"
+		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 event.Data == nil {
 				event.Data = new(string)
 			}
 			*event.Data += data
-		} else if strings.HasPrefix(line, "event: ") {
-			eventName := strings.TrimPrefix(line, "event: ")
+		} 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:")
+			}
 			event.Event = &eventName
-		} else if strings.HasPrefix(line, "id: ") {
-			id := strings.TrimPrefix(line, "id: ")
-			event.ID = &id
-		} else if strings.HasPrefix(line, "retry: ") {
+		} 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:")
+			}
 			var retry int
-			fmt.Sscanf(strings.TrimPrefix(line, "retry: "), "%d", &retry)
+			fmt.Sscanf(retryStr, "%d", &retry)
 			event.Retry = &retry
 		}
 	}
-	// Remove last newline from data
 	if event.Data != nil {
 		data := strings.TrimRight(*event.Data, "\n")
 		event.Data = &data
@@ -132,6 +151,65 @@ func ReadAsSSE(body io.ReadCloser, eventChannel chan *SSEEvent, errorChannel cha
 			eventLines = append(eventLines, line)
 		}
 	}()
-
-	return
+}
+
+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)
+		}
+	}()
 }