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"`
|
IgnoreSSL *bool `json:"ignoreSSL" xml:"ignoreSSL"`
|
||||||
ReadTimeout *int `json:"readTimeout" xml:"readTimeout"`
|
ReadTimeout *int `json:"readTimeout" xml:"readTimeout"`
|
||||||
ConnectTimeout *int `json:"connectTimeout" xml:"connectTimeout"`
|
ConnectTimeout *int `json:"connectTimeout" xml:"connectTimeout"`
|
||||||
|
IdleTimeout *int `json:"idleTimeout" xml:"idleTimeout"`
|
||||||
LocalAddr *string `json:"localAddr" xml:"localAddr"`
|
LocalAddr *string `json:"localAddr" xml:"localAddr"`
|
||||||
HttpProxy *string `json:"httpProxy" xml:"httpProxy"`
|
HttpProxy *string `json:"httpProxy" xml:"httpProxy"`
|
||||||
HttpsProxy *string `json:"httpsProxy" xml:"httpsProxy"`
|
HttpsProxy *string `json:"httpsProxy" xml:"httpsProxy"`
|
||||||
@@ -114,7 +115,7 @@ type RuntimeObject struct {
|
|||||||
|
|
||||||
func (r *RuntimeObject) getClientTag(domain string) string {
|
func (r *RuntimeObject) getClientTag(domain string) string {
|
||||||
return strconv.FormatBool(BoolValue(r.IgnoreSSL)) + strconv.Itoa(IntValue(r.ReadTimeout)) +
|
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
|
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"]),
|
IgnoreSSL: TransInterfaceToBool(runtime["ignoreSSL"]),
|
||||||
ReadTimeout: TransInterfaceToInt(runtime["readTimeout"]),
|
ReadTimeout: TransInterfaceToInt(runtime["readTimeout"]),
|
||||||
ConnectTimeout: TransInterfaceToInt(runtime["connectTimeout"]),
|
ConnectTimeout: TransInterfaceToInt(runtime["connectTimeout"]),
|
||||||
|
IdleTimeout: TransInterfaceToInt(runtime["idleTimeout"]),
|
||||||
LocalAddr: TransInterfaceToString(runtime["localAddr"]),
|
LocalAddr: TransInterfaceToString(runtime["localAddr"]),
|
||||||
HttpProxy: TransInterfaceToString(runtime["httpProxy"]),
|
HttpProxy: TransInterfaceToString(runtime["httpProxy"]),
|
||||||
HttpsProxy: TransInterfaceToString(runtime["httpsProxy"]),
|
HttpsProxy: TransInterfaceToString(runtime["httpsProxy"]),
|
||||||
@@ -321,7 +323,7 @@ func DoRequest(request *Request, runtimeObject *RuntimeObject) (response *Respon
|
|||||||
if !defaultClient.ifInit || defaultClient.httpClient.Transport == nil {
|
if !defaultClient.ifInit || defaultClient.httpClient.Transport == nil {
|
||||||
defaultClient.httpClient.Transport = trans
|
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.ifInit = true
|
||||||
defaultClient.Unlock()
|
defaultClient.Unlock()
|
||||||
}
|
}
|
||||||
@@ -440,7 +442,7 @@ func DoRequestWithCtx(ctx context.Context, request *Request, runtimeObject *Runt
|
|||||||
if !defaultClient.ifInit || defaultClient.httpClient.Transport == nil {
|
if !defaultClient.ifInit || defaultClient.httpClient.Transport == nil {
|
||||||
defaultClient.httpClient.Transport = trans
|
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.ifInit = true
|
||||||
defaultClient.Unlock()
|
defaultClient.Unlock()
|
||||||
}
|
}
|
||||||
@@ -502,6 +504,7 @@ func DoRequestWithCtx(ctx context.Context, request *Request, runtimeObject *Runt
|
|||||||
|
|
||||||
func getHttpTransport(req *Request, runtime *RuntimeObject) (*http.Transport, error) {
|
func getHttpTransport(req *Request, runtime *RuntimeObject) (*http.Transport, error) {
|
||||||
trans := new(http.Transport)
|
trans := new(http.Transport)
|
||||||
|
trans.ResponseHeaderTimeout = time.Duration(IntValue(runtime.ReadTimeout)) * time.Millisecond
|
||||||
httpProxy, err := getHttpProxy(StringValue(req.Protocol), StringValue(req.Domain), runtime)
|
httpProxy, err := getHttpProxy(StringValue(req.Protocol), StringValue(req.Domain), runtime)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@@ -555,7 +558,7 @@ func getHttpTransport(req *Request, runtime *RuntimeObject) (*http.Transport, er
|
|||||||
Password: password,
|
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{
|
&net.Dialer{
|
||||||
Timeout: time.Duration(IntValue(runtime.ConnectTimeout)) * time.Millisecond,
|
Timeout: time.Duration(IntValue(runtime.ConnectTimeout)) * time.Millisecond,
|
||||||
DualStack: true,
|
DualStack: true,
|
||||||
@@ -573,6 +576,9 @@ func getHttpTransport(req *Request, runtime *RuntimeObject) (*http.Transport, er
|
|||||||
trans.MaxIdleConns = IntValue(runtime.MaxIdleConns)
|
trans.MaxIdleConns = IntValue(runtime.MaxIdleConns)
|
||||||
trans.MaxIdleConnsPerHost = 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
|
return trans, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -691,7 +697,7 @@ func getSocks5Proxy(runtime *RuntimeObject) (proxy *url.URL, err error) {
|
|||||||
func getLocalAddr(localAddr string) (addr *net.TCPAddr) {
|
func getLocalAddr(localAddr string) (addr *net.TCPAddr) {
|
||||||
if localAddr != "" {
|
if localAddr != "" {
|
||||||
addr = &net.TCPAddr{
|
addr = &net.TCPAddr{
|
||||||
IP: []byte(localAddr),
|
IP: net.ParseIP(localAddr),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return addr
|
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) {
|
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) {
|
return func(ctx context.Context, network, address string) (net.Conn, error) {
|
||||||
if runtime.LocalAddr != nil && StringValue(runtime.LocalAddr) != "" {
|
timeout := time.Duration(IntValue(runtime.ConnectTimeout)) * time.Millisecond
|
||||||
netAddr := &net.TCPAddr{
|
dialer := &net.Dialer{
|
||||||
IP: []byte(StringValue(runtime.LocalAddr)),
|
Timeout: timeout,
|
||||||
}
|
Resolver: &net.Resolver{
|
||||||
return (&net.Dialer{
|
PreferGo: false,
|
||||||
Timeout: time.Duration(IntValue(runtime.ConnectTimeout)) * time.Second,
|
},
|
||||||
DualStack: true,
|
|
||||||
LocalAddr: netAddr,
|
|
||||||
}).DialContext(ctx, network, address)
|
|
||||||
}
|
}
|
||||||
return (&net.Dialer{
|
if runtime.LocalAddr != nil && StringValue(runtime.LocalAddr) != "" {
|
||||||
Timeout: time.Duration(IntValue(runtime.ConnectTimeout)) * time.Second,
|
dialer.LocalAddr = getLocalAddr(StringValue(runtime.LocalAddr))
|
||||||
DualStack: true,
|
}
|
||||||
}).DialContext(ctx, network, address)
|
return dialer.DialContext(ctx, network, address)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -732,12 +732,37 @@ func Test_DoRequest(t *testing.T) {
|
|||||||
return mockResponse(200, ``, nil)
|
return mockResponse(200, ``, nil)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
// Test socks5 proxy with user:password@host format
|
||||||
runtimeObj["socks5Proxy"] = "socks5://someuser:somepassword@ecs.aliyun.com"
|
runtimeObj["socks5Proxy"] = "socks5://someuser:somepassword@ecs.aliyun.com"
|
||||||
runtimeObj["localAddr"] = "127.0.0.1"
|
runtimeObj["localAddr"] = "127.0.0.1"
|
||||||
resp, err = DoRequest(request, NewRuntimeObject(runtimeObj))
|
resp, err = DoRequest(request, NewRuntimeObject(runtimeObj))
|
||||||
utils.AssertNil(t, err)
|
utils.AssertNil(t, err)
|
||||||
utils.AssertEqual(t, "test", StringValue(resp.Headers["tea"]))
|
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["key"] = "private rsa key"
|
||||||
runtimeObj["cert"] = "private certification"
|
runtimeObj["cert"] = "private certification"
|
||||||
runtimeObj["ca"] = "private ca"
|
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 (
|
import (
|
||||||
"bufio"
|
"bufio"
|
||||||
"bytes"
|
"bytes"
|
||||||
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
@@ -10,7 +11,6 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
)
|
)
|
||||||
|
|
||||||
// 定义 Event 结构体
|
|
||||||
type SSEEvent struct {
|
type SSEEvent struct {
|
||||||
Id *string
|
Id *string
|
||||||
Event *string
|
Event *string
|
||||||
@@ -18,7 +18,6 @@ type SSEEvent struct {
|
|||||||
Retry *int
|
Retry *int
|
||||||
}
|
}
|
||||||
|
|
||||||
// 解析单个事件
|
|
||||||
func parseEvent(lines []string) *SSEEvent {
|
func parseEvent(lines []string) *SSEEvent {
|
||||||
event := &SSEEvent{}
|
event := &SSEEvent{}
|
||||||
for _, line := range lines {
|
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
|
package dara
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"io/ioutil"
|
"io/ioutil"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/alibabacloud-go/tea/utils"
|
"github.com/alibabacloud-go/tea/utils"
|
||||||
)
|
)
|
||||||
@@ -342,3 +344,168 @@ func Test_parseEvent(t *testing.T) {
|
|||||||
utils.AssertNil(t, event.Retry)
|
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 {
|
if !defaultClient.ifInit || defaultClient.httpClient.Transport == nil {
|
||||||
defaultClient.httpClient.Transport = trans
|
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.ifInit = true
|
||||||
defaultClient.Unlock()
|
defaultClient.Unlock()
|
||||||
}
|
}
|
||||||
@@ -442,6 +442,7 @@ func DoRequest(request *Request, requestRuntime map[string]interface{}) (respons
|
|||||||
|
|
||||||
func getHttpTransport(req *Request, runtime *RuntimeObject) (*http.Transport, error) {
|
func getHttpTransport(req *Request, runtime *RuntimeObject) (*http.Transport, error) {
|
||||||
trans := new(http.Transport)
|
trans := new(http.Transport)
|
||||||
|
trans.ResponseHeaderTimeout = time.Duration(IntValue(runtime.ReadTimeout)) * time.Millisecond
|
||||||
httpProxy, err := getHttpProxy(StringValue(req.Protocol), StringValue(req.Domain), runtime)
|
httpProxy, err := getHttpProxy(StringValue(req.Protocol), StringValue(req.Domain), runtime)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@@ -495,7 +496,7 @@ func getHttpTransport(req *Request, runtime *RuntimeObject) (*http.Transport, er
|
|||||||
Password: password,
|
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{
|
&net.Dialer{
|
||||||
Timeout: time.Duration(IntValue(runtime.ConnectTimeout)) * time.Millisecond,
|
Timeout: time.Duration(IntValue(runtime.ConnectTimeout)) * time.Millisecond,
|
||||||
DualStack: true,
|
DualStack: true,
|
||||||
@@ -602,7 +603,7 @@ func getSocks5Proxy(runtime *RuntimeObject) (proxy *url.URL, err error) {
|
|||||||
func getLocalAddr(localAddr string) (addr *net.TCPAddr) {
|
func getLocalAddr(localAddr string) (addr *net.TCPAddr) {
|
||||||
if localAddr != "" {
|
if localAddr != "" {
|
||||||
addr = &net.TCPAddr{
|
addr = &net.TCPAddr{
|
||||||
IP: []byte(localAddr),
|
IP: net.ParseIP(localAddr),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return addr
|
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) {
|
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) {
|
return func(ctx context.Context, network, address string) (net.Conn, error) {
|
||||||
if runtime.LocalAddr != nil && StringValue(runtime.LocalAddr) != "" {
|
timeout := time.Duration(IntValue(runtime.ConnectTimeout)) * time.Millisecond
|
||||||
netAddr := &net.TCPAddr{
|
dialer := &net.Dialer{
|
||||||
IP: []byte(StringValue(runtime.LocalAddr)),
|
Timeout: timeout,
|
||||||
}
|
Resolver: &net.Resolver{
|
||||||
return (&net.Dialer{
|
PreferGo: false,
|
||||||
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,
|
|
||||||
DualStack: true,
|
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)
|
return mockResponse(200, ``, nil)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
// Test socks5 proxy with user:password@host format
|
||||||
runtimeObj["socks5Proxy"] = "socks5://someuser:somepassword@ecs.aliyun.com"
|
runtimeObj["socks5Proxy"] = "socks5://someuser:somepassword@ecs.aliyun.com"
|
||||||
runtimeObj["localAddr"] = "127.0.0.1"
|
runtimeObj["localAddr"] = "127.0.0.1"
|
||||||
resp, err = DoRequest(request, runtimeObj)
|
resp, err = DoRequest(request, runtimeObj)
|
||||||
utils.AssertNil(t, err)
|
utils.AssertNil(t, err)
|
||||||
utils.AssertEqual(t, "test", StringValue(resp.Headers["tea"]))
|
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["key"] = "private rsa key"
|
||||||
runtimeObj["cert"] = "private certification"
|
runtimeObj["cert"] = "private certification"
|
||||||
runtimeObj["ca"] = "private ca"
|
runtimeObj["ca"] = "private ca"
|
||||||
@@ -942,3 +967,49 @@ func Test_TransInt32AndInt(t *testing.T) {
|
|||||||
b := ToInt32(a)
|
b := ToInt32(a)
|
||||||
utils.AssertEqual(t, Int32Value(b), int32(10))
|
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