feat: support websocket

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=