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
@@ -111,6 +205,18 @@ 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 {
@@ -140,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)
@@ -156,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
 }
 
@@ -269,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")
 	}
@@ -388,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")
 	}

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=