Compare commits
13 Commits
httpclient
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
| 75efc7f189 | |||
| af5f206bd8 | |||
| 5c6eb77b42 | |||
| 0f6d084c67 | |||
| df04276c08 | |||
| 622f315135 | |||
| 31de98dc53 | |||
| f93724f865 | |||
| fdbb29ef15 | |||
| a4d8c8f7f5 | |||
| 4820a881e8 | |||
| 6d4a89b84e | |||
| a17e6d71ba |
+254
-40
File diff suppressed because it is too large
Load Diff
+393
-11
File diff suppressed because it is too large
Load Diff
+23
-3
@@ -6,6 +6,7 @@ import (
|
|||||||
jsoniter "github.com/json-iterator/go"
|
jsoniter "github.com/json-iterator/go"
|
||||||
"github.com/modern-go/reflect2"
|
"github.com/modern-go/reflect2"
|
||||||
"io"
|
"io"
|
||||||
|
"io/ioutil"
|
||||||
"math"
|
"math"
|
||||||
"reflect"
|
"reflect"
|
||||||
"strconv"
|
"strconv"
|
||||||
@@ -332,9 +333,28 @@ func (decoder *nullableFuzzyFloat64Decoder) Decode(ptr unsafe.Pointer, iter *jso
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func Stringify(m interface{}) string {
|
func Stringify(a interface{}) string {
|
||||||
byt, _ := json.Marshal(m)
|
switch v := a.(type) {
|
||||||
return string(byt)
|
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{} {
|
func ParseJSON(a string) interface{} {
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package dara
|
|||||||
import (
|
import (
|
||||||
"reflect"
|
"reflect"
|
||||||
"testing"
|
"testing"
|
||||||
|
"strings"
|
||||||
|
|
||||||
"github.com/alibabacloud-go/tea/utils"
|
"github.com/alibabacloud-go/tea/utils"
|
||||||
jsoniter "github.com/json-iterator/go"
|
jsoniter "github.com/json-iterator/go"
|
||||||
@@ -878,3 +879,42 @@ func TestUnmarshalWithDefaultDecoders(t *testing.T) {
|
|||||||
err = jsoniter.Unmarshal(from, toUint64)
|
err = jsoniter.Unmarshal(from, toUint64)
|
||||||
utils.AssertNotNil(t, err)
|
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
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
+94
-16
@@ -3,44 +3,63 @@ package dara
|
|||||||
import (
|
import (
|
||||||
"bufio"
|
"bufio"
|
||||||
"bytes"
|
"bytes"
|
||||||
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"io/ioutil"
|
"io/ioutil"
|
||||||
"strings"
|
"strings"
|
||||||
"fmt"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// 定义 Event 结构体
|
|
||||||
type SSEEvent struct {
|
type SSEEvent struct {
|
||||||
ID *string
|
Id *string
|
||||||
Event *string
|
Event *string
|
||||||
Data *string
|
Data *string
|
||||||
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 {
|
||||||
if strings.HasPrefix(line, "data: ") {
|
if strings.HasPrefix(line, "data:") {
|
||||||
data := strings.TrimPrefix(line, "data: ") + "\n"
|
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 {
|
if event.Data == nil {
|
||||||
event.Data = new(string)
|
event.Data = new(string)
|
||||||
}
|
}
|
||||||
*event.Data += data
|
*event.Data += data
|
||||||
} else if strings.HasPrefix(line, "event: ") {
|
} else if strings.HasPrefix(line, "event:") {
|
||||||
eventName := strings.TrimPrefix(line, "event: ")
|
var eventName string
|
||||||
|
if strings.HasPrefix(line, "event: ") {
|
||||||
|
eventName = strings.TrimPrefix(line, "event: ")
|
||||||
|
} else {
|
||||||
|
eventName = strings.TrimPrefix(line, "event:")
|
||||||
|
}
|
||||||
event.Event = &eventName
|
event.Event = &eventName
|
||||||
} else if strings.HasPrefix(line, "id: ") {
|
} else if strings.HasPrefix(line, "id:") {
|
||||||
id := strings.TrimPrefix(line, "id: ")
|
var id string
|
||||||
event.ID = &id
|
if strings.HasPrefix(line, "id: ") {
|
||||||
} else if strings.HasPrefix(line, "retry: ") {
|
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
|
var retry int
|
||||||
fmt.Sscanf(strings.TrimPrefix(line, "retry: "), "%d", &retry)
|
fmt.Sscanf(retryStr, "%d", &retry)
|
||||||
event.Retry = &retry
|
event.Retry = &retry
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// Remove last newline from data
|
|
||||||
if event.Data != nil {
|
if event.Data != nil {
|
||||||
data := strings.TrimRight(*event.Data, "\n")
|
data := strings.TrimRight(*event.Data, "\n")
|
||||||
event.Data = &data
|
event.Data = &data
|
||||||
@@ -132,6 +151,65 @@ func ReadAsSSE(body io.ReadCloser, eventChannel chan *SSEEvent, errorChannel cha
|
|||||||
eventLines = append(eventLines, line)
|
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)
|
||||||
|
}
|
||||||
|
}()
|
||||||
}
|
}
|
||||||
|
|||||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,35 @@
|
|||||||
|
{
|
||||||
|
"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"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,8 @@
|
|||||||
|
typedef HttpRequest;
|
||||||
|
typedef HttpResponse;
|
||||||
|
typedef HttpTransport;
|
||||||
|
|
||||||
|
init(){
|
||||||
|
}
|
||||||
|
|
||||||
|
async function call(request: HttpRequest, transport: HttpTransport): HttpResponse;
|
||||||
@@ -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"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,8 @@
|
|||||||
|
typedef HttpRequest;
|
||||||
|
typedef HttpResponse;
|
||||||
|
typedef HttpTransport;
|
||||||
|
|
||||||
|
init(){
|
||||||
|
}
|
||||||
|
|
||||||
|
async function call(request: HttpRequest, transport: HttpTransport): HttpResponse;
|
||||||
+61
-34
@@ -31,7 +31,28 @@ import (
|
|||||||
|
|
||||||
var debugLog = debug.Init("tea")
|
var debugLog = debug.Init("tea")
|
||||||
|
|
||||||
var hookDo = func(fn func(req *http.Request) (*http.Response, error)) func(req *http.Request) (*http.Response, error) {
|
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) {
|
||||||
return fn
|
return fn
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -97,12 +118,7 @@ type RuntimeObject struct {
|
|||||||
Listener utils.ProgressListener `json:"listener" xml:"listener"`
|
Listener utils.ProgressListener `json:"listener" xml:"listener"`
|
||||||
Tracker *utils.ReaderTracker `json:"tracker" xml:"tracker"`
|
Tracker *utils.ReaderTracker `json:"tracker" xml:"tracker"`
|
||||||
Logger *utils.Logger `json:"logger" xml:"logger"`
|
Logger *utils.Logger `json:"logger" xml:"logger"`
|
||||||
}
|
HttpClient
|
||||||
|
|
||||||
type teaClient struct {
|
|
||||||
sync.Mutex
|
|
||||||
httpClient *http.Client
|
|
||||||
ifInit bool
|
|
||||||
}
|
}
|
||||||
|
|
||||||
var clientPool = &sync.Map{}
|
var clientPool = &sync.Map{}
|
||||||
@@ -143,6 +159,9 @@ func NewRuntimeObject(runtime map[string]interface{}) *RuntimeObject {
|
|||||||
if runtime["logger"] != nil {
|
if runtime["logger"] != nil {
|
||||||
runtimeObject.Logger = runtime["logger"].(*utils.Logger)
|
runtimeObject.Logger = runtime["logger"].(*utils.Logger)
|
||||||
}
|
}
|
||||||
|
if runtime["httpClient"] != nil {
|
||||||
|
runtimeObject.HttpClient = runtime["httpClient"].(HttpClient)
|
||||||
|
}
|
||||||
return runtimeObject
|
return runtimeObject
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -351,18 +370,27 @@ func DoRequest(request *Request, requestRuntime map[string]interface{}) (respons
|
|||||||
}
|
}
|
||||||
httpRequest.Host = StringValue(request.Domain)
|
httpRequest.Host = StringValue(request.Domain)
|
||||||
|
|
||||||
client := getTeaClient(runtimeObject.getClientTag(StringValue(request.Domain)))
|
var client HttpClient
|
||||||
client.Lock()
|
if runtimeObject.HttpClient == nil {
|
||||||
if !client.ifInit {
|
client = getTeaClient(runtimeObject.getClientTag(StringValue(request.Domain)))
|
||||||
trans, err := getHttpTransport(request, runtimeObject)
|
} else {
|
||||||
if err != nil {
|
client = runtimeObject.HttpClient
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
client.httpClient.Timeout = time.Duration(IntValue(runtimeObject.ReadTimeout)) * time.Millisecond
|
|
||||||
client.httpClient.Transport = trans
|
|
||||||
client.ifInit = true
|
|
||||||
}
|
}
|
||||||
client.Unlock()
|
|
||||||
|
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
|
||||||
|
}
|
||||||
|
defaultClient.httpClient.Timeout = time.Duration(IntValue(runtimeObject.ConnectTimeout)+IntValue(runtimeObject.ReadTimeout)) * time.Millisecond
|
||||||
|
defaultClient.ifInit = true
|
||||||
|
defaultClient.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
for key, value := range request.Headers {
|
for key, value := range request.Headers {
|
||||||
if value == nil || key == "content-length" {
|
if value == nil || key == "content-length" {
|
||||||
continue
|
continue
|
||||||
@@ -384,7 +412,7 @@ func DoRequest(request *Request, requestRuntime map[string]interface{}) (respons
|
|||||||
putMsgToMap(fieldMap, httpRequest)
|
putMsgToMap(fieldMap, httpRequest)
|
||||||
startTime := time.Now()
|
startTime := time.Now()
|
||||||
fieldMap["{start_time}"] = startTime.Format("2006-01-02 15:04:05")
|
fieldMap["{start_time}"] = startTime.Format("2006-01-02 15:04:05")
|
||||||
res, err := hookDo(client.httpClient.Do)(httpRequest)
|
res, err := hookDo(client.Call)(httpRequest, trans)
|
||||||
fieldMap["{cost}"] = time.Since(startTime).String()
|
fieldMap["{cost}"] = time.Since(startTime).String()
|
||||||
completedBytes := int64(0)
|
completedBytes := int64(0)
|
||||||
if runtimeObject.Tracker != nil {
|
if runtimeObject.Tracker != nil {
|
||||||
@@ -414,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
|
||||||
@@ -467,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,
|
||||||
@@ -574,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
|
||||||
@@ -582,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)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+113
-10
@@ -487,11 +487,36 @@ func Test_GetBackoffTime(t *testing.T) {
|
|||||||
utils.AssertEqual(t, true, IntValue(ms) <= 3)
|
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) {
|
func Test_DoRequest(t *testing.T) {
|
||||||
origTestHookDo := hookDo
|
origTestHookDo := hookDo
|
||||||
defer func() { hookDo = origTestHookDo }()
|
defer func() { hookDo = origTestHookDo }()
|
||||||
hookDo = func(fn func(req *http.Request) (*http.Response, error)) func(req *http.Request) (*http.Response, error) {
|
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) (*http.Response, error) {
|
return func(req *http.Request, transport *http.Transport) (*http.Response, error) {
|
||||||
return mockResponse(200, ``, errors.New("Internal error"))
|
return mockResponse(200, ``, errors.New("Internal error"))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -530,17 +555,42 @@ func Test_DoRequest(t *testing.T) {
|
|||||||
utils.AssertNil(t, resp)
|
utils.AssertNil(t, resp)
|
||||||
utils.AssertContains(t, err.Error(), ` invalid URL escape "%gf"`)
|
utils.AssertContains(t, err.Error(), ` invalid URL escape "%gf"`)
|
||||||
|
|
||||||
hookDo = func(fn func(req *http.Request) (*http.Response, error)) func(req *http.Request) (*http.Response, error) {
|
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) (*http.Response, error) {
|
return func(req *http.Request, transport *http.Transport) (*http.Response, error) {
|
||||||
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"
|
||||||
@@ -579,8 +629,8 @@ func Test_DoRequest(t *testing.T) {
|
|||||||
utils.AssertNil(t, err)
|
utils.AssertNil(t, err)
|
||||||
utils.AssertEqual(t, "test", StringValue(resp.Headers["tea"]))
|
utils.AssertEqual(t, "test", StringValue(resp.Headers["tea"]))
|
||||||
|
|
||||||
hookDo = func(fn func(req *http.Request) (*http.Response, error)) func(req *http.Request) (*http.Response, error) {
|
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) (*http.Response, error) {
|
return func(req *http.Request, transport *http.Transport) (*http.Response, error) {
|
||||||
utils.AssertEqual(t, "tea-cn-hangzhou.aliyuncs.com:1080", req.Host)
|
utils.AssertEqual(t, "tea-cn-hangzhou.aliyuncs.com:1080", req.Host)
|
||||||
return mockResponse(200, ``, errors.New("Internal error"))
|
return mockResponse(200, ``, errors.New("Internal error"))
|
||||||
}
|
}
|
||||||
@@ -592,13 +642,20 @@ func Test_DoRequest(t *testing.T) {
|
|||||||
resp, err = DoRequest(request, runtimeObj)
|
resp, err = DoRequest(request, runtimeObj)
|
||||||
utils.AssertNil(t, resp)
|
utils.AssertNil(t, resp)
|
||||||
utils.AssertEqual(t, `Internal error`, err.Error())
|
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) {
|
func Test_DoRequestWithConcurrent(t *testing.T) {
|
||||||
origTestHookDo := hookDo
|
origTestHookDo := hookDo
|
||||||
defer func() { hookDo = origTestHookDo }()
|
defer func() { hookDo = origTestHookDo }()
|
||||||
hookDo = func(fn func(req *http.Request) (*http.Response, error)) func(req *http.Request) (*http.Response, error) {
|
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) (*http.Response, error) {
|
return func(req *http.Request, transport *http.Transport) (*http.Response, error) {
|
||||||
return mockResponse(200, ``, nil)
|
return mockResponse(200, ``, nil)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -694,11 +751,11 @@ func Test_SetDialContext(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func Test_hookdo(t *testing.T) {
|
func Test_hookdo(t *testing.T) {
|
||||||
fn := func(req *http.Request) (*http.Response, error) {
|
fn := func(req *http.Request, transport *http.Transport) (*http.Response, error) {
|
||||||
return nil, errors.New("hookdo")
|
return nil, errors.New("hookdo")
|
||||||
}
|
}
|
||||||
result := hookDo(fn)
|
result := hookDo(fn)
|
||||||
resp, err := result(nil)
|
resp, err := result(nil, nil)
|
||||||
utils.AssertNil(t, resp)
|
utils.AssertNil(t, resp)
|
||||||
utils.AssertEqual(t, "hookdo", err.Error())
|
utils.AssertEqual(t, "hookdo", err.Error())
|
||||||
}
|
}
|
||||||
@@ -910,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