feat: support websocket

dara/core.go

@@ -31,8 +31,101 @@ import (
 )
 
 type RuntimeOptions = util.RuntimeOptions
+
 type ExtendsParameters = util.ExtendsParameters
 
+func GetWebSocketPingInterval(runtime interface{}) *int {
+	if runtime == nil {
+		return nil
+	}
+	if rt, ok := runtime.(*RuntimeOptions); ok {
+		return rt.WebSocketPingInterval
+	}
+	return nil
+}
+
+func GetWebSocketPongTimeout(runtime interface{}) *int {
+	if runtime == nil {
+		return nil
+	}
+	if rt, ok := runtime.(*RuntimeOptions); ok {
+		return rt.WebSocketPongTimeout
+	}
+	return nil
+}
+
+func GetWebSocketEnableReconnect(runtime interface{}) *bool {
+	if runtime == nil {
+		return nil
+	}
+	if rt, ok := runtime.(*RuntimeOptions); ok {
+		return rt.WebSocketEnableReconnect
+	}
+	return nil
+}
+
+func GetWebSocketReconnectInterval(runtime interface{}) *int {
+	if runtime == nil {
+		return nil
+	}
+	if rt, ok := runtime.(*RuntimeOptions); ok {
+		return rt.WebSocketReconnectInterval
+	}
+	return nil
+}
+
+func GetWebSocketMaxReconnectTimes(runtime interface{}) *int {
+	if runtime == nil {
+		return nil
+	}
+	if rt, ok := runtime.(*RuntimeOptions); ok {
+		return rt.WebSocketMaxReconnectTimes
+	}
+	return nil
+}
+
+func GetWebSocketWriteTimeout(runtime interface{}) *int {
+	if runtime == nil {
+		return nil
+	}
+	if rt, ok := runtime.(*RuntimeOptions); ok {
+		return rt.WebSocketWriteTimeout
+	}
+	return nil
+}
+
+func GetWebSocketHandshakeTimeout(runtime interface{}) *int {
+	if runtime == nil {
+		return nil
+	}
+	if rt, ok := runtime.(*RuntimeOptions); ok {
+		return rt.WebSocketHandshakeTimeout
+	}
+	return nil
+}
+
+// GetWebSocketHandler safely extracts WebSocketHandler from runtime
+// Returns the handler as interface{} which can be type-asserted to WebSocketHandler
+func GetWebSocketHandler(runtime interface{}) interface{} {
+	if runtime == nil {
+		return nil
+	}
+	if rt, ok := runtime.(*RuntimeOptions); ok {
+		return rt.WebSocketHandler
+	}
+	return nil
+}
+
+func GetWebsocketSubProtocol(runtime interface{}) *string {
+	if runtime == nil {
+		return nil
+	}
+	if rt, ok := runtime.(*RuntimeObject); ok {
+		return rt.WebsocketSubProtocol
+	}
+	return nil
+}
+
 var debugLog = debug.Init("dara")
 
 type HttpRequest interface {
@@ -83,10 +176,11 @@ type Request struct {
 
 // Response is use d wrap http response
 type Response struct {
-	Body          io.ReadCloser
-	StatusCode    *int
-	StatusMessage *string
-	Headers       map[string]*string
+	Body            io.ReadCloser
+	StatusCode      *int
+	StatusMessage   *string
+	Headers         map[string]*string
+	WebSocketClient *DefaultWebSocketClient
 }
 
 // RuntimeObject is used for converting http configuration
@@ -94,6 +188,7 @@ 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"`
@@ -110,11 +205,23 @@ type RuntimeObject struct {
 	RetryOptions      *RetryOptions          `json:"retryOptions" xml:"retryOptions"`
 	ExtendsParameters *ExtendsParameters     `json:"extendsParameters,omitempty" xml:"extendsParameters,omitempty"`
 	HttpClient
+
+	// WebSocket-specific configuration
+	IsWebSocket                *bool       `json:"isWebSocket" xml:"isWebSocket"`
+	WebSocketPingInterval      *int        `json:"webSocketPingInterval" xml:"webSocketPingInterval"`
+	WebSocketPongTimeout       *int        `json:"webSocketPongTimeout" xml:"webSocketPongTimeout"`
+	WebSocketEnableReconnect   *bool       `json:"webSocketEnableReconnect" xml:"webSocketEnableReconnect"`
+	WebSocketReconnectInterval *int        `json:"webSocketReconnectInterval" xml:"webSocketReconnectInterval"`
+	WebSocketMaxReconnectTimes *int        `json:"webSocketMaxReconnectTimes" xml:"webSocketMaxReconnectTimes"`
+	WebSocketWriteTimeout      *int        `json:"webSocketWriteTimeout" xml:"webSocketWriteTimeout"`
+	WebSocketHandshakeTimeout  *int        `json:"webSocketHandshakeTimeout" xml:"webSocketHandshakeTimeout"`
+	WebSocketHandler           interface{} `json:"-" xml:"-"`                                                           // WebSocket handler (not serialized)
+	WebsocketSubProtocol       *string     `json:"websocketSubProtocol,omitempty" xml:"websocketSubProtocol,omitempty"` // WebSocket sub-protocol (awap or general)
 }
 
 func (r *RuntimeObject) getClientTag(domain string) string {
 	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
 }
 
@@ -128,6 +235,7 @@ 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"]),
@@ -138,6 +246,15 @@ func NewRuntimeObject(runtime map[string]interface{}) *RuntimeObject {
 		Key:            TransInterfaceToString(runtime["key"]),
 		Cert:           TransInterfaceToString(runtime["cert"]),
 		Ca:             TransInterfaceToString(runtime["ca"]),
+		// WebSocket-specific configuration
+		IsWebSocket:                TransInterfaceToBool(runtime["isWebSocket"]),
+		WebSocketPingInterval:      TransInterfaceToInt(runtime["webSocketPingInterval"]),
+		WebSocketPongTimeout:       TransInterfaceToInt(runtime["webSocketPongTimeout"]),
+		WebSocketEnableReconnect:   TransInterfaceToBool(runtime["webSocketEnableReconnect"]),
+		WebSocketReconnectInterval: TransInterfaceToInt(runtime["webSocketReconnectInterval"]),
+		WebSocketMaxReconnectTimes: TransInterfaceToInt(runtime["webSocketMaxReconnectTimes"]),
+		WebSocketWriteTimeout:      TransInterfaceToInt(runtime["webSocketWriteTimeout"]),
+		WebSocketHandshakeTimeout:  TransInterfaceToInt(runtime["webSocketHandshakeTimeout"]),
 	}
 	if runtime["listener"] != nil {
 		runtimeObject.Listener = runtime["listener"].(utils.ProgressListener)
@@ -154,6 +271,12 @@ func NewRuntimeObject(runtime map[string]interface{}) *RuntimeObject {
 	if runtime["retryOptions"] != nil {
 		runtimeObject.RetryOptions = runtime["retryOptions"].(*RetryOptions)
 	}
+	if runtime["webSocketHandler"] != nil {
+		runtimeObject.WebSocketHandler = runtime["webSocketHandler"]
+	}
+	if runtime["websocketSubProtocol"] != nil {
+		runtimeObject.WebsocketSubProtocol = runtime["websocketSubProtocol"].(*string)
+	}
 	return runtimeObject
 }
 
@@ -184,6 +307,39 @@ 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 {
@@ -234,6 +390,12 @@ func DoRequest(request *Request, runtimeObject *RuntimeObject) (response *Respon
 			runtimeObject.Logger.PrintLog(fieldMap, err)
 		}
 	}()
+
+	if runtimeObject.IsWebSocket != nil && *runtimeObject.IsWebSocket {
+		// WebSocket request
+		return newWebSocketClientAndConnect(request, runtimeObject)
+	}
+
 	if request.Method == nil {
 		request.Method = String("GET")
 	}
@@ -353,6 +515,12 @@ func DoRequestWithCtx(ctx context.Context, request *Request, runtimeObject *Runt
 			runtimeObject.Logger.PrintLog(fieldMap, err)
 		}
 	}()
+
+	if runtimeObject.IsWebSocket != nil && *runtimeObject.IsWebSocket {
+		// WebSocket request
+		return newWebSocketClientAndConnect(request, runtimeObject)
+	}
+
 	if request.Method == nil {
 		request.Method = String("GET")
 	}
@@ -540,6 +708,9 @@ 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

@@ -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)

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=