feat: SSE introduce sync pool for rawEvent to optimize memory usage (#1011)

* sse: introduce a sync pool for `rawEvent` to optimize memory usage.

* fix deferred resource cleanup in event parsing

Move the `defer putRawEvent(ed)` call after error handling to ensure resources are properly released only when parsing succeeds. This prevents potential resource leaks when errors occur during event parsing.

Author: liuzengh

sse.go

@@ -534,45 +534,54 @@ func (es *EventSource) listenStream(res *http.Response) error {
 			return nil
 		}
 
-		e, err := readEvent(scanner)
-		if err != nil {
-			if err == io.EOF {
-				return err
-			}
-			es.triggerOnError(err)
+		if err := es.processEvent(scanner); err != nil {
 			return err
 		}
+	}
+}
 
-		ed, err := parseEvent(e)
-		if err != nil {
-			es.triggerOnError(err)
-			continue // parsing errors, just continue
+func (es *EventSource) processEvent(scanner *bufio.Scanner) error {
+	e, err := readEvent(scanner)
+	if err != nil {
+		if err == io.EOF {
+			return err
 		}
+		es.triggerOnError(err)
+		return err
+	}
+
+	ed, err := parseEvent(e)
+	if err != nil {
+		es.triggerOnError(err)
+		return nil // parsing errors, will not return error.
+	}
+	defer putRawEvent(ed)
+
+	if len(ed.ID) > 0 {
+		es.lock.Lock()
+		es.lastEventID = string(ed.ID)
+		es.lock.Unlock()
+	}
 
-		if len(ed.ID) > 0 {
+	if len(ed.Retry) > 0 {
+		if retry, err := strconv.Atoi(string(ed.Retry)); err == nil {
 			es.lock.Lock()
-			es.lastEventID = string(ed.ID)
+			es.serverSentRetry = time.Millisecond * time.Duration(retry)
 			es.lock.Unlock()
+		} else {
+			es.triggerOnError(err)
 		}
+	}
 
-		if len(ed.Retry) > 0 {
-			if retry, err := strconv.Atoi(string(ed.Retry)); err == nil {
-				es.lock.Lock()
-				es.serverSentRetry = time.Millisecond * time.Duration(retry)
-				es.lock.Unlock()
-			} else {
-				es.triggerOnError(err)
-			}
-		}
-
-		if len(ed.Data) > 0 {
-			es.handleCallback(&Event{
-				ID:   string(ed.ID),
-				Name: string(ed.Event),
-				Data: string(ed.Data),
-			})
-		}
+	if len(ed.Data) > 0 {
+		es.handleCallback(&Event{
+			ID:   string(ed.ID),
+			Name: string(ed.Event),
+			Data: string(ed.Data),
+		})
 	}
+
+	return nil
 }
 
 func (es *EventSource) handleCallback(e *Event) {
@@ -633,7 +642,7 @@ func parseEventFunc(msg []byte) (*rawEvent, error) {
 		return nil, errors.New("resty:sse: event message was empty")
 	}
 
-	e := new(rawEvent)
+	e := newRawEvent()
 
 	// Split the line by "\n"
 	for _, line := range bytes.FieldsFunc(msg, func(r rune) bool { return r == '\n' }) {
@@ -670,3 +679,18 @@ func trimHeader(size int, data []byte) []byte {
 	data = bytes.TrimSuffix(data, []byte("\n"))
 	return data
 }
+
+var rawEventPool = &sync.Pool{New: func() any { return new(rawEvent) }}
+
+func newRawEvent() *rawEvent {
+	e := rawEventPool.Get().(*rawEvent)
+	e.ID = e.ID[:0]
+	e.Data = e.Data[:0]
+	e.Event = e.Event[:0]
+	e.Retry = e.Retry[:0]
+	return e
+}
+
+func putRawEvent(e *rawEvent) {
+	rawEventPool.Put(e)
+}