Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| cdfd54484c | |||
| af5f206bd8 | |||
| 5c6eb77b42 | |||
| 0f6d084c67 | |||
| df04276c08 |
+21
-18
@@ -94,6 +94,7 @@ type RuntimeObject struct {
|
||||
IgnoreSSL *bool `json:"ignoreSSL" xml:"ignoreSSL"`
|
||||
ReadTimeout *int `json:"readTimeout" xml:"readTimeout"`
|
||||
ConnectTimeout *int `json:"connectTimeout" xml:"connectTimeout"`
|
||||
IdleTimeout *int `json:"idleTimeout" xml:"idleTimeout"`
|
||||
LocalAddr *string `json:"localAddr" xml:"localAddr"`
|
||||
HttpProxy *string `json:"httpProxy" xml:"httpProxy"`
|
||||
HttpsProxy *string `json:"httpsProxy" xml:"httpsProxy"`
|
||||
@@ -114,7 +115,7 @@ type RuntimeObject struct {
|
||||
|
||||
func (r *RuntimeObject) getClientTag(domain string) string {
|
||||
return strconv.FormatBool(BoolValue(r.IgnoreSSL)) + strconv.Itoa(IntValue(r.ReadTimeout)) +
|
||||
strconv.Itoa(IntValue(r.ConnectTimeout)) + StringValue(r.LocalAddr) + StringValue(r.HttpProxy) +
|
||||
strconv.Itoa(IntValue(r.ConnectTimeout)) + strconv.Itoa(IntValue(r.IdleTimeout)) + StringValue(r.LocalAddr) + StringValue(r.HttpProxy) +
|
||||
StringValue(r.HttpsProxy) + StringValue(r.NoProxy) + StringValue(r.Socks5Proxy) + StringValue(r.Socks5NetWork) + domain
|
||||
}
|
||||
|
||||
@@ -128,6 +129,7 @@ func NewRuntimeObject(runtime map[string]interface{}) *RuntimeObject {
|
||||
IgnoreSSL: TransInterfaceToBool(runtime["ignoreSSL"]),
|
||||
ReadTimeout: TransInterfaceToInt(runtime["readTimeout"]),
|
||||
ConnectTimeout: TransInterfaceToInt(runtime["connectTimeout"]),
|
||||
IdleTimeout: TransInterfaceToInt(runtime["idleTimeout"]),
|
||||
LocalAddr: TransInterfaceToString(runtime["localAddr"]),
|
||||
HttpProxy: TransInterfaceToString(runtime["httpProxy"]),
|
||||
HttpsProxy: TransInterfaceToString(runtime["httpsProxy"]),
|
||||
@@ -321,7 +323,7 @@ func DoRequest(request *Request, runtimeObject *RuntimeObject) (response *Respon
|
||||
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()
|
||||
}
|
||||
@@ -440,7 +442,7 @@ func DoRequestWithCtx(ctx context.Context, request *Request, runtimeObject *Runt
|
||||
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()
|
||||
}
|
||||
@@ -502,6 +504,7 @@ func DoRequestWithCtx(ctx context.Context, request *Request, runtimeObject *Runt
|
||||
|
||||
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
|
||||
@@ -555,7 +558,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,
|
||||
@@ -573,6 +576,9 @@ func getHttpTransport(req *Request, runtime *RuntimeObject) (*http.Transport, er
|
||||
trans.MaxIdleConns = IntValue(runtime.MaxIdleConns)
|
||||
trans.MaxIdleConnsPerHost = IntValue(runtime.MaxIdleConns)
|
||||
}
|
||||
if runtime.IdleTimeout != nil && *runtime.IdleTimeout > 0 {
|
||||
trans.IdleConnTimeout = time.Duration(IntValue(runtime.IdleTimeout)) * time.Millisecond
|
||||
}
|
||||
return trans, nil
|
||||
}
|
||||
|
||||
@@ -691,7 +697,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
|
||||
@@ -699,20 +705,17 @@ 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)
|
||||
timeout := time.Duration(IntValue(runtime.ConnectTimeout)) * time.Millisecond
|
||||
dialer := &net.Dialer{
|
||||
Timeout: timeout,
|
||||
Resolver: &net.Resolver{
|
||||
PreferGo: false,
|
||||
},
|
||||
}
|
||||
return (&net.Dialer{
|
||||
Timeout: time.Duration(IntValue(runtime.ConnectTimeout)) * time.Second,
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -732,12 +732,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, NewRuntimeObject(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, NewRuntimeObject(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, NewRuntimeObject(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, NewRuntimeObject(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, NewRuntimeObject(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"
|
||||
@@ -1506,3 +1531,49 @@ func TestForceUint64(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
+308
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
+62
-2
@@ -3,6 +3,7 @@ package dara
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
@@ -10,7 +11,6 @@ import (
|
||||
"strings"
|
||||
)
|
||||
|
||||
// 定义 Event 结构体
|
||||
type SSEEvent struct {
|
||||
Id *string
|
||||
Event *string
|
||||
@@ -18,7 +18,6 @@ type SSEEvent struct {
|
||||
Retry *int
|
||||
}
|
||||
|
||||
// 解析单个事件
|
||||
func parseEvent(lines []string) *SSEEvent {
|
||||
event := &SSEEvent{}
|
||||
for _, line := range lines {
|
||||
@@ -153,3 +152,64 @@ func ReadAsSSE(body io.ReadCloser, eventChannel chan *SSEEvent, errorChannel cha
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
package dara
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io/ioutil"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/alibabacloud-go/tea/utils"
|
||||
)
|
||||
@@ -342,3 +344,168 @@ func Test_parseEvent(t *testing.T) {
|
||||
utils.AssertNil(t, event.Retry)
|
||||
})
|
||||
}
|
||||
|
||||
func Test_ReadAsSSEWithContext(t *testing.T) {
|
||||
// Test case 1: Basic SSE event with context
|
||||
t.Run("BasicSSEEventWithContext", func(t *testing.T) {
|
||||
sseData := "data: hello world\n\n"
|
||||
reader := ioutil.NopCloser(strings.NewReader(sseData))
|
||||
ctx := context.Background()
|
||||
|
||||
eventChannel := make(chan *SSEEvent, 1)
|
||||
errorChannel := make(chan error, 1)
|
||||
ReadAsSSEWithContext(ctx, reader, eventChannel, errorChannel)
|
||||
|
||||
event := <-eventChannel
|
||||
err := <-errorChannel
|
||||
|
||||
utils.AssertNil(t, err)
|
||||
utils.AssertNotNil(t, event)
|
||||
utils.AssertNotNil(t, event.Data)
|
||||
utils.AssertEqual(t, "hello world", *event.Data)
|
||||
utils.AssertNil(t, event.Event)
|
||||
utils.AssertNil(t, event.Id)
|
||||
utils.AssertNil(t, event.Retry)
|
||||
})
|
||||
|
||||
// Test case 2: SSE event with context timeout
|
||||
t.Run("SSEWithContextTimeout", func(t *testing.T) {
|
||||
// 使用一个快过期的context来测试超时检查
|
||||
sseData := "data: hello world\n\n"
|
||||
reader := ioutil.NopCloser(strings.NewReader(sseData))
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Nanosecond)
|
||||
defer cancel()
|
||||
|
||||
// 等待context超时
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
eventChannel := make(chan *SSEEvent, 1)
|
||||
errorChannel := make(chan error, 1)
|
||||
ReadAsSSEWithContext(ctx, reader, eventChannel, errorChannel)
|
||||
|
||||
// 应该立即收到超时错误
|
||||
select {
|
||||
case err := <-errorChannel:
|
||||
utils.AssertNotNil(t, err)
|
||||
utils.AssertEqual(t, context.DeadlineExceeded, err)
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
t.Fatal("Expected timeout error but didn't receive one")
|
||||
}
|
||||
})
|
||||
|
||||
// Test case 3: SSE event with context cancellation
|
||||
t.Run("SSEWithContextCancellation", func(t *testing.T) {
|
||||
// 使用一个已经取消的context
|
||||
sseData := "data: hello world\n\n"
|
||||
reader := ioutil.NopCloser(strings.NewReader(sseData))
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel() // 立即取消
|
||||
|
||||
eventChannel := make(chan *SSEEvent, 1)
|
||||
errorChannel := make(chan error, 1)
|
||||
ReadAsSSEWithContext(ctx, reader, eventChannel, errorChannel)
|
||||
|
||||
// 应该立即收到取消错误
|
||||
select {
|
||||
case err := <-errorChannel:
|
||||
utils.AssertNotNil(t, err)
|
||||
utils.AssertEqual(t, context.Canceled, err)
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
t.Fatal("Expected cancellation error but didn't receive one")
|
||||
}
|
||||
})
|
||||
|
||||
// Test case 4: Complete SSE event with context
|
||||
t.Run("CompleteSSEEventWithContext", func(t *testing.T) {
|
||||
sseData := "id: 456\nevent: notification\ndata: welcome\ndata: to sse\nretry: 3000\n\n"
|
||||
reader := ioutil.NopCloser(strings.NewReader(sseData))
|
||||
ctx := context.Background()
|
||||
|
||||
eventChannel := make(chan *SSEEvent, 1)
|
||||
errorChannel := make(chan error, 1)
|
||||
ReadAsSSEWithContext(ctx, reader, eventChannel, errorChannel)
|
||||
|
||||
event := <-eventChannel
|
||||
err := <-errorChannel
|
||||
|
||||
utils.AssertNil(t, err)
|
||||
utils.AssertNotNil(t, event)
|
||||
utils.AssertNotNil(t, event.Data)
|
||||
utils.AssertEqual(t, "welcome\nto sse", *event.Data)
|
||||
utils.AssertNotNil(t, event.Event)
|
||||
utils.AssertEqual(t, "notification", *event.Event)
|
||||
utils.AssertNotNil(t, event.Id)
|
||||
utils.AssertEqual(t, "456", *event.Id)
|
||||
utils.AssertNotNil(t, event.Retry)
|
||||
utils.AssertEqual(t, 3000, *event.Retry)
|
||||
})
|
||||
|
||||
// Test case 5: Multiple SSE events with context
|
||||
t.Run("MultipleSSEEventsWithContext", func(t *testing.T) {
|
||||
sseData := "data: first\n\n" + "data: second\n\n"
|
||||
reader := ioutil.NopCloser(strings.NewReader(sseData))
|
||||
ctx := context.Background()
|
||||
|
||||
eventChannel := make(chan *SSEEvent, 2)
|
||||
errorChannel := make(chan error, 1)
|
||||
ReadAsSSEWithContext(ctx, reader, eventChannel, errorChannel)
|
||||
|
||||
event1 := <-eventChannel
|
||||
event2 := <-eventChannel
|
||||
err := <-errorChannel
|
||||
|
||||
utils.AssertNil(t, err)
|
||||
utils.AssertNotNil(t, event1)
|
||||
utils.AssertNotNil(t, event1.Data)
|
||||
utils.AssertEqual(t, "first", *event1.Data)
|
||||
utils.AssertNotNil(t, event2)
|
||||
utils.AssertNotNil(t, event2.Data)
|
||||
utils.AssertEqual(t, "second", *event2.Data)
|
||||
})
|
||||
|
||||
// Test case 6: Context cancellation during event processing
|
||||
t.Run("ContextCancellationDuringProcessing", func(t *testing.T) {
|
||||
// 验证context取消机制的基本功能
|
||||
sseData := "data: event1\n\n"
|
||||
reader := ioutil.NopCloser(strings.NewReader(sseData))
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
eventChannel := make(chan *SSEEvent, 1)
|
||||
errorChannel := make(chan error, 1)
|
||||
ReadAsSSEWithContext(ctx, reader, eventChannel, errorChannel)
|
||||
|
||||
// 读取第一个事件
|
||||
event1 := <-eventChannel
|
||||
utils.AssertNotNil(t, event1)
|
||||
utils.AssertNotNil(t, event1.Data)
|
||||
utils.AssertEqual(t, "event1", *event1.Data)
|
||||
|
||||
// 正常结束,收到nil错误
|
||||
err := <-errorChannel
|
||||
utils.AssertNil(t, err)
|
||||
|
||||
cancel() // 清理资源
|
||||
})
|
||||
|
||||
// Test case 7: Empty stream with context
|
||||
t.Run("EmptyStreamWithContext", func(t *testing.T) {
|
||||
reader := ioutil.NopCloser(strings.NewReader(""))
|
||||
ctx := context.Background()
|
||||
|
||||
eventChannel := make(chan *SSEEvent, 1)
|
||||
errorChannel := make(chan error, 1)
|
||||
ReadAsSSEWithContext(ctx, reader, eventChannel, errorChannel)
|
||||
|
||||
// 应该立即收到nil错误(EOF)
|
||||
err := <-errorChannel
|
||||
utils.AssertNil(t, err)
|
||||
|
||||
// 事件通道应该为空
|
||||
select {
|
||||
case event := <-eventChannel:
|
||||
utils.AssertNil(t, event) // 不应该收到任何事件
|
||||
default:
|
||||
// 这是期望的行为
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
+15
-16
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user