fix socks5 proxy

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)
+}

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)
+		}
+	}()
 }

internal/darafilev2/Darafile

@@ -0,0 +1,35 @@
+{
+  "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

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

tea/tea.go

@@ -495,7 +495,7 @@ func getHttpTransport(req *Request, runtime *RuntimeObject) (*http.Transport, er
 					Password: password,
 				}
 			}
-			dialer, err := proxy.SOCKS5(strings.ToLower(StringValue(runtime.Socks5NetWork)), socks5Proxy.String(), auth,
+			dialer, err := proxy.SOCKS5(strings.ToLower(StringValue(runtime.Socks5NetWork)), socks5Proxy.Host, auth,
 				&net.Dialer{
 					Timeout:   time.Duration(IntValue(runtime.ConnectTimeout)) * time.Millisecond,
 					DualStack: true,

tea/tea_test.go

@@ -560,12 +560,37 @@ func Test_DoRequest(t *testing.T) {
 			return mockResponse(200, ``, nil)
 		}
 	}
+	// Test socks5 proxy with user:password@host format
 	runtimeObj["socks5Proxy"] = "socks5://someuser:somepassword@ecs.aliyun.com"
 	runtimeObj["localAddr"] = "127.0.0.1"
 	resp, err = DoRequest(request, runtimeObj)
 	utils.AssertNil(t, err)
 	utils.AssertEqual(t, "test", StringValue(resp.Headers["tea"]))
 
+	// Test socks5 proxy with explicit port
+	runtimeObj["socks5Proxy"] = "socks5://someuser:somepassword@ecs.aliyun.com:1080"
+	resp, err = DoRequest(request, runtimeObj)
+	utils.AssertNil(t, err)
+	utils.AssertEqual(t, "test", StringValue(resp.Headers["tea"]))
+
+	// Test socks5 proxy without authentication
+	runtimeObj["socks5Proxy"] = "socks5://proxy.example.com:1080"
+	resp, err = DoRequest(request, runtimeObj)
+	utils.AssertNil(t, err)
+	utils.AssertEqual(t, "test", StringValue(resp.Headers["tea"]))
+
+	// Test socks5 proxy with IPv6 address
+	runtimeObj["socks5Proxy"] = "socks5://[::1]:1080"
+	resp, err = DoRequest(request, runtimeObj)
+	utils.AssertNil(t, err)
+	utils.AssertEqual(t, "test", StringValue(resp.Headers["tea"]))
+
+	// Test socks5 proxy with IPv6 and authentication
+	runtimeObj["socks5Proxy"] = "socks5://user:pass@[2001:db8::1]:1080"
+	resp, err = DoRequest(request, runtimeObj)
+	utils.AssertNil(t, err)
+	utils.AssertEqual(t, "test", StringValue(resp.Headers["tea"]))
+
 	runtimeObj["key"] = "private rsa key"
 	runtimeObj["cert"] = "private certification"
 	runtimeObj["ca"] = "private ca"