Compare commits
1 Commits
v1.3.4
..
fix-reader
| Author | SHA1 | Date | |
|---|---|---|---|
| abae7f41c3 |
+2
-22
@@ -294,28 +294,8 @@ func ConcatArr(arr1 interface{}, arr2 interface{}) interface{} {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// ArrAppend inserts a new pointer at a specified index in a pointer array.
|
// ArrAppend inserts a new pointer at a specified index in a pointer array.
|
||||||
func ArrAppend(arr interface{}, value interface{}, index int) {
|
func ArrAppend(arr interface{}, value interface{}) {
|
||||||
arrV := reflect.ValueOf(arr)
|
ArrPush(arr, value)
|
||||||
if arrV.Kind() != reflect.Ptr || arrV.Elem().Kind() != reflect.Slice {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
sliceV := arrV.Elem()
|
|
||||||
|
|
||||||
if index < 0 || index > sliceV.Len() {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
valueV := reflect.ValueOf(value)
|
|
||||||
|
|
||||||
// 创建一个容纳新值的切片
|
|
||||||
newSlice := reflect.Append(sliceV, reflect.Zero(sliceV.Type().Elem()))
|
|
||||||
reflect.Copy(newSlice.Slice(index+1, newSlice.Len()), newSlice.Slice(index, newSlice.Len()-1))
|
|
||||||
newSlice.Index(index).Set(valueV)
|
|
||||||
|
|
||||||
// 更新原始切片
|
|
||||||
sliceV.Set(newSlice)
|
|
||||||
return
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// ArrRemove removes an element from the array
|
// ArrRemove removes an element from the array
|
||||||
|
|||||||
@@ -405,54 +405,6 @@ func TestConcatArr(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestArrAppend(t *testing.T) {
|
|
||||||
// 测试用例1:插入中间位置
|
|
||||||
t.Run("Append to middle of an array", func(t *testing.T) {
|
|
||||||
numbers := []*int{new(int), new(int), new(int)}
|
|
||||||
for i := range numbers {
|
|
||||||
*numbers[i] = i + 1
|
|
||||||
}
|
|
||||||
|
|
||||||
// 将 9 插入到索引 1
|
|
||||||
valueToInsert := new(int)
|
|
||||||
*valueToInsert = 9
|
|
||||||
|
|
||||||
// 期望的结果
|
|
||||||
expected := []*int{new(int), new(int), new(int), new(int)}
|
|
||||||
*expected[0], *expected[1], *expected[2], *expected[3] = 1, 9, 2, 3
|
|
||||||
|
|
||||||
defer func() {
|
|
||||||
if !reflect.DeepEqual(numbers, expected) {
|
|
||||||
t.Errorf("Expected %+v, but got %+v", expected, numbers)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
ArrAppend(&numbers, valueToInsert, 1)
|
|
||||||
})
|
|
||||||
|
|
||||||
// 测试用例2: 尝试在越界处插入
|
|
||||||
t.Run("Index out of bounds", func(t *testing.T) {
|
|
||||||
numbers := []*int{new(int), new(int), new(int)}
|
|
||||||
for i := range numbers {
|
|
||||||
*numbers[i] = i + 1
|
|
||||||
}
|
|
||||||
|
|
||||||
defer func() {
|
|
||||||
// Defer 检查:越界情况下,数组应保持不变
|
|
||||||
expected := []*int{new(int), new(int), new(int)}
|
|
||||||
*expected[0], *expected[1], *expected[2] = 1, 2, 3
|
|
||||||
if !reflect.DeepEqual(numbers, expected) {
|
|
||||||
t.Errorf("Index out of bounds should not modify array, but got %+v", numbers)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
valueToInsert := new(int)
|
|
||||||
*valueToInsert = 9
|
|
||||||
ArrAppend(&numbers, valueToInsert, 10) // 超出范围
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
func TestArrRemove(t *testing.T) {
|
func TestArrRemove(t *testing.T) {
|
||||||
// Create test data for string pointer array
|
// Create test data for string pointer array
|
||||||
str1 := "A"
|
str1 := "A"
|
||||||
|
|||||||
+2
-28
@@ -24,8 +24,8 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/alibabacloud-go/debug/debug"
|
"github.com/alibabacloud-go/debug/debug"
|
||||||
util "github.com/alibabacloud-go/tea-utils/v2/service"
|
|
||||||
"github.com/alibabacloud-go/tea/utils"
|
"github.com/alibabacloud-go/tea/utils"
|
||||||
|
util "github.com/alibabacloud-go/tea-utils/v2/service"
|
||||||
|
|
||||||
"golang.org/x/net/proxy"
|
"golang.org/x/net/proxy"
|
||||||
)
|
)
|
||||||
@@ -202,7 +202,7 @@ func getDaraClient(tag string) *daraClient {
|
|||||||
|
|
||||||
// DoRequest is used send request to server
|
// DoRequest is used send request to server
|
||||||
func DoRequest(request *Request, runtimeObject *RuntimeObject) (response *Response, err error) {
|
func DoRequest(request *Request, runtimeObject *RuntimeObject) (response *Response, err error) {
|
||||||
if runtimeObject == nil {
|
if(runtimeObject == nil) {
|
||||||
runtimeObject = &RuntimeObject{}
|
runtimeObject = &RuntimeObject{}
|
||||||
}
|
}
|
||||||
fieldMap := make(map[string]string)
|
fieldMap := make(map[string]string)
|
||||||
@@ -429,32 +429,6 @@ func ToReader(obj interface{}) io.Reader {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func ToWriter(obj interface{}) io.Writer {
|
|
||||||
switch obj.(type) {
|
|
||||||
case string:
|
|
||||||
var buf bytes.Buffer
|
|
||||||
buf.WriteString(obj.(string))
|
|
||||||
return &buf
|
|
||||||
case *string:
|
|
||||||
var buf bytes.Buffer
|
|
||||||
tmp := obj.(*string)
|
|
||||||
buf.WriteString(*tmp)
|
|
||||||
return &buf
|
|
||||||
case []byte:
|
|
||||||
var buf bytes.Buffer
|
|
||||||
buf.Write(obj.([]byte))
|
|
||||||
return &buf
|
|
||||||
case io.Writer:
|
|
||||||
return obj.(io.Writer)
|
|
||||||
case *bytes.Buffer:
|
|
||||||
return obj.(*bytes.Buffer)
|
|
||||||
case *os.File:
|
|
||||||
return obj.(*os.File)
|
|
||||||
default:
|
|
||||||
panic("Invalid Writer. Please provide a valid Writer.")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func ToString(val interface{}) string {
|
func ToString(val interface{}) string {
|
||||||
switch v := val.(type) {
|
switch v := val.(type) {
|
||||||
case []byte:
|
case []byte:
|
||||||
|
|||||||
@@ -735,42 +735,6 @@ func Test_ToReader(t *testing.T) {
|
|||||||
utils.AssertEqual(t, "", string(byt))
|
utils.AssertEqual(t, "", string(byt))
|
||||||
}
|
}
|
||||||
|
|
||||||
func Test_ToWriter(t *testing.T) {
|
|
||||||
str := "abc"
|
|
||||||
writer := ToWriter(str).(*bytes.Buffer)
|
|
||||||
utils.AssertEqual(t, "abc", writer.String())
|
|
||||||
|
|
||||||
strPtr := new(string)
|
|
||||||
*strPtr = "def"
|
|
||||||
writer = ToWriter(strPtr).(*bytes.Buffer)
|
|
||||||
utils.AssertEqual(t, "def", writer.String())
|
|
||||||
|
|
||||||
bytesData := []byte("ghi")
|
|
||||||
writer = ToWriter(bytesData).(*bytes.Buffer)
|
|
||||||
utils.AssertEqual(t, "ghi", writer.String())
|
|
||||||
|
|
||||||
buffer := new(bytes.Buffer)
|
|
||||||
writer = ToWriter(buffer).(*bytes.Buffer)
|
|
||||||
utils.AssertEqual(t, buffer, writer)
|
|
||||||
|
|
||||||
fileWriter := ToWriter(os.Stdout)
|
|
||||||
utils.AssertEqual(t, os.Stdout, fileWriter)
|
|
||||||
|
|
||||||
var buf bytes.Buffer
|
|
||||||
writer2 := ToWriter(&buf)
|
|
||||||
writer2.Write([]byte("test"))
|
|
||||||
utils.AssertEqual(t, "test", buf.String())
|
|
||||||
|
|
||||||
// Test a non-writer to trigger panic
|
|
||||||
defer func() {
|
|
||||||
if r := recover(); r != nil {
|
|
||||||
utils.AssertEqual(t, "Invalid Writer. Please provide a valid Writer.", r)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
num := 10
|
|
||||||
ToWriter(num) // This should cause a panic
|
|
||||||
}
|
|
||||||
|
|
||||||
func Test_ToString(t *testing.T) {
|
func Test_ToString(t *testing.T) {
|
||||||
str := ToString(10)
|
str := ToString(10)
|
||||||
utils.AssertEqual(t, "10", str)
|
utils.AssertEqual(t, "10", str)
|
||||||
|
|||||||
+5
-40
@@ -7,19 +7,18 @@ import (
|
|||||||
"net/http"
|
"net/http"
|
||||||
"reflect"
|
"reflect"
|
||||||
"strconv"
|
"strconv"
|
||||||
"github.com/alibabacloud-go/tea/tea"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
type BaseError interface {
|
type BaseError interface {
|
||||||
error
|
error
|
||||||
GetName() *string
|
ErrorName() *string
|
||||||
GetCode() *string
|
ErrorCode() *string
|
||||||
}
|
}
|
||||||
|
|
||||||
type ResponseError interface {
|
type ResponseError interface {
|
||||||
BaseError
|
BaseError
|
||||||
GetRetryAfter() *int64
|
ErrorRetryAfter() *int
|
||||||
GetStatusCode() *int
|
ErrorStatusCode() *int
|
||||||
}
|
}
|
||||||
|
|
||||||
// SDKError struct is used save error code and message
|
// SDKError struct is used save error code and message
|
||||||
@@ -41,40 +40,6 @@ type CastError struct {
|
|||||||
Message *string
|
Message *string
|
||||||
}
|
}
|
||||||
|
|
||||||
func TeaSDKError(err error) error {
|
|
||||||
if(err == nil) {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
if te, ok := err.(*SDKError); ok {
|
|
||||||
return tea.NewSDKError(map[string]interface{}{
|
|
||||||
"code": StringValue(te.Code),
|
|
||||||
"statusCode": IntValue(te.StatusCode),
|
|
||||||
"message": StringValue(te.Message),
|
|
||||||
"data": te.Data,
|
|
||||||
"description": StringValue(te.Description),
|
|
||||||
"accessDeniedDetail": te.AccessDeniedDetail,
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
if respErr, ok := err.(ResponseError); ok {
|
|
||||||
return tea.NewSDKError(map[string]interface{}{
|
|
||||||
"code": StringValue(respErr.GetCode()),
|
|
||||||
"statusCode": IntValue(respErr.GetStatusCode()),
|
|
||||||
"message": respErr.Error(),
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
if baseErr, ok := err.(BaseError); ok {
|
|
||||||
return tea.NewSDKError(map[string]interface{}{
|
|
||||||
"code": StringValue(baseErr.GetCode()),
|
|
||||||
"message": baseErr.Error(),
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// NewSDKError is used for shortly create SDKError object
|
// NewSDKError is used for shortly create SDKError object
|
||||||
func NewSDKError(obj map[string]interface{}) *SDKError {
|
func NewSDKError(obj map[string]interface{}) *SDKError {
|
||||||
err := &SDKError{}
|
err := &SDKError{}
|
||||||
@@ -155,7 +120,7 @@ func (err *SDKError) ErrorMessage() *string {
|
|||||||
return err.Message
|
return err.Message
|
||||||
}
|
}
|
||||||
|
|
||||||
func (err *SDKError) GetCode() *string {
|
func (err *SDKError) ErrorCode() *string {
|
||||||
return err.Code
|
return err.Code
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+10
-10
@@ -273,12 +273,12 @@ func ShouldRetry(options *RetryOptions, ctx *RetryPolicyContext) bool {
|
|||||||
|
|
||||||
for _, condition := range conditions {
|
for _, condition := range conditions {
|
||||||
for _, exc := range condition.Exception {
|
for _, exc := range condition.Exception {
|
||||||
if exc == StringValue(baseErr.GetName()) {
|
if exc == StringValue(baseErr.ErrorName()) {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
for _, code := range condition.ErrorCode {
|
for _, code := range condition.ErrorCode {
|
||||||
if code == StringValue(baseErr.GetCode()) {
|
if code == StringValue(baseErr.ErrorCode()) {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -287,7 +287,7 @@ func ShouldRetry(options *RetryOptions, ctx *RetryPolicyContext) bool {
|
|||||||
conditions = options.RetryCondition
|
conditions = options.RetryCondition
|
||||||
for _, condition := range conditions {
|
for _, condition := range conditions {
|
||||||
for _, exc := range condition.Exception {
|
for _, exc := range condition.Exception {
|
||||||
if exc == StringValue(baseErr.GetName()) {
|
if exc == StringValue(baseErr.ErrorName()) {
|
||||||
if retriesAttempted >= condition.MaxAttempts {
|
if retriesAttempted >= condition.MaxAttempts {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
@@ -295,7 +295,7 @@ func ShouldRetry(options *RetryOptions, ctx *RetryPolicyContext) bool {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
for _, code := range condition.ErrorCode {
|
for _, code := range condition.ErrorCode {
|
||||||
if code == StringValue(baseErr.GetCode()) {
|
if code == StringValue(baseErr.ErrorCode()) {
|
||||||
if retriesAttempted >= condition.MaxAttempts {
|
if retriesAttempted >= condition.MaxAttempts {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
@@ -323,13 +323,13 @@ func GetBackoffDelay(options *RetryOptions, ctx *RetryPolicyContext) int {
|
|||||||
if baseErr, ok := ex.(BaseError); ok {
|
if baseErr, ok := ex.(BaseError); ok {
|
||||||
for _, condition := range conditions {
|
for _, condition := range conditions {
|
||||||
for _, exc := range condition.Exception {
|
for _, exc := range condition.Exception {
|
||||||
if exc == StringValue(baseErr.GetName()) {
|
if exc == StringValue(baseErr.ErrorName()) {
|
||||||
maxDelay := condition.MaxDelay
|
maxDelay := condition.MaxDelay
|
||||||
// Simulated "retryAfter" from an error response
|
// Simulated "retryAfter" from an error response
|
||||||
if respErr, ok := ex.(ResponseError); ok {
|
if respErr, ok := ex.(ResponseError); ok {
|
||||||
retryAfter := Int64Value(respErr.GetRetryAfter())
|
retryAfter := IntValue(respErr.ErrorRetryAfter())
|
||||||
if retryAfter != 0 {
|
if retryAfter != 0 {
|
||||||
return min(int(retryAfter), maxDelay)
|
return min(retryAfter, maxDelay)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// This would be set properly based on your error handling
|
// This would be set properly based on your error handling
|
||||||
@@ -342,13 +342,13 @@ func GetBackoffDelay(options *RetryOptions, ctx *RetryPolicyContext) int {
|
|||||||
}
|
}
|
||||||
|
|
||||||
for _, code := range condition.ErrorCode {
|
for _, code := range condition.ErrorCode {
|
||||||
if code == StringValue(baseErr.GetCode()) {
|
if code == StringValue(baseErr.ErrorCode()) {
|
||||||
maxDelay := condition.MaxDelay
|
maxDelay := condition.MaxDelay
|
||||||
// Simulated "retryAfter" from an error response
|
// Simulated "retryAfter" from an error response
|
||||||
if respErr, ok := ex.(ResponseError); ok {
|
if respErr, ok := ex.(ResponseError); ok {
|
||||||
retryAfter := Int64Value(respErr.GetRetryAfter())
|
retryAfter := IntValue(respErr.ErrorRetryAfter())
|
||||||
if retryAfter != 0 {
|
if retryAfter != 0 {
|
||||||
return min(int(retryAfter), maxDelay)
|
return min(retryAfter, maxDelay)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+12
-12
@@ -29,11 +29,11 @@ func (err *AErr) New(obj map[string]interface{}) *AErr {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (err *AErr) GetCode() *string {
|
func (err *AErr) ErrorCode() *string {
|
||||||
return err.Code
|
return err.Code
|
||||||
}
|
}
|
||||||
|
|
||||||
func (err *AErr) GetName() *string {
|
func (err *AErr) ErrorName() *string {
|
||||||
return err.Name
|
return err.Name
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -59,11 +59,11 @@ func (err *BErr) New(obj map[string]interface{}) *BErr {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (err *BErr) GetCode() *string {
|
func (err *BErr) ErrorCode() *string {
|
||||||
return err.Code
|
return err.Code
|
||||||
}
|
}
|
||||||
|
|
||||||
func (err *BErr) GetName() *string {
|
func (err *BErr) ErrorName() *string {
|
||||||
return err.Name
|
return err.Name
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -72,7 +72,7 @@ type CErr struct {
|
|||||||
Code *string
|
Code *string
|
||||||
Name *string
|
Name *string
|
||||||
Message *string
|
Message *string
|
||||||
RetryAfter *int64
|
RetryAfter *int
|
||||||
StatusCode *int
|
StatusCode *int
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -91,26 +91,26 @@ func (err *CErr) New(obj map[string]interface{}) *CErr {
|
|||||||
err.StatusCode = Int(statusCode)
|
err.StatusCode = Int(statusCode)
|
||||||
}
|
}
|
||||||
|
|
||||||
if retryAfter, ok := obj["RetryAfter"].(int64); ok {
|
if retryAfter, ok := obj["RetryAfter"].(int); ok {
|
||||||
err.RetryAfter = Int64(retryAfter)
|
err.RetryAfter = Int(retryAfter)
|
||||||
}
|
}
|
||||||
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (err *CErr) GetCode() *string {
|
func (err *CErr) ErrorCode() *string {
|
||||||
return err.Code
|
return err.Code
|
||||||
}
|
}
|
||||||
|
|
||||||
func (err *CErr) GetName() *string {
|
func (err *CErr) ErrorName() *string {
|
||||||
return err.Name
|
return err.Name
|
||||||
}
|
}
|
||||||
|
|
||||||
func (err *CErr) GetRetryAfter() *int64 {
|
func (err *CErr) ErrorRetryAfter() *int {
|
||||||
return err.RetryAfter
|
return err.RetryAfter
|
||||||
}
|
}
|
||||||
|
|
||||||
func (err *CErr) GetStatusCode() *int {
|
func (err *CErr) ErrorStatusCode() *int {
|
||||||
return err.StatusCode
|
return err.StatusCode
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -584,7 +584,7 @@ func TestRetryAfter(t *testing.T) {
|
|||||||
RetriesAttempted: 2,
|
RetriesAttempted: 2,
|
||||||
Exception: new(CErr).New(map[string]interface{}{
|
Exception: new(CErr).New(map[string]interface{}{
|
||||||
"Code": "CErr",
|
"Code": "CErr",
|
||||||
"RetryAfter": int64(3000),
|
"RetryAfter": 3000,
|
||||||
}),
|
}),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+21
-48
@@ -7,7 +7,6 @@ import (
|
|||||||
"io"
|
"io"
|
||||||
"io/ioutil"
|
"io/ioutil"
|
||||||
"strings"
|
"strings"
|
||||||
"fmt"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// 定义 Event 结构体
|
// 定义 Event 结构体
|
||||||
@@ -15,36 +14,24 @@ type SSEEvent struct {
|
|||||||
ID *string
|
ID *string
|
||||||
Event *string
|
Event *string
|
||||||
Data *string
|
Data *string
|
||||||
Retry *int
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// 解析单个事件
|
// 解析单个事件
|
||||||
func parseEvent(lines []string) *SSEEvent {
|
func parseEvent(eventLines []string) *SSEEvent {
|
||||||
event := &SSEEvent{}
|
var event *SSEEvent
|
||||||
for _, line := range lines {
|
var data string
|
||||||
if strings.HasPrefix(line, "data: ") {
|
var id string
|
||||||
data := strings.TrimPrefix(line, "data: ") + "\n"
|
|
||||||
if event.Data == nil {
|
for _, line := range eventLines {
|
||||||
event.Data = new(string)
|
if strings.HasPrefix(line, "data:") {
|
||||||
}
|
data += strings.TrimPrefix(line, "data:") + "\n"
|
||||||
*event.Data += data
|
} else if strings.HasPrefix(line, "id:") {
|
||||||
} else if strings.HasPrefix(line, "event: ") {
|
id += strings.TrimPrefix(line, "data:") + "\n"
|
||||||
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: ") {
|
|
||||||
var retry int
|
|
||||||
fmt.Sscanf(strings.TrimPrefix(line, "retry: "), "%d", &retry)
|
|
||||||
event.Retry = &retry
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// Remove last newline from data
|
|
||||||
if event.Data != nil {
|
event.Data = String(data)
|
||||||
data := strings.TrimRight(*event.Data, "\n")
|
event.ID = String(id)
|
||||||
event.Data = &data
|
|
||||||
}
|
|
||||||
return event
|
return event
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -90,48 +77,34 @@ func ReadAsString(body io.Reader) (string, error) {
|
|||||||
return string(byt), nil
|
return string(byt), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func ReadAsSSE(body io.ReadCloser, eventChannel chan *SSEEvent, errorChannel chan error) {
|
func ReadAsSSE(body io.ReadCloser) (<-chan *SSEEvent, <-chan error) {
|
||||||
|
eventChannel := make(chan *SSEEvent)
|
||||||
|
|
||||||
|
// 启动 Goroutine 解析 SSE 数据
|
||||||
go func() {
|
go func() {
|
||||||
defer func() {
|
defer body.Close()
|
||||||
body.Close()
|
defer close(eventChannel)
|
||||||
close(eventChannel)
|
var eventLines []string
|
||||||
}()
|
|
||||||
|
|
||||||
reader := bufio.NewReader(body)
|
reader := bufio.NewReader(body)
|
||||||
var eventLines []string
|
|
||||||
|
|
||||||
for {
|
for {
|
||||||
line, err := reader.ReadString('\n')
|
line, err := reader.ReadString('\n')
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if err == io.EOF {
|
|
||||||
// Handle the end of the stream and possibly pending event
|
|
||||||
if len(eventLines) > 0 {
|
|
||||||
event := parseEvent(eventLines)
|
|
||||||
eventChannel <- event
|
|
||||||
}
|
|
||||||
errorChannel <- nil
|
|
||||||
return
|
|
||||||
}
|
|
||||||
errorChannel <- err
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
line = strings.TrimRight(line, "\n")
|
line = strings.TrimRight(line, "\n")
|
||||||
|
|
||||||
if line == "" {
|
if line == "" {
|
||||||
// End of an SSE event
|
|
||||||
if len(eventLines) > 0 {
|
if len(eventLines) > 0 {
|
||||||
event := parseEvent(eventLines)
|
event := parseEvent(eventLines)
|
||||||
eventChannel <- event
|
eventChannel <- event
|
||||||
eventLines = []string{} // Reset for the next event
|
eventLines = []string{}
|
||||||
}
|
}
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
eventLines = append(eventLines, line)
|
eventLines = append(eventLines, line)
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
return eventChannel, nil
|
||||||
return
|
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user