refactor: migrate SSE streaming from channel-based to iterator pattern

- Replace channel-based streaming in Client.SendStream with iter.Seq2
- Extract SSE parsing logic into dedicated sseHandler function
- Update WorkflowService.RunStream to return iterator instead of channel
- Use sync.Pool for buffer reuse to reduce allocations
- Remove unused Event struct and defaultChannelBufferSize constant

This simplifies the streaming API, eliminates goroutine management,
and provides better resource cleanup through iterator yield control.

Author: yumosx

client/api/v1/workflow.go

@@ -7,14 +7,14 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"iter"
 	"net/http"
 
 	"github.com/yeeaiclub/dify-go/internal/handler"
 	"github.com/yeeaiclub/dify-go/schema"
 )
 
 const (
-	defaultChannelBufferSize = 10
 	// StreamMode represents the streaming response mode for workflow execution
 	StreamMode = "streaming"
 	// BlockingMode represents the blocking response mode for workflow execution
@@ -41,10 +41,9 @@ func NewWorkflowService(baseURL, apiKey string) *WorkflowService {
 func (w *WorkflowService) RunStream(
 	ctx context.Context,
 	req schema.RunWorkflowRequest,
-	respCh chan schema.StreamEvent[schema.RunWorkflowResponse],
-) error {
+) (iter.Seq2[[]byte, error], error) {
 	if req.ResponseMode != StreamMode {
-		return nil
+		return nil, errors.New("invalid response mode")
 	}
 
 	r, err := handler.NewRequestBuilder().
@@ -55,56 +54,10 @@ func (w *WorkflowService) RunStream(
 		Body(req).
 		Build()
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	go func() {
-		evh := make(chan *handler.Event, defaultChannelBufferSize)
-		defer close(respCh) // Ensure the response channel is closed when done
-
-		// Send the streaming request
-		err := w.client.SendStream(ctx, r, evh)
-		if err != nil {
-			respCh <- schema.StreamEvent[schema.RunWorkflowResponse]{
-				Err: err.Error(), // Send the actual error, not ctx.Err()
-			}
-			return
-		}
-
-		// Process the stream events
-		for {
-			select {
-			case <-ctx.Done():
-				respCh <- schema.StreamEvent[schema.RunWorkflowResponse]{
-					Err: ctx.Err().Error(),
-				}
-				return
-			case ev, ok := <-evh:
-				if !ok {
-					return
-				}
-				if ev.Done {
-					respCh <- schema.StreamEvent[schema.RunWorkflowResponse]{
-						Done: true,
-					}
-					return
-				}
-				var data schema.RunWorkflowResponse
-				err := json.NewDecoder(ev.Data).Decode(&data)
-				if err != nil {
-					respCh <- schema.StreamEvent[schema.RunWorkflowResponse]{
-						Err: err.Error(),
-					}
-					return
-				}
-				respCh <- schema.StreamEvent[schema.RunWorkflowResponse]{
-					Type: ev.Type,
-					Data: data,
-				}
-			}
-		}
-	}()
-	return nil
+	return w.client.SendStream(ctx, r)
 }
 
 // Run executes a workflow in blocking mode, Cannot execute if there is no published workflow.

internal/handler/client.go

@@ -4,13 +4,12 @@
 package handler
 
 import (
-	"bufio"
 	"bytes"
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
+	"iter"
 	"net/http"
 	"net/url"
 	"time"
@@ -69,15 +68,15 @@ func (c *Client) Send(ctx context.Context, req Request) (*Response, error) {
 }
 
 // SendStream sends an HTTP request and returns streaming responses via the provided channel.
-func (c *Client) SendStream(ctx context.Context, req Request, evh chan *Event) error {
+func (c *Client) SendStream(ctx context.Context, req Request) (iter.Seq2[[]byte, error], error) {
 	httpReq, err := c.buildRequest(ctx, req)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	httpReq.Header.Set("Accept", "text/event-stream")
 	httpReq.Header.Set("Cache-Control", "no-cache")
 	httpReq.Header.Set("Connection", "keep-alive")
-	return c.doStreamRequest(ctx, httpReq, evh)
+	return c.doStreamRequest(ctx, httpReq)
 }
 
 // marshalBody serializes the request body to JSON.
@@ -146,57 +145,18 @@ func (c *Client) doRequest(req *http.Request) (*Response, error) {
 }
 
 // doStreamRequest executes the HTTP request and returns streaming responses via the provided channel.
-func (c *Client) doStreamRequest(ctx context.Context, req *http.Request, evCh chan *Event) error {
+func (c *Client) doStreamRequest(ctx context.Context, req *http.Request) (iter.Seq2[[]byte, error], error) {
 	ctxReq := req.WithContext(ctx)
-	resp, err := c.client.Do(ctxReq) //nolint:bodyclose // ignore the error, it will be handled in the next step
+	resp, err := c.client.Do(ctxReq)
 	if err != nil {
-		return fmt.Errorf("failed to send HTTP request: %w", err)
+		return nil, fmt.Errorf("failed to send HTTP request: %w", err)
 	}
 
 	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("HTTP error resp code: %d", resp.StatusCode)
+		resp.Body.Close()
+		return nil, fmt.Errorf("HTTP error resp code: %d", resp.StatusCode)
 	}
-
-	go func() {
-		defer func() {
-			err = resp.Body.Close()
-			if err != nil {
-				log.Errorf("failed to close the http body: %v", err)
-			}
-		}()
-		reader := bufio.NewReader(resp.Body)
-		for {
-			line, err := reader.ReadBytes('\n')
-			if err != nil {
-				if errors.Is(err, io.EOF) {
-					evCh <- &Event{Done: true}
-					return
-				}
-				log.Errorf("failed to read: %v", err)
-				return
-			}
-			line = bytes.TrimRight(line, "\r\n")
-			if bytes.IndexByte(line, ':') == -1 {
-				continue
-			}
-			field := line
-			value := []byte{}
-			if i := bytes.IndexByte(line, ':'); i >= 0 {
-				field = line[:i]
-				if i+1 < len(line) {
-					value = line[i+1:]
-					if len(value) > 0 && value[0] == ' ' {
-						value = value[1:]
-					}
-				}
-			}
-
-			if string(field) == "data" {
-				evCh <- &Event{Data: bytes.NewBuffer(value), Type: string(field)}
-			}
-		}
-	}()
-	return nil
+	return sseHandler(resp.Body), nil
 }
 
 // buildURL constructs a complete URL from base URL, path, and query parameters.

internal/handler/sse.go

@@ -0,0 +1,59 @@
+package handler
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"io"
+	"iter"
+	"sync"
+)
+
+var (
+	sseBufferSize = 64 * 1024 // 64KB
+	sseDataPrefix = "data:"
+
+	pool = sync.Pool{
+		New: func() any {
+			buf := make([]byte, 0, 4096)
+			return &buf
+		},
+	}
+)
+
+func getBuffer() *[]byte {
+	return pool.Get().(*[]byte)
+}
+
+func putBuffer(buf *[]byte) {
+	*buf = (*buf)[:0]
+}
+
+func sseHandler(body io.ReadCloser) iter.Seq2[[]byte, error] {
+	return func(yield func([]byte, error) bool) {
+		defer body.Close()
+
+		scanner := bufio.NewScanner(body)
+		buf := getBuffer()
+		defer putBuffer(buf)
+
+		scanner.Buffer(*buf, sseBufferSize)
+		for scanner.Scan() {
+			lineBytes := scanner.Bytes()
+			prefix := []byte(sseDataPrefix)
+			if !bytes.HasPrefix(lineBytes, prefix) {
+				continue
+			}
+			data := lineBytes[len(prefix):]
+			if len(data) > 0 && data[0] == ' ' {
+				data = data[1:]
+			}
+			if !yield(data, nil) {
+				return
+			}
+		}
+		if err := scanner.Err(); err != nil {
+			yield(nil, fmt.Errorf("sseHandler scanning error: %v", err))
+		}
+	}
+}