Merge branch 'master' into fix-resp-err

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
 }

internal/darafile/Darafile

@@ -1,35 +0,0 @@
-{
-  "scope": "darabonba",
-  "name": "HttpClient",
-  "version": "0.0.1",
-  "main": "./main.dara",
-  "maintainers": [
-    {
-      "name": "Alibaba Cloud SDK",
-      "email": "sdk-team@alibabacloud.com"
-    }
-  ],
-  "libraries": {
-  },
-  "releases": {
-    "go": "github.com/alibabacloud-go/tea/tea:v1.2.3-0.20240605082020-e6e537a31150"
-  },
-  "go": {
-    "interface": true,
-    "clientName": "HttpClient",
-    "typedef": {
-      "HttpRequest": {
-        "import": "net/http",
-        "type": "http.Request"
-      },
-      "HttpResponse": {
-        "import": "net/http",
-        "type": "http.Response"
-      },
-      "HttpTransport": {
-        "import": "net/http",
-        "type": "http.Transport"
-      }
-    }
-  }
-}

internal/darafile/main.dara

@@ -1,8 +0,0 @@
-typedef HttpRequest;
-typedef HttpResponse;
-typedef HttpTransport;
-
-init(){
-}
-
-async function call(request: HttpRequest, transport: HttpTransport): HttpResponse;

internal/darafilev2/Darafile

@@ -1,35 +0,0 @@
-{
-  "scope": "darabonba",
-  "name": "HttpClientV2",
-  "version": "0.0.1",
-  "main": "./main.dara",
-  "maintainers": [
-    {
-      "name": "Alibaba Cloud SDK",
-      "email": "sdk-team@alibabacloud.com"
-    }
-  ],
-  "libraries": {
-  },
-  "releases": {
-    "go": "github.com/alibabacloud-go/tea/dara:v1.3.5"
-  },
-  "go": {
-    "interface": true,
-    "clientName": "HttpClient",
-    "typedef": {
-      "HttpRequest": {
-        "import": "net/http",
-        "type": "http.Request"
-      },
-      "HttpResponse": {
-        "import": "net/http",
-        "type": "http.Response"
-      },
-      "HttpTransport": {
-        "import": "net/http",
-        "type": "http.Transport"
-      }
-    }
-  }
-}

internal/darafilev2/main.dara

@@ -1,8 +0,0 @@
-typedef HttpRequest;
-typedef HttpResponse;
-typedef HttpTransport;
-
-init(){
-}
-
-async function call(request: HttpRequest, transport: HttpTransport): HttpResponse;

tea/tea.go

@@ -31,28 +31,7 @@ import (
 
 var debugLog = debug.Init("tea")
 
-type HttpRequest interface {
-}
-
-type HttpResponse interface {
-}
-
-type HttpClient interface {
-	Call(request *http.Request, transport *http.Transport) (response *http.Response, err error)
-}
-
-type teaClient struct {
-	sync.Mutex
-	httpClient *http.Client
-	ifInit     bool
-}
-
-func (client *teaClient) Call(request *http.Request, transport *http.Transport) (response *http.Response, err error) {
-	response, err = client.httpClient.Do(request)
-	return
-}
-
-var hookDo = func(fn func(req *http.Request, transport *http.Transport) (*http.Response, error)) func(req *http.Request, transport *http.Transport) (*http.Response, error) {
+var hookDo = func(fn func(req *http.Request) (*http.Response, error)) func(req *http.Request) (*http.Response, error) {
 	return fn
 }
 
@@ -118,7 +97,12 @@ type RuntimeObject struct {
 	Listener       utils.ProgressListener `json:"listener" xml:"listener"`
 	Tracker        *utils.ReaderTracker   `json:"tracker" xml:"tracker"`
 	Logger         *utils.Logger          `json:"logger" xml:"logger"`
-	HttpClient
+}
+
+type teaClient struct {
+	sync.Mutex
+	httpClient *http.Client
+	ifInit     bool
 }
 
 var clientPool = &sync.Map{}
@@ -159,9 +143,6 @@ func NewRuntimeObject(runtime map[string]interface{}) *RuntimeObject {
 	if runtime["logger"] != nil {
 		runtimeObject.Logger = runtime["logger"].(*utils.Logger)
 	}
-	if runtime["httpClient"] != nil {
-		runtimeObject.HttpClient = runtime["httpClient"].(HttpClient)
-	}
 	return runtimeObject
 }
 
@@ -370,27 +351,18 @@ func DoRequest(request *Request, requestRuntime map[string]interface{}) (respons
 	}
 	httpRequest.Host = StringValue(request.Domain)
 
-	var client HttpClient
-	if runtimeObject.HttpClient == nil {
-		client = getTeaClient(runtimeObject.getClientTag(StringValue(request.Domain)))
-	} else {
-		client = runtimeObject.HttpClient
-	}
-
-	trans, err := getHttpTransport(request, runtimeObject)
-	if err != nil {
-		return
-	}
-	if defaultClient, ok := client.(*teaClient); ok {
-		defaultClient.Lock()
-		if !defaultClient.ifInit || defaultClient.httpClient.Transport == nil {
-			defaultClient.httpClient.Transport = trans
+	client := getTeaClient(runtimeObject.getClientTag(StringValue(request.Domain)))
+	client.Lock()
+	if !client.ifInit {
+		trans, err := getHttpTransport(request, runtimeObject)
+		if err != nil {
+			return nil, err
 		}
-		defaultClient.httpClient.Timeout = time.Duration(IntValue(runtimeObject.ReadTimeout)) * time.Millisecond
-		defaultClient.ifInit = true
-		defaultClient.Unlock()
+		client.httpClient.Timeout = time.Duration(IntValue(runtimeObject.ReadTimeout)) * time.Millisecond
+		client.httpClient.Transport = trans
+		client.ifInit = true
 	}
-
+	client.Unlock()
 	for key, value := range request.Headers {
 		if value == nil || key == "content-length" {
 			continue
@@ -412,7 +384,7 @@ func DoRequest(request *Request, requestRuntime map[string]interface{}) (respons
 	putMsgToMap(fieldMap, httpRequest)
 	startTime := time.Now()
 	fieldMap["{start_time}"] = startTime.Format("2006-01-02 15:04:05")
-	res, err := hookDo(client.Call)(httpRequest, trans)
+	res, err := hookDo(client.httpClient.Do)(httpRequest)
 	fieldMap["{cost}"] = time.Since(startTime).String()
 	completedBytes := int64(0)
 	if runtimeObject.Tracker != nil {

tea/tea_test.go

@@ -487,36 +487,11 @@ func Test_GetBackoffTime(t *testing.T) {
 	utils.AssertEqual(t, true, IntValue(ms) <= 3)
 }
 
-type httpClient struct {
-	HttpClient
-	httpClient *http.Client
-}
-
-func newHttpClient() (*httpClient, error) {
-	client := new(httpClient)
-	err := client.Init()
-	return client, err
-}
-
-func (client *httpClient) Init() (_err error) {
-	return nil
-}
-
-func (client *httpClient) Call(request *http.Request, transport *http.Transport) (_result *http.Response, _err error) {
-	if transport != nil {
-		trans := transport.Clone()
-		client.httpClient.Transport = trans
-	} else {
-		client.httpClient.Transport = http.DefaultTransport.(*http.Transport).Clone()
-	}
-	return client.httpClient.Do(request)
-}
-
 func Test_DoRequest(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) {
+	hookDo = func(fn func(req *http.Request) (*http.Response, error)) func(req *http.Request) (*http.Response, error) {
+		return func(req *http.Request) (*http.Response, error) {
 			return mockResponse(200, ``, errors.New("Internal error"))
 		}
 	}
@@ -555,8 +530,8 @@ func Test_DoRequest(t *testing.T) {
 	utils.AssertNil(t, resp)
 	utils.AssertContains(t, err.Error(), ` invalid URL escape "%gf"`)
 
-	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) {
+	hookDo = func(fn func(req *http.Request) (*http.Response, error)) func(req *http.Request) (*http.Response, error) {
+		return func(req *http.Request) (*http.Response, error) {
 			return mockResponse(200, ``, nil)
 		}
 	}
@@ -604,8 +579,8 @@ func Test_DoRequest(t *testing.T) {
 	utils.AssertNil(t, err)
 	utils.AssertEqual(t, "test", StringValue(resp.Headers["tea"]))
 
-	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) {
+	hookDo = func(fn func(req *http.Request) (*http.Response, error)) func(req *http.Request) (*http.Response, error) {
+		return func(req *http.Request) (*http.Response, error) {
 			utils.AssertEqual(t, "tea-cn-hangzhou.aliyuncs.com:1080", req.Host)
 			return mockResponse(200, ``, errors.New("Internal error"))
 		}
@@ -617,20 +592,13 @@ func Test_DoRequest(t *testing.T) {
 	resp, err = DoRequest(request, runtimeObj)
 	utils.AssertNil(t, resp)
 	utils.AssertEqual(t, `Internal error`, err.Error())
-
-	httpClient, err := newHttpClient()
-	utils.AssertNil(t, err)
-	runtimeObj["httpClient"] = httpClient
-	resp, err = DoRequest(request, runtimeObj)
-	utils.AssertNil(t, resp)
-	utils.AssertEqual(t, `Internal error`, err.Error())
 }
 
 func Test_DoRequestWithConcurrent(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) {
+	hookDo = func(fn func(req *http.Request) (*http.Response, error)) func(req *http.Request) (*http.Response, error) {
+		return func(req *http.Request) (*http.Response, error) {
 			return mockResponse(200, ``, nil)
 		}
 	}
@@ -726,11 +694,11 @@ func Test_SetDialContext(t *testing.T) {
 }
 
 func Test_hookdo(t *testing.T) {
-	fn := func(req *http.Request, transport *http.Transport) (*http.Response, error) {
+	fn := func(req *http.Request) (*http.Response, error) {
 		return nil, errors.New("hookdo")
 	}
 	result := hookDo(fn)
-	resp, err := result(nil, nil)
+	resp, err := result(nil)
 	utils.AssertNil(t, resp)
 	utils.AssertEqual(t, "hookdo", err.Error())
 }