feat: support websocket

dara/core_test.go

@@ -193,6 +193,150 @@ 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)
@@ -361,11 +505,27 @@ type Test struct {
 }
 
 func TestToMap(t *testing.T) {
+	inStr := map[string]string{
+		"tea":  "test",
+		"test": "test2",
+	}
+	result := ToMap(inStr)
+	utils.AssertEqual(t, "test", result["tea"])
+	utils.AssertEqual(t, "test2", result["test"])
+
+	inInt := map[string]int{
+		"tea":  12,
+		"test": 13,
+	}
+	result = ToMap(inInt)
+	utils.AssertEqual(t, 12, result["tea"])
+	utils.AssertEqual(t, 13, result["test"])
+
 	in := map[string]*string{
 		"tea": String("test"),
 		"nil": nil,
 	}
-	result := ToMap(in)
+	result = ToMap(in)
 	utils.AssertEqual(t, "test", result["tea"])
 	utils.AssertNil(t, result["nil"])
 
@@ -498,6 +658,7 @@ 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
@@ -637,6 +798,41 @@ 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 }()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Millisecond)
+	defer cancel()
+
+	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()
+			}
+		}
+	}
+
+	request := &Request{
+		Method:   String("GET"),
+		Protocol: String("http"),
+		Headers:  map[string]*string{"host": String("ecs.cn-hangzhou.aliyuncs.com")},
+	}
+
+	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")
+}
+
 func Test_DoRequestWithConcurrent(t *testing.T) {
 	origTestHookDo := hookDo
 	defer func() { hookDo = origTestHookDo }()
@@ -668,6 +864,53 @@ 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")

dara/json_parser.go

@@ -6,6 +6,7 @@ import (
 	jsoniter "github.com/json-iterator/go"
 	"github.com/modern-go/reflect2"
 	"io"
+	"io/ioutil"
 	"math"
 	"reflect"
 	"strconv"
@@ -332,9 +333,28 @@ func (decoder *nullableFuzzyFloat64Decoder) Decode(ptr unsafe.Pointer, iter *jso
 	}
 }
 
-func Stringify(m interface{}) string {
-	byt, _ := json.Marshal(m)
-	return string(byt)
+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 ParseJSON(a string) interface{} {

dara/json_parser_test.go

@@ -3,6 +3,7 @@ package dara
 import (
 	"reflect"
 	"testing"
+	"strings"
 
 	"github.com/alibabacloud-go/tea/utils"
 	jsoniter "github.com/json-iterator/go"
@@ -878,3 +879,42 @@ 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)
+}

dara/stream.go

@@ -3,44 +3,63 @@ 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: ") {
-			data := strings.TrimPrefix(line, "data: ") + "\n"
+		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"
+			}
 			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:") {
+			var eventName string
+			if strings.HasPrefix(line, "event: ") {
+				eventName = strings.TrimPrefix(line, "event: ")
+			} else {
+				eventName = strings.TrimPrefix(line, "event:")
+			}
 			event.Event = &eventName
-		} 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, "id:") {
+			var id string
+			if strings.HasPrefix(line, "id: ") {
+				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
-			fmt.Sscanf(strings.TrimPrefix(line, "retry: "), "%d", &retry)
+			fmt.Sscanf(retryStr, "%d", &retry)
 			event.Retry = &retry
 		}
 	}
-	// Remove last newline from data
 	if event.Data != nil {
 		data := strings.TrimRight(*event.Data, "\n")
 		event.Data = &data
@@ -132,6 +151,65 @@ func ReadAsSSE(body io.ReadCloser, eventChannel chan *SSEEvent, errorChannel cha
 			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)
+		}
+	}()
 }

dara/websocket_awap.go

@@ -0,0 +1,173 @@
+package dara
+
+import (
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+type AwapMessageType string
+type AwapMessageFormat string
+
+const (
+	AwapMessageFormatText   AwapMessageFormat = "text"
+	AwapMessageFormatBinary AwapMessageFormat = "binary"
+)
+
+type AwapMessage struct {
+	// only type and id is required for awap message, and id can be autofilled when sending message
+	// other fields are optional and can be set using withHeader
+	Type    AwapMessageType        `json:"type"`
+	ID      string                 `json:"id"`
+	Headers map[string]string      `json:"headers,omitempty"`
+	Payload interface{}            `json:"payload,omitempty"`
+	Format  AwapMessageFormat      `json:"format,omitempty"`
+	Status  int                    `json:"status,omitempty"`
+	Error   string                 `json:"error,omitempty"`
+	Data    map[string]interface{} `json:"data,omitempty"`
+}
+
+func (m *AwapMessage) ToJSON() ([]byte, error) {
+	return json.Marshal(m)
+}
+
+func (m *AwapMessage) WithHeader(key, value string) *AwapMessage {
+	if m.Headers == nil {
+		m.Headers = make(map[string]string)
+	}
+	m.Headers[key] = value
+	return m
+}
+
+func (m *AwapMessage) WithFormat(format AwapMessageFormat) *AwapMessage {
+	m.Format = format
+	return m
+}
+
+type AwapWebSocketHandler interface {
+	WebSocketHandler
+
+	HandleAwapMessage(session *WebSocketSessionInfo, message *AwapMessage) error
+}
+
+type AbstractAwapWebSocketHandler struct {
+	supportsPartial bool
+}
+
+func (h *AbstractAwapWebSocketHandler) AfterConnectionEstablished(session *WebSocketSessionInfo) error {
+	return nil
+}
+
+func (h *AbstractAwapWebSocketHandler) HandleRawMessage(session *WebSocketSessionInfo, message *WebSocketMessage) error {
+	// This method is only called if:
+	// 1. DefaultWebSocketClient.readMessages() doesn't recognize the handler as AwapWebSocketHandler, OR GeneralWebSocketHandler
+	// 2. User explicitly overrides this method for custom handling
+	//
+	// In normal AWAP protocol usage, readMessages() will directly call HandleAwapMessage,
+	// so this default implementation won't be called.
+	//
+	// If you need custom protocol handling, override this method in your handler.
+	return nil
+}
+
+// ErrUseRawMessage is a sentinel error that indicates HandleRawMessage should be used instead
+var ErrUseRawMessage = fmt.Errorf("use HandleRawMessage")
+
+func (h *AbstractAwapWebSocketHandler) HandleAwapMessage(session *WebSocketSessionInfo, message *AwapMessage) error {
+	// Default implementation returns ErrUseRawMessage to indicate HandleRawMessage should be used
+	// If user overrides this method, they should return nil or their own error (not ErrUseRawMessage)
+	return ErrUseRawMessage
+}
+
+func (h *AbstractAwapWebSocketHandler) HandleError(session *WebSocketSessionInfo, err error) error {
+	return nil
+}
+
+func (h *AbstractAwapWebSocketHandler) AfterConnectionClosed(session *WebSocketSessionInfo, code int, reason string) error {
+	return nil
+}
+
+func (h *AbstractAwapWebSocketHandler) SupportsPartialMessages() bool {
+	return h.supportsPartial
+}
+
+func (h *AbstractAwapWebSocketHandler) SetSupportsPartialMessages(supports bool) {
+	h.supportsPartial = supports
+}
+
+// ParseAwapMessage parses a WebSocket message as AWAP format
+// AWAP protocol uses frame format: text headers + JSON payload
+// Format: "type:request\ntimestamp:1234567890\nid:msg-001\n\n{JSON payload} or [header bytes (text converted to binary)] + [\n\n separator] + [binary body]"
+func ParseAwapMessage(message *WebSocketMessage) (*AwapMessage, error) {
+	data := message.Payload
+
+	// Check if it's awap format (has \n\n separator)
+	headerEndIndex := -1
+	for i := 0; i < len(data)-1; i++ {
+		if data[i] == '\n' && data[i+1] == '\n' {
+			headerEndIndex = i
+			break
+		}
+	}
+
+	awapMsg := &AwapMessage{
+		Headers: make(map[string]string),
+	}
+
+	if headerEndIndex == -1 {
+		return nil, fmt.Errorf("failed to parse AWAP message: no \n\n separator found")
+	}
+
+	// Frame format: parse headers and payload separately
+	headerBytes := data[:headerEndIndex]
+	payloadBytes := data[headerEndIndex+2:] // Skip \n\n
+
+	// Parse headers (key:value format)
+	headerStr := string(headerBytes)
+	headerLines := strings.Split(headerStr, "\n")
+	for _, line := range headerLines {
+		line = strings.TrimSpace(line)
+		if line == "" {
+			continue
+		}
+		colonIndex := strings.Index(line, ":")
+		if colonIndex > 0 {
+			key := strings.TrimSpace(line[:colonIndex])
+			value := strings.TrimSpace(line[colonIndex+1:])
+			// Extract AWAP-required fields
+			switch key {
+			case "type":
+				awapMsg.Type = AwapMessageType(value)
+			case "id":
+				awapMsg.ID = value
+			case "status":
+				if status, err := strconv.Atoi(value); err == nil {
+					awapMsg.Status = status
+				}
+			case "error":
+				awapMsg.Error = value
+			default:
+				// Only non-AWAP-required fields go into Headers map
+				awapMsg.Headers[key] = value
+			}
+		}
+	}
+
+	if message.Type == WebSocketMessageTypeBinary {
+		awapMsg.Format = AwapMessageFormatBinary
+		awapMsg.Payload = payloadBytes
+	} else {
+		awapMsg.Format = AwapMessageFormatText
+		if len(payloadBytes) > 0 {
+			var payload interface{}
+			if err := json.Unmarshal(payloadBytes, &payload); err != nil {
+				awapMsg.Payload = payloadBytes
+			} else {
+				awapMsg.Payload = payload
+			}
+		}
+	}
+
+	return awapMsg, nil
+}

dara/websocket_general.go

@@ -0,0 +1,93 @@
+package dara
+
+import (
+	"encoding/json"
+)
+
+type GeneralMessageFormat string
+
+const (
+	GeneralMessageFormatText   GeneralMessageFormat = "text"
+	GeneralMessageFormatBinary GeneralMessageFormat = "binary"
+)
+
+type GeneralMessage struct {
+	Body   interface{}          `json:"body,omitempty"`
+	Format GeneralMessageFormat `json:"format,omitempty"`
+}
+
+type GeneralWebSocketHandler interface {
+	WebSocketHandler
+
+	HandleGeneralMessage(session *WebSocketSessionInfo, message *GeneralMessage) error
+}
+
+type AbstractGeneralWebSocketHandler struct {
+	supportsPartial bool
+}
+
+func (h *AbstractGeneralWebSocketHandler) AfterConnectionEstablished(session *WebSocketSessionInfo) error {
+	return nil
+}
+
+func (h *AbstractGeneralWebSocketHandler) HandleRawMessage(session *WebSocketSessionInfo, message *WebSocketMessage) error {
+	// This method is only called if:
+	// 1. DefaultWebSocketClient.readMessages() doesn't recognize the handler as GeneralWebSocketHandler or AwapWebSocketHandler,
+	// 2. User explicitly overrides this method for custom handling
+	//
+	// In normal General protocol usage, readMessages() will directly call HandleGeneralTextMessage/HandleGeneralBinaryMessage,
+	// so this default implementation won't be called.
+	//
+	// If you need custom protocol handling, override this method in your handler.
+	return nil
+}
+
+func (h *AbstractGeneralWebSocketHandler) HandleGeneralMessage(session *WebSocketSessionInfo, message *GeneralMessage) error {
+	// Default implementation returns ErrUseRawMessage to indicate HandleRawMessage should be used
+	// If user overrides this method, they should return nil or their own error (not ErrUseRawMessage)
+	return ErrUseRawMessage
+}
+
+func (h *AbstractGeneralWebSocketHandler) HandleError(session *WebSocketSessionInfo, err error) error {
+	return nil
+}
+
+func (h *AbstractGeneralWebSocketHandler) AfterConnectionClosed(session *WebSocketSessionInfo, code int, reason string) error {
+	return nil
+}
+
+func (h *AbstractGeneralWebSocketHandler) SupportsPartialMessages() bool {
+	return h.supportsPartial
+}
+
+func (h *AbstractGeneralWebSocketHandler) SetSupportsPartialMessages(supports bool) {
+	h.supportsPartial = supports
+}
+
+func ParseGeneralMessage(message *WebSocketMessage) (*GeneralMessage, error) {
+	if message.Type == WebSocketMessageTypeBinary {
+		return &GeneralMessage{
+			Body:   message.Payload,
+			Format: GeneralMessageFormatBinary,
+		}, nil
+	}
+	// Try to parse the entire JSON payload as the body
+	// For text messages, payload is already a string (UTF-8 encoded bytes)
+	// If JSON parsing fails, return the original bytes as []byte to preserve the data
+	var body interface{}
+	if err := json.Unmarshal(message.Payload, &body); err != nil {
+		return &GeneralMessage{
+			Body:   message.Payload,
+			Format: GeneralMessageFormatText,
+		}, nil
+	}
+	// Successfully parsed as JSON, return the parsed object
+	return &GeneralMessage{
+		Body:   body,
+		Format: GeneralMessageFormatText,
+	}, nil
+}
+
+func (m *GeneralMessage) ToJSON() ([]byte, error) {
+	return json.Marshal(m)
+}

go.mod

@@ -4,8 +4,9 @@ go 1.14
 
 require (
 	github.com/alibabacloud-go/debug v1.0.0
-	github.com/alibabacloud-go/tea-utils/v2 v2.0.7
+	github.com/alibabacloud-go/tea-utils/v2 v2.0.9-0.20251128100122-896ae2c17fa8
 	github.com/clbanning/mxj/v2 v2.7.0
+	github.com/gorilla/websocket v1.5.3
 	github.com/json-iterator/go v1.1.12
 	github.com/modern-go/reflect2 v1.0.2
 	golang.org/x/net v0.26.0

go.sum

@@ -3,6 +3,8 @@ github.com/alibabacloud-go/debug v1.0.0/go.mod h1:8gfgZCCAC3+SCzjWtY053FrOcd4/ql
 github.com/alibabacloud-go/tea v1.2.2/go.mod h1:CF3vOzEMAG+bR4WOql8gc2G9H3EkH3ZLAQdpmpXMgwk=
 github.com/alibabacloud-go/tea-utils/v2 v2.0.7 h1:WDx5qW3Xa5ZgJ1c8NfqJkF6w+AU5wB8835UdhPr6Ax0=
 github.com/alibabacloud-go/tea-utils/v2 v2.0.7/go.mod h1:qxn986l+q33J5VkialKMqT/TTs3E+U9MJpd001iWQ9I=
+github.com/alibabacloud-go/tea-utils/v2 v2.0.9-0.20251128100122-896ae2c17fa8 h1:7DpaFH4RdhUZ9Sduawgv1lvXkczXB4T4XzDVANxameA=
+github.com/alibabacloud-go/tea-utils/v2 v2.0.9-0.20251128100122-896ae2c17fa8/go.mod h1:qxn986l+q33J5VkialKMqT/TTs3E+U9MJpd001iWQ9I=
 github.com/clbanning/mxj/v2 v2.7.0 h1:WA/La7UGCanFe5NpHF0Q3DNtnCsVoxbPKuyBNHWRyME=
 github.com/clbanning/mxj/v2 v2.7.0/go.mod h1:hNiWqW14h+kc+MdF9C6/YoRfjEJoR3ou6tn/Qo+ve2s=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -11,6 +13,8 @@ github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 h1:ZqeYNhU3OHLH3mGKHDcjJRFFRrJa6eAM5H+CtDdOsPc=
@@ -35,13 +39,13 @@ golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/mod v0.15.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
 golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
-golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
 golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk=
+golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY=
 golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
 golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM=
 golang.org/x/net v0.26.0 h1:soB7SVo0PWrY4vPW/+ay0jKDNScG2X9wFeYlXIvJsOQ=