Compare commits

...

11 Commits

Author SHA1 Message Date
yndu13 75efc7f189 fix: improve DNS robustness and refine timeout logic
- Fix DNS resolution failure by prioritizing CGO resolver (PreferGo: false)
  to stay consistent with system tools like curl when /etc/resolv.conf
  contains invalid nameservers.
- Fix logic bug in getLocalAddr where LocalAddr string was incorrectly
  cast to []byte instead of being parsed via net.ParseIP.
- Refactor HTTP timeout logic:
    * Set httpClient.Timeout to ConnectTimeout + ReadTimeout.
    * Set http.Transport.ResponseHeaderTimeout to ReadTimeout for
      granular control.
- Modernize net.Dialer by removing deprecated DualStack field (now
  enabled by default in Go 1.12+).
- Add unit tests for LocalAddr parsing, dial context, and timeout
  configuration in both tea and dara packages.
2026-01-06 20:52:19 +08:00
weeping af5f206bd8 fix socks5 proxy 2025-12-08 15:28:43 +08:00
weeping 5c6eb77b42 support IdleTimeout 2025-10-13 11:43:22 +08:00
weeping 0f6d084c67 fix validate 2025-09-11 18:13:59 +08:00
weeping df04276c08 add ReadAsSSEWithContext 2025-09-11 17:00:10 +08:00
weeping 622f315135 fix sse 2025-08-19 16:58:21 +08:00
peze 31de98dc53 add context 2025-07-23 20:05:57 +08:00
peze f93724f865 fix the stringify 2025-04-21 18:18:26 +08:00
peze fdbb29ef15 fix the toMap when input is basic map 2025-04-07 17:32:23 +08:00
peze a4d8c8f7f5 fix default condition 2025-04-07 17:32:23 +08:00
peze 4820a881e8 add retry options config 2025-04-07 17:32:23 +08:00
10 changed files with 2061 additions and 58 deletions
+209 -22
View File
File diff suppressed because it is too large Load Diff
+352 -1
View File
File diff suppressed because it is too large Load Diff
+23 -3
View File
@@ -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{} {
+40
View File
@@ -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)
}
+308
View File
File diff suppressed because it is too large Load Diff
+485
View File
File diff suppressed because it is too large Load Diff
+94 -16
View File
@@ -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)
}
}()
}
+464
View File
File diff suppressed because it is too large Load Diff
+15 -16
View File
@@ -386,7 +386,7 @@ func DoRequest(request *Request, requestRuntime map[string]interface{}) (respons
if !defaultClient.ifInit || defaultClient.httpClient.Transport == nil {
defaultClient.httpClient.Transport = trans
}
defaultClient.httpClient.Timeout = time.Duration(IntValue(runtimeObject.ReadTimeout)) * time.Millisecond
defaultClient.httpClient.Timeout = time.Duration(IntValue(runtimeObject.ConnectTimeout)+IntValue(runtimeObject.ReadTimeout)) * time.Millisecond
defaultClient.ifInit = true
defaultClient.Unlock()
}
@@ -442,6 +442,7 @@ func DoRequest(request *Request, requestRuntime map[string]interface{}) (respons
func getHttpTransport(req *Request, runtime *RuntimeObject) (*http.Transport, error) {
trans := new(http.Transport)
trans.ResponseHeaderTimeout = time.Duration(IntValue(runtime.ReadTimeout)) * time.Millisecond
httpProxy, err := getHttpProxy(StringValue(req.Protocol), StringValue(req.Domain), runtime)
if err != nil {
return nil, err
@@ -495,7 +496,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,
@@ -602,7 +603,7 @@ func getSocks5Proxy(runtime *RuntimeObject) (proxy *url.URL, err error) {
func getLocalAddr(localAddr string) (addr *net.TCPAddr) {
if localAddr != "" {
addr = &net.TCPAddr{
IP: []byte(localAddr),
IP: net.ParseIP(localAddr),
}
}
return addr
@@ -610,20 +611,18 @@ func getLocalAddr(localAddr string) (addr *net.TCPAddr) {
func setDialContext(runtime *RuntimeObject) func(cxt context.Context, net, addr string) (c net.Conn, err error) {
return func(ctx context.Context, network, address string) (net.Conn, error) {
if runtime.LocalAddr != nil && StringValue(runtime.LocalAddr) != "" {
netAddr := &net.TCPAddr{
IP: []byte(StringValue(runtime.LocalAddr)),
}
return (&net.Dialer{
Timeout: time.Duration(IntValue(runtime.ConnectTimeout)) * time.Second,
DualStack: true,
LocalAddr: netAddr,
}).DialContext(ctx, network, address)
}
return (&net.Dialer{
Timeout: time.Duration(IntValue(runtime.ConnectTimeout)) * time.Second,
timeout := time.Duration(IntValue(runtime.ConnectTimeout)) * time.Millisecond
dialer := &net.Dialer{
Timeout: timeout,
Resolver: &net.Resolver{
PreferGo: false,
},
DualStack: true,
}).DialContext(ctx, network, address)
}
if runtime.LocalAddr != nil && StringValue(runtime.LocalAddr) != "" {
dialer.LocalAddr = getLocalAddr(StringValue(runtime.LocalAddr))
}
return dialer.DialContext(ctx, network, address)
}
}
+71
View File
@@ -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"
@@ -942,3 +967,49 @@ func Test_TransInt32AndInt(t *testing.T) {
b := ToInt32(a)
utils.AssertEqual(t, Int32Value(b), int32(10))
}
func Test_getLocalAddr(t *testing.T) {
// Test empty address
addr := getLocalAddr("")
utils.AssertNil(t, addr)
// Test valid IPv4 address
addr = getLocalAddr("127.0.0.1")
utils.AssertNotNil(t, addr)
utils.AssertEqual(t, "127.0.0.1", addr.IP.String())
// Test valid IPv6 address
addr = getLocalAddr("::1")
utils.AssertNotNil(t, addr)
utils.AssertEqual(t, "::1", addr.IP.String())
// Test invalid IP
addr = getLocalAddr("invalid")
utils.AssertNotNil(t, addr)
utils.AssertNil(t, addr.IP)
}
func Test_setDialContext(t *testing.T) {
runtime := &RuntimeObject{
ConnectTimeout: Int(1000),
LocalAddr: String("127.0.0.1"),
}
dialerFunc := setDialContext(runtime)
utils.AssertNotNil(t, dialerFunc)
}
func Test_TimeoutLogic(t *testing.T) {
runtime := &RuntimeObject{
ConnectTimeout: Int(1000),
ReadTimeout: Int(2000),
}
req := &Request{
Protocol: String("http"),
Domain: String("localhost"),
}
trans, err := getHttpTransport(req, runtime)
utils.AssertNil(t, err)
utils.AssertNotNil(t, trans)
utils.AssertEqual(t, time.Duration(2000)*time.Millisecond, trans.ResponseHeaderTimeout)
}