Compare commits

..

1 Commits

Author SHA1 Message Date
weeping bbd3ea5344 add ReadAsSSEWithContext 2025-09-09 15:40:06 +08:00
9 changed files with 7 additions and 3043 deletions
+5 -143
View File
@@ -31,101 +31,8 @@ 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 {
@@ -176,11 +83,10 @@ type Request struct {
// Response is use d wrap http response
type Response struct {
Body io.ReadCloser
StatusCode *int
StatusMessage *string
Headers map[string]*string
WebSocketClient *DefaultWebSocketClient
Body io.ReadCloser
StatusCode *int
StatusMessage *string
Headers map[string]*string
}
// RuntimeObject is used for converting http configuration
@@ -188,7 +94,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"`
@@ -205,23 +110,11 @@ 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)) + 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
}
@@ -235,7 +128,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"]),
@@ -246,15 +138,6 @@ 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)
@@ -271,12 +154,6 @@ 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
}
@@ -390,12 +267,6 @@ 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")
}
@@ -515,12 +386,6 @@ 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")
}
@@ -708,9 +573,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
}
-308
View File
File diff suppressed because it is too large Load Diff
-485
View File
File diff suppressed because it is too large Load Diff
-1295
View File
File diff suppressed because it is too large Load Diff
-173
View File
@@ -1,173 +0,0 @@
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
}
-93
View File
@@ -1,93 +0,0 @@
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)
}
File diff suppressed because it is too large Load Diff
+1 -2
View File
@@ -4,9 +4,8 @@ go 1.14
require (
github.com/alibabacloud-go/debug v1.0.0
github.com/alibabacloud-go/tea-utils/v2 v2.0.9-0.20251128100122-896ae2c17fa8
github.com/alibabacloud-go/tea-utils/v2 v2.0.7
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
+1 -5
View File
@@ -3,8 +3,6 @@ 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=
@@ -13,8 +11,6 @@ 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=
@@ -39,13 +35,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=