Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 21389214ea | |||
| c2e107f1af | |||
| 5ff9708f43 |
-125
@@ -341,131 +341,6 @@ func DoRequest(request *Request, runtimeObject *RuntimeObject) (response *Respon
|
||||
return
|
||||
}
|
||||
|
||||
func DoRequestWithCtx(ctx context.Context, request *Request, runtimeObject *RuntimeObject) (response *Response, err error) {
|
||||
if runtimeObject == nil {
|
||||
runtimeObject = &RuntimeObject{}
|
||||
}
|
||||
fieldMap := make(map[string]string)
|
||||
utils.InitLogMsg(fieldMap)
|
||||
defer func() {
|
||||
if runtimeObject.Logger != nil {
|
||||
runtimeObject.Logger.PrintLog(fieldMap, err)
|
||||
}
|
||||
}()
|
||||
if request.Method == nil {
|
||||
request.Method = String("GET")
|
||||
}
|
||||
|
||||
if request.Protocol == nil {
|
||||
request.Protocol = String("http")
|
||||
} else {
|
||||
request.Protocol = String(strings.ToLower(StringValue(request.Protocol)))
|
||||
}
|
||||
|
||||
requestURL := ""
|
||||
request.Domain = request.Headers["host"]
|
||||
if request.Port != nil {
|
||||
request.Domain = String(fmt.Sprintf("%s:%d", StringValue(request.Domain), IntValue(request.Port)))
|
||||
}
|
||||
requestURL = fmt.Sprintf("%s://%s%s", StringValue(request.Protocol), StringValue(request.Domain), StringValue(request.Pathname))
|
||||
queryParams := request.Query
|
||||
// sort QueryParams by key
|
||||
q := url.Values{}
|
||||
for key, value := range queryParams {
|
||||
q.Add(key, StringValue(value))
|
||||
}
|
||||
querystring := q.Encode()
|
||||
if len(querystring) > 0 {
|
||||
if strings.Contains(requestURL, "?") {
|
||||
requestURL = fmt.Sprintf("%s&%s", requestURL, querystring)
|
||||
} else {
|
||||
requestURL = fmt.Sprintf("%s?%s", requestURL, querystring)
|
||||
}
|
||||
}
|
||||
debugLog("> %s %s", StringValue(request.Method), requestURL)
|
||||
|
||||
httpRequest, err := http.NewRequestWithContext(ctx, StringValue(request.Method), requestURL, request.Body)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
httpRequest.Host = StringValue(request.Domain)
|
||||
|
||||
var client HttpClient
|
||||
if runtimeObject.HttpClient == nil {
|
||||
client = getDaraClient(runtimeObject.getClientTag(StringValue(request.Domain)))
|
||||
} else {
|
||||
client = runtimeObject.HttpClient
|
||||
}
|
||||
|
||||
trans, err := getHttpTransport(request, runtimeObject)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if defaultClient, ok := client.(*daraClient); ok {
|
||||
defaultClient.Lock()
|
||||
if !defaultClient.ifInit || defaultClient.httpClient.Transport == nil {
|
||||
defaultClient.httpClient.Transport = trans
|
||||
}
|
||||
defaultClient.httpClient.Timeout = time.Duration(IntValue(runtimeObject.ReadTimeout)) * time.Millisecond
|
||||
defaultClient.ifInit = true
|
||||
defaultClient.Unlock()
|
||||
}
|
||||
|
||||
for key, value := range request.Headers {
|
||||
if value == nil || key == "content-length" {
|
||||
continue
|
||||
} else if key == "host" {
|
||||
httpRequest.Header["Host"] = []string{*value}
|
||||
delete(httpRequest.Header, "host")
|
||||
} else if key == "user-agent" {
|
||||
httpRequest.Header["User-Agent"] = []string{*value}
|
||||
delete(httpRequest.Header, "user-agent")
|
||||
} else {
|
||||
httpRequest.Header[key] = []string{*value}
|
||||
}
|
||||
debugLog("> %s: %s", key, StringValue(value))
|
||||
}
|
||||
contentlength, _ := strconv.Atoi(StringValue(request.Headers["content-length"]))
|
||||
event := utils.NewProgressEvent(utils.TransferStartedEvent, 0, int64(contentlength), 0)
|
||||
utils.PublishProgress(runtimeObject.Listener, event)
|
||||
|
||||
putMsgToMap(fieldMap, httpRequest)
|
||||
startTime := time.Now()
|
||||
fieldMap["{start_time}"] = startTime.Format("2006-01-02 15:04:05")
|
||||
res, err := hookDo(client.Call)(httpRequest, trans)
|
||||
fieldMap["{cost}"] = time.Since(startTime).String()
|
||||
completedBytes := int64(0)
|
||||
if runtimeObject.Tracker != nil {
|
||||
completedBytes = runtimeObject.Tracker.CompletedBytes
|
||||
}
|
||||
if err != nil {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
err = TeaSDKError(ctx.Err())
|
||||
default:
|
||||
}
|
||||
|
||||
event = utils.NewProgressEvent(utils.TransferFailedEvent, completedBytes, int64(contentlength), 0)
|
||||
utils.PublishProgress(runtimeObject.Listener, event)
|
||||
return
|
||||
}
|
||||
|
||||
event = utils.NewProgressEvent(utils.TransferCompletedEvent, completedBytes, int64(contentlength), 0)
|
||||
utils.PublishProgress(runtimeObject.Listener, event)
|
||||
|
||||
response = NewResponse(res)
|
||||
fieldMap["{code}"] = strconv.Itoa(res.StatusCode)
|
||||
fieldMap["{res_headers}"] = Stringify(res.Header)
|
||||
debugLog("< HTTP/1.1 %s", res.Status)
|
||||
for key, value := range res.Header {
|
||||
debugLog("< %s: %s", key, strings.Join(value, ""))
|
||||
if len(value) != 0 {
|
||||
response.Headers[strings.ToLower(key)] = String(value[0])
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func getHttpTransport(req *Request, runtime *RuntimeObject) (*http.Transport, error) {
|
||||
trans := new(http.Transport)
|
||||
httpProxy, err := getHttpProxy(StringValue(req.Protocol), StringValue(req.Domain), runtime)
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"os"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -654,45 +653,6 @@ func Test_DoRequest(t *testing.T) {
|
||||
utils.AssertEqual(t, `Internal error`, err.Error())
|
||||
}
|
||||
|
||||
func Test_DoRequestWithContextCancellation(t *testing.T) {
|
||||
origTestHookDo := hookDo
|
||||
defer func() { hookDo = origTestHookDo }()
|
||||
|
||||
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, transport *http.Transport) (*http.Response, error) {
|
||||
time.Sleep(2 * time.Second)
|
||||
return &http.Response{StatusCode: 200, Header: http.Header{}, Body: http.NoBody}, nil
|
||||
}
|
||||
}
|
||||
|
||||
// 创建带超时的上下文
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 1 * time.Second)
|
||||
defer cancel()
|
||||
|
||||
// 创建请求
|
||||
request := NewRequest()
|
||||
request.Method = String("GET")
|
||||
request.Protocol = String("http")
|
||||
request.Headers["host"] = String("tea-cn-hangzhou.aliyuncs.com")
|
||||
|
||||
// 将上下文放入运行时对象
|
||||
runtimeObj := map[string]interface{}{
|
||||
"ctx": ctx,
|
||||
}
|
||||
runtimeObject := NewRuntimeObject(runtimeObj)
|
||||
|
||||
fmt.Printf("runtime: %v \n", runtimeObject)
|
||||
resp, err := DoRequest(request, runtimeObject)
|
||||
|
||||
fmt.Printf("response: %v, error: %v\n", resp, err)
|
||||
utils.AssertNil(t, resp)
|
||||
if err == nil {
|
||||
t.Fatal("Expected an error due to context timeout, got nil")
|
||||
}
|
||||
utils.AssertContains(t, err.Error(), "context deadline exceeded")
|
||||
}
|
||||
|
||||
func Test_DoRequestWithConcurrent(t *testing.T) {
|
||||
origTestHookDo := hookDo
|
||||
defer func() { hookDo = origTestHookDo }()
|
||||
@@ -724,53 +684,6 @@ func Test_DoRequestWithConcurrent(t *testing.T) {
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
func Test_DoRequestWithConcurrentAndContext(t *testing.T) {
|
||||
origTestHookDo := hookDo
|
||||
defer func() { hookDo = origTestHookDo }()
|
||||
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, transport *http.Transport) (*http.Response, error) {
|
||||
time.Sleep(100 * time.Millisecond) // 模拟请求延迟
|
||||
return mockResponse(200, ``, nil)
|
||||
}
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 50; i++ {
|
||||
wg.Add(1)
|
||||
go func(readTimeout int) {
|
||||
defer wg.Done()
|
||||
|
||||
// 每个 goroutine 有自己的上下文,设定超时控制在 500 毫秒
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
runtime := NewRuntimeObject(map[string]interface{}{
|
||||
"readTimeout": readTimeout,
|
||||
"ctx": ctx, // 将上下文集成到运行时对象中
|
||||
})
|
||||
|
||||
for j := 0; j < 50; j++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
|
||||
request := NewRequest()
|
||||
resp, err := DoRequest(request, runtime)
|
||||
|
||||
if err != nil {
|
||||
// 检查是否是由于上下文超时导致的错误
|
||||
utils.AssertContains(t, err.Error(), "context deadline exceeded")
|
||||
} else {
|
||||
utils.AssertNil(t, err)
|
||||
utils.AssertNotNil(t, resp)
|
||||
}
|
||||
}()
|
||||
}
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
func Test_getHttpProxy(t *testing.T) {
|
||||
originHttpProxy := os.Getenv("HTTP_PROXY")
|
||||
originHttpsProxy := os.Getenv("HTTPS_PROXY")
|
||||
|
||||
+3
-23
@@ -6,7 +6,6 @@ import (
|
||||
jsoniter "github.com/json-iterator/go"
|
||||
"github.com/modern-go/reflect2"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"math"
|
||||
"reflect"
|
||||
"strconv"
|
||||
@@ -333,28 +332,9 @@ func (decoder *nullableFuzzyFloat64Decoder) Decode(ptr unsafe.Pointer, iter *jso
|
||||
}
|
||||
}
|
||||
|
||||
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 Stringify(m interface{}) string {
|
||||
byt, _ := json.Marshal(m)
|
||||
return string(byt)
|
||||
}
|
||||
|
||||
func ParseJSON(a string) interface{} {
|
||||
|
||||
@@ -3,7 +3,6 @@ package dara
|
||||
import (
|
||||
"reflect"
|
||||
"testing"
|
||||
"strings"
|
||||
|
||||
"github.com/alibabacloud-go/tea/utils"
|
||||
jsoniter "github.com/json-iterator/go"
|
||||
@@ -879,42 +878,3 @@ 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)
|
||||
}
|
||||
+8
-9
@@ -22,21 +22,21 @@ type SSEEvent struct {
|
||||
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: ") {
|
||||
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: ") {
|
||||
eventName := strings.TrimPrefix(line, "event: ")
|
||||
event.Event = &eventName
|
||||
} else if strings.HasPrefix(line, "id:") {
|
||||
id := strings.TrimPrefix(line, "id:")
|
||||
} 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, "retry: ") {
|
||||
var retry int
|
||||
fmt.Sscanf(strings.TrimPrefix(line, "retry:"), "%d", &retry)
|
||||
fmt.Sscanf(strings.TrimPrefix(line, "retry: "), "%d", &retry)
|
||||
event.Retry = &retry
|
||||
}
|
||||
}
|
||||
@@ -45,7 +45,6 @@ func parseEvent(lines []string) *SSEEvent {
|
||||
data := strings.TrimRight(*event.Data, "\n")
|
||||
event.Data = &data
|
||||
}
|
||||
|
||||
return event
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user