fix the sse

dara/core.go

@@ -34,7 +34,6 @@ type RuntimeOptions = util.RuntimeOptions
 type ExtendsParameters = util.ExtendsParameters
 
 var debugLog = debug.Init("dara")
-
 type HttpRequest interface {
 }
 
@@ -94,7 +93,6 @@ 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"`
@@ -115,7 +113,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)) + strconv.Itoa(IntValue(r.IdleTimeout)) + StringValue(r.LocalAddr) + StringValue(r.HttpProxy) +
+		strconv.Itoa(IntValue(r.ConnectTimeout)) + StringValue(r.LocalAddr) + StringValue(r.HttpProxy) +
 		StringValue(r.HttpsProxy) + StringValue(r.NoProxy) + StringValue(r.Socks5Proxy) + StringValue(r.Socks5NetWork) + domain
 }
 
@@ -129,7 +127,6 @@ 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"]),
@@ -186,39 +183,6 @@ func Convert(in interface{}, out interface{}) error {
 	return err
 }
 
-// ConvertChan converts the source data to the target type and sends it to the specified channel.
-// @param src - source data
-// @param destChan - target channel
-// @return error - error during the conversion process
-func ConvertChan(src interface{}, destChan interface{}) error {
-	destChanValue := reflect.ValueOf(destChan)
-	if destChanValue.Kind() != reflect.Chan {
-		return fmt.Errorf("destChan must be a channel")
-	}
-
-	if destChanValue.Type().ChanDir() == reflect.SendDir {
-		return fmt.Errorf("destChan must be a receive or bidirectional channel")
-	}
-
-	elemType := destChanValue.Type().Elem()
-
-	destValue := reflect.New(elemType).Interface()
-
-	err := Convert(src, destValue)
-	if err != nil {
-		return err
-	}
-	destValueElem := reflect.ValueOf(destValue).Elem()
-
-	defer func() {
-		if r := recover(); r != nil {
-		}
-	}()
-
-	destChanValue.TrySend(destValueElem)
-	return nil
-}
-
 // Recover is used to format error
 func Recover(in interface{}) error {
 	if in == nil {
@@ -557,7 +521,7 @@ func getHttpTransport(req *Request, runtime *RuntimeObject) (*http.Transport, er
 					Password: password,
 				}
 			}
-			dialer, err := proxy.SOCKS5(strings.ToLower(StringValue(runtime.Socks5NetWork)), socks5Proxy.Host, auth,
+			dialer, err := proxy.SOCKS5(strings.ToLower(StringValue(runtime.Socks5NetWork)), socks5Proxy.String(), auth,
 				&net.Dialer{
 					Timeout:   time.Duration(IntValue(runtime.ConnectTimeout)) * time.Millisecond,
 					DualStack: true,
@@ -575,9 +539,6 @@ 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
 }
 

dara/core_test.go

@@ -8,6 +8,7 @@ import (
 	"io/ioutil"
 	"net/http"
 	"os"
+	"fmt"
 	"reflect"
 	"strconv"
 	"strings"
@@ -193,150 +194,6 @@ func TestConvertType(t *testing.T) {
 	utils.AssertEqual(t, "test", string(out.Body))
 }
 
-// TestConvertChan tests the ConvertChan function
-func TestConvertChan(t *testing.T) {
-	// Test case 1: Successful conversion and sending to channel
-	t.Run("SuccessfulConversion", func(t *testing.T) {
-		// Create a channel to receive the converted data
-		ch := make(chan test, 1)
-
-		// Source data to convert
-		src := map[string]interface{}{
-			"key":  123,
-			"body": []byte("test"),
-		}
-
-		// Perform conversion
-		err := ConvertChan(src, ch)
-
-		// Check for errors
-		if err != nil {
-			t.Errorf("Expected no error, got %v", err)
-		}
-		utils.AssertNil(t, err)
-
-		// Check if data was sent to channel
-		select {
-		case result := <-ch:
-			utils.AssertEqual(t, "123", result.Key)
-			utils.AssertEqual(t, "test", string(result.Body))
-		}
-	})
-
-	// Test case 2: Invalid destination channel (not a channel)
-	t.Run("InvalidDestChan", func(t *testing.T) {
-		src := map[string]interface{}{
-			"test": "data",
-		}
-
-		// Pass a non-channel type
-		err := ConvertChan(src, "not_a_channel")
-
-		if err == nil {
-			t.Error("Expected error for invalid channel, got nil")
-		}
-
-		expected := "destChan must be a channel"
-		if err.Error() != expected {
-			t.Errorf("Expected error message '%s', got '%s'", expected, err.Error())
-		}
-	})
-
-	// Test case 3: Send-only channel
-	t.Run("SendOnlyChannel", func(t *testing.T) {
-		// Create a send-only channel
-		ch := make(chan<- map[string]interface{}, 1)
-
-		src := map[string]interface{}{
-			"test": "data",
-		}
-
-		err := ConvertChan(src, ch)
-
-		if err == nil {
-			t.Error("Expected error for send-only channel, got nil")
-		}
-
-		expected := "destChan must be a receive or bidirectional channel"
-		if err.Error() != expected {
-			t.Errorf("Expected error message '%s', got '%s'", expected, err.Error())
-		}
-	})
-
-	// Test case 4: Conversion with struct
-	t.Run("StructConversion", func(t *testing.T) {
-		type TestStruct struct {
-			Name string `json:"name"`
-			Age  int    `json:"age"`
-		}
-
-		ch := make(chan TestStruct, 1)
-
-		// Source data matching struct fields
-		src := map[string]interface{}{
-			"name": "Alice",
-			"age":  30,
-		}
-
-		err := ConvertChan(src, ch)
-
-		if err != nil {
-			t.Errorf("Expected no error, got %v", err)
-		}
-
-		select {
-		case result := <-ch:
-			if result.Name != "Alice" || result.Age != 30 {
-				t.Errorf("Expected {Name: Alice, Age: 30}, got %v", result)
-			}
-		default:
-			t.Error("Expected data in channel, but channel is empty")
-		}
-	})
-
-	// Test case 5: Nil source data
-	t.Run("NilSource", func(t *testing.T) {
-		ch := make(chan map[string]interface{}, 1)
-
-		err := ConvertChan(nil, ch)
-
-		// Depending on implementation, this might error or succeed
-		// Here we're testing it doesn't panic
-		if err != nil {
-			// If there's an error, that's fine, just make sure it's handled
-			t.Logf("Nil source resulted in error (acceptable): %v", err)
-		} else {
-			// If no error, check what was sent
-			select {
-			case result := <-ch:
-				if result == nil {
-					t.Log("Nil source correctly handled")
-				} else {
-					t.Errorf("Expected nil or error, got %v", result)
-				}
-			default:
-				t.Log("Channel empty after nil source, which is acceptable")
-			}
-		}
-	})
-}
-
-// BenchmarkConvertChan benchmarks the ConvertChan function
-func BenchmarkConvertChan(b *testing.B) {
-	ch := make(chan map[string]interface{}, 1)
-	src := map[string]interface{}{
-		"name":   "benchmark_test",
-		"value":  12345,
-		"active": true,
-	}
-
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		_ = ConvertChan(src, ch)
-		<-ch // Clear the channel
-	}
-}
-
 func TestRuntimeObject(t *testing.T) {
 	runtimeobject := NewRuntimeObject(nil)
 	utils.AssertNil(t, runtimeobject.IgnoreSSL)
@@ -506,7 +363,7 @@ type Test struct {
 
 func TestToMap(t *testing.T) {
 	inStr := map[string]string{
-		"tea":  "test",
+		"tea": "test",
 		"test": "test2",
 	}
 	result := ToMap(inStr)
@@ -514,7 +371,7 @@ func TestToMap(t *testing.T) {
 	utils.AssertEqual(t, "test2", result["test"])
 
 	inInt := map[string]int{
-		"tea":  12,
+		"tea": 12,
 		"test": 13,
 	}
 	result = ToMap(inInt)
@@ -658,7 +515,6 @@ func Test_GetBackoffTime(t *testing.T) {
 	ms = GetBackoffTime(backoff, Int(1))
 	utils.AssertEqual(t, true, IntValue(ms) <= 3)
 }
-
 type httpClient struct {
 	HttpClient
 	httpClient *http.Client
@@ -732,37 +588,12 @@ 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"
@@ -824,38 +655,42 @@ func Test_DoRequest(t *testing.T) {
 }
 
 func Test_DoRequestWithContextCancellation(t *testing.T) {
-	origTestHookDo := hookDo
-	defer func() { hookDo = origTestHookDo }()
+    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()
 
-	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Millisecond)
-	defer cancel()
+    // 创建请求
+    request := NewRequest()
+    request.Method = String("GET")
+    request.Protocol = String("http")
+    request.Headers["host"] = String("tea-cn-hangzhou.aliyuncs.com")
 
-	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) {
-			select {
-			case <-time.After(2 * time.Second):
-				return &http.Response{StatusCode: 200, Header: http.Header{}, Body: http.NoBody}, nil
-			case <-ctx.Done():
-				return nil, ctx.Err()
-			}
-		}
-	}
+    // 将上下文放入运行时对象
+    runtimeObj := map[string]interface{}{
+        "ctx": ctx,
+    }
+    runtimeObject := NewRuntimeObject(runtimeObj)
 
-	request := &Request{
-		Method:   String("GET"),
-		Protocol: String("http"),
-		Headers:  map[string]*string{"host": String("ecs.cn-hangzhou.aliyuncs.com")},
-	}
+	fmt.Printf("runtime: %v \n", runtimeObject)
+    resp, err := DoRequest(request, runtimeObject)
 
-	runtimeObject := &RuntimeObject{}
-
-	resp, err := DoRequestWithCtx(ctx, request, runtimeObject)
-
-	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")
+	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) {

dara/stream.go

@@ -3,67 +3,49 @@ package dara
 import (
 	"bufio"
 	"bytes"
-	"context"
 	"encoding/json"
-	"fmt"
 	"io"
 	"io/ioutil"
 	"strings"
+	"fmt"
 )
 
+// 定义 Event 结构体
 type SSEEvent struct {
-	Id    *string
+	ID    *string
 	Event *string
 	Data  *string
 	Retry *int
 }
 
+// 解析单个事件
 func parseEvent(lines []string) *SSEEvent {
 	event := &SSEEvent{}
 	for _, line := range lines {
 		if strings.HasPrefix(line, "data:") {
-			var data string
-			if strings.HasPrefix(line, "data: ") {
-				data = strings.TrimPrefix(line, "data: ") + "\n"
-			} else {
-				data = strings.TrimPrefix(line, "data:") + "\n"
-			}
+			data := strings.TrimPrefix(line, "data:") + "\n"
 			if event.Data == nil {
 				event.Data = new(string)
 			}
 			*event.Data += data
 		} else if strings.HasPrefix(line, "event:") {
-			var eventName string
-			if strings.HasPrefix(line, "event: ") {
-				eventName = strings.TrimPrefix(line, "event: ")
-			} else {
-				eventName = strings.TrimPrefix(line, "event:")
-			}
+			eventName := strings.TrimPrefix(line, "event:")
 			event.Event = &eventName
 		} else if strings.HasPrefix(line, "id:") {
-			var id string
-			if strings.HasPrefix(line, "id: ") {
-				id = strings.TrimPrefix(line, "id: ")
-			} else {
-				id = strings.TrimPrefix(line, "id:")
-			}
-			event.Id = &id
+			id := strings.TrimPrefix(line, "id:")
+			event.ID = &id
 		} else if strings.HasPrefix(line, "retry:") {
-			var retryStr string
-			if strings.HasPrefix(line, "retry: ") {
-				retryStr = strings.TrimPrefix(line, "retry: ")
-			} else {
-				retryStr = strings.TrimPrefix(line, "retry:")
-			}
 			var retry int
-			fmt.Sscanf(retryStr, "%d", &retry)
+			fmt.Sscanf(strings.TrimPrefix(line, "retry:"), "%d", &retry)
 			event.Retry = &retry
 		}
 	}
+	// Remove last newline from data
 	if event.Data != nil {
 		data := strings.TrimRight(*event.Data, "\n")
 		event.Data = &data
 	}
+	
 	return event
 }
 
@@ -151,65 +133,6 @@ func ReadAsSSE(body io.ReadCloser, eventChannel chan *SSEEvent, errorChannel cha
 			eventLines = append(eventLines, line)
 		}
 	}()
-}
-
-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)
-		}
-	}()
+
+	return
 }

tea/tea.go

@@ -495,7 +495,7 @@ func getHttpTransport(req *Request, runtime *RuntimeObject) (*http.Transport, er
 					Password: password,
 				}
 			}
-			dialer, err := proxy.SOCKS5(strings.ToLower(StringValue(runtime.Socks5NetWork)), socks5Proxy.Host, auth,
+			dialer, err := proxy.SOCKS5(strings.ToLower(StringValue(runtime.Socks5NetWork)), socks5Proxy.String(), auth,
 				&net.Dialer{
 					Timeout:   time.Duration(IntValue(runtime.ConnectTimeout)) * time.Millisecond,
 					DualStack: true,

tea/tea_test.go

@@ -560,37 +560,12 @@ 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"