From 896987fc6598ef3fb6e149ae31a9b58fd8820c08 Mon Sep 17 00:00:00 2001 From: Joe Date: Fri, 10 Jul 2026 12:58:53 +0800 Subject: [PATCH] feat(http): add SSE streaming support via Client.Stream Add server-sent events (SSE) support to the http package: - Client.Stream sends a request with "Accept: text/event-stream" and returns a pull-based *Stream; request/connection errors and non-200 responses (*ApiError) are returned synchronously. - Stream follows the bufio.Scanner / sql.Rows style: Next / Current / Err / Close, plus an Events() iter.Seq[[]byte] adapter for range-over-func consumption. Mid-stream read failures, timeouts and cancellation are reported by Err. - Streaming uses a copy of the configured http.Client with Timeout cleared, since http.Client.Timeout caps the whole request and would abort a long-lived stream; the lifetime is bounded by ctx instead. - Extract request building/signing from Call into newRequest, shared by Call and Stream. Co-Authored-By: Claude Fable 5 --- http/client.go | 39 +++++--- http/stream.go | 185 ++++++++++++++++++++++++++++++++++++ http/stream_test.go | 225 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 435 insertions(+), 14 deletions(-) create mode 100644 http/stream.go create mode 100644 http/stream_test.go diff --git a/http/client.go b/http/client.go index 1ca9d8e..1d1d723 100644 --- a/http/client.go +++ b/http/client.go @@ -100,15 +100,10 @@ func (c *Client) GetOTPV2(ctx context.Context, ropts ...RequestOption) (string, return res.Otp, nil } -// Call will send request with signature to http server -func (c *Client) Call(ctx context.Context, method, path string, queryParams interface{}, body interface{}, resp interface{}, ropts ...RequestOption) (err error) { - var ( - br io.Reader - bb []byte - httpResp *nhttp.Response - rb []byte - ) - +// newRequest builds a signed *http.Request. It also returns the marshalled +// request body (bb) so callers can log it or reuse it. It is shared by Call +// and Stream. +func (c *Client) newRequest(ctx context.Context, method, path string, queryParams interface{}, body interface{}, ropts ...RequestOption) (req *nhttp.Request, bb []byte, err error) { ro := &RequestOptions{} for _, opt := range ropts { opt(ro) @@ -118,17 +113,18 @@ func (c *Client) Call(ctx context.Context, method, path string, queryParams inte body = ro.body } + var br io.Reader if body != nil { bb, err = json.Marshal(body) if err != nil { - return err + return nil, nil, err } br = bytes.NewBuffer(bb) } - req, err := nhttp.NewRequestWithContext(ctx, method, c.opts.URL+path, br) + req, err = nhttp.NewRequestWithContext(ctx, method, c.opts.URL+path, br) if err != nil { - return err + return nil, nil, err } appKey := c.opts.AppKey @@ -137,7 +133,7 @@ func (c *Client) Call(ctx context.Context, method, path string, queryParams inte if c.opts.OAuthClient != nil { token, err := c.opts.OAuthClient.AccessToken(ctx) if err != nil { - return err + return nil, nil, err } appKey = c.opts.OAuthClient.ClientID() accessToken = "Bearer " + token @@ -165,7 +161,7 @@ func (c *Client) Call(ctx context.Context, method, path string, queryParams inte vals, ok := queryParams.(url.Values) if !ok { if vals, err = query.Values(queryParams); err != nil { - return + return nil, nil, err } } req.URL.RawQuery = vals.Encode() @@ -173,6 +169,21 @@ func (c *Client) Call(ctx context.Context, method, path string, queryParams inte // set signature (no-op for OAuth when appSecret is empty) signature(req, appSecret, bb) + return req, bb, nil +} + +// Call will send request with signature to http server +func (c *Client) Call(ctx context.Context, method, path string, queryParams interface{}, body interface{}, resp interface{}, ropts ...RequestOption) (err error) { + var ( + httpResp *nhttp.Response + rb []byte + ) + + req, bb, err := c.newRequest(ctx, method, path, queryParams, body, ropts...) + if err != nil { + return err + } + log.Debugf("http call method:%v url:%v body:%v", req.Method, req.URL, string(bb)) httpResp, err = c.httpClient.Do(req) if err != nil { diff --git a/http/stream.go b/http/stream.go new file mode 100644 index 0000000..ebb016d --- /dev/null +++ b/http/stream.go @@ -0,0 +1,185 @@ +package http + +import ( + "bufio" + "bytes" + "context" + "io" + "iter" + nhttp "net/http" + + "github.com/longbridge/openapi-go/log" +) + +// Stream represents a server-sent events (SSE) response. It is a pull-based +// reader in the style of bufio.Scanner / sql.Rows: no background goroutine, +// events are read from the connection on demand. +// +// stream, err := client.Stream(ctx, "GET", "/v1/xxx", nil, nil) +// if err != nil { +// return err // request/connection errors and non-200 responses +// } +// defer stream.Close() +// +// for stream.Next() { +// handle(stream.Current()) +// } +// if err := stream.Err(); err != nil { +// // mid-stream read failure, timeout, or cancellation +// } +// +// Or equivalently, range over Events(): +// +// for data := range stream.Events() { +// handle(data) +// } +// if err := stream.Err(); err != nil { ... } +// +// A Stream is not safe for concurrent use. To abort a stream from another +// goroutine, cancel the context passed to Client.Stream. +type Stream struct { + resp *nhttp.Response + reader *bufio.Reader + cur []byte + err error + closed bool +} + +// Stream sends a request expecting a server-sent events (SSE) response. It +// sets the "Accept: text/event-stream" request header and blocks until the +// response headers arrive. Request build errors, connection failures and +// non-200 responses (as *ApiError) are returned immediately; errors occurring +// after that — mid-stream read failures, timeouts, cancellation — are +// reported by Stream.Err. +// +// The stream's lifetime is bounded by ctx: the client-wide timeout does not +// apply, so pass a context with a deadline or cancellation to limit it. +// Always Close the returned stream. +func (c *Client) Stream(ctx context.Context, method, path string, queryParams interface{}, body interface{}, ropts ...RequestOption) (*Stream, error) { + req, bb, err := c.newRequest(ctx, method, path, queryParams, body, ropts...) + if err != nil { + return nil, err + } + req.Header.Set("accept", "text/event-stream") + + log.Debugf("http stream method:%v url:%v body:%v", req.Method, req.URL, string(bb)) + resp, err := c.streamHTTPClient().Do(req) + if err != nil { + return nil, err + } + + if resp.StatusCode != nhttp.StatusOK { + defer resp.Body.Close() + rb, _ := io.ReadAll(resp.Body) + apiResp := &apiResponse{TraceID: resp.Header.Get("x-trace-id")} + if isJSON(resp.Header.Get("content-type")) { + _ = jsonUnmarshal(bytes.NewReader(rb), apiResp) + } else { + apiResp.Message = string(rb) + } + return nil, NewError(resp.StatusCode, apiResp) + } + + return &Stream{ + resp: resp, + reader: bufio.NewReader(resp.Body), + }, nil +} + +// streamHTTPClient returns an *http.Client suitable for a long-lived stream. +// It is a shallow copy of the configured client with Timeout cleared, since a +// non-zero http.Client.Timeout caps the whole request (headers + body read) +// and would abort the stream. Callers control the lifetime via the context. +func (c *Client) streamHTTPClient() *nhttp.Client { + cli := *c.httpClient + cli.Timeout = 0 + return &cli +} + +// Next reads the next event from the stream. It returns false when the +// stream ends — because the server closed it, an error occurred, or the +// stream was closed. After Next returns false, check Err. +func (s *Stream) Next() bool { + if s.closed || s.err != nil { + return false + } + + var data bytes.Buffer + for { + line, err := s.reader.ReadBytes('\n') + if len(line) > 0 { + // strip the trailing line terminator (\n or \r\n) + line = bytes.TrimRight(line, "\r\n") + switch { + case len(line) == 0: + // blank line marks the end of an event + if data.Len() > 0 { + s.cur = data.Bytes() + return true + } + case bytes.HasPrefix(line, []byte(":")): + // comment / heartbeat, ignore + case bytes.HasPrefix(line, []byte("data:")): + val := bytes.TrimPrefix(line[len("data:"):], []byte(" ")) + if data.Len() > 0 { + data.WriteByte('\n') + } + data.Write(val) + default: + // other fields (event/id/retry) are not part of the payload + } + } + + if err != nil { + if err == io.EOF { + // flush a trailing event that had no terminating blank line + if data.Len() > 0 { + s.cur = data.Bytes() + return true + } + } else if !s.closed { + s.err = err + } + return false + } + } +} + +// Current returns the payload (the concatenated "data" fields) of the event +// read by the last successful call to Next. +func (s *Stream) Current() []byte { + return s.cur +} + +// Events returns an iterator over the remaining events of the stream, +// yielding each event's payload: +// +// for data := range stream.Events() { ... } +// +// It is a convenience wrapper around Next/Current; after the loop, check Err. +func (s *Stream) Events() iter.Seq[[]byte] { + return func(yield func([]byte) bool) { + for s.Next() { + if !yield(s.cur) { + return + } + } + } +} + +// Err returns the error, if any, that terminated the stream. Mid-stream read +// failures, timeouts and cancellation are all reported here. It returns nil +// for a stream that ended normally or was closed by Close. +func (s *Stream) Err() error { + return s.err +} + +// Close releases the underlying connection. It is safe to call multiple +// times. Always Close a stream, even one that was fully consumed. +func (s *Stream) Close() error { + if s.closed { + return nil + } + s.closed = true + return s.resp.Body.Close() +} diff --git a/http/stream_test.go b/http/stream_test.go new file mode 100644 index 0000000..bde1a79 --- /dev/null +++ b/http/stream_test.go @@ -0,0 +1,225 @@ +package http + +import ( + "context" + "fmt" + nhttp "net/http" + "net/http/httptest" + "testing" + "time" +) + +func newTestClient(t *testing.T, url string) *Client { + t.Helper() + c, err := New( + WithURL(url), + WithAppKey("key"), + WithAppSecret("secret"), + WithAccessToken("token"), + ) + if err != nil { + t.Fatalf("new client: %v", err) + } + return c +} + +func sseTestServer(t *testing.T) *httptest.Server { + t.Helper() + return httptest.NewServer(nhttp.HandlerFunc(func(w nhttp.ResponseWriter, r *nhttp.Request) { + if got := r.Header.Get("accept"); got != "text/event-stream" { + t.Errorf("accept header = %q, want text/event-stream", got) + } + w.Header().Set("content-type", "text/event-stream") + w.WriteHeader(nhttp.StatusOK) + fl := w.(nhttp.Flusher) + // two data lines -> joined with \n + fmt.Fprint(w, "data: hello\ndata: world\n\n") + fl.Flush() + fmt.Fprint(w, ": heartbeat\n\n") // comment-only event, skipped + fl.Flush() + fmt.Fprint(w, "event: msg\r\ndata: {\"a\":1}\r\n\r\n") // CRLF + non-data field + fl.Flush() + fmt.Fprint(w, "data: trailing") // no terminating blank line -> flushed on EOF + })) +} + +func checkEvents(t *testing.T, got []string) { + t.Helper() + want := []string{"hello\nworld", `{"a":1}`, "trailing"} + if len(got) != len(want) { + t.Fatalf("got %d events %q, want %d %q", len(got), got, len(want), want) + } + for i := range want { + if got[i] != want[i] { + t.Errorf("event[%d] = %q, want %q", i, got[i], want[i]) + } + } +} + +func TestStreamNext(t *testing.T) { + srv := sseTestServer(t) + defer srv.Close() + + c := newTestClient(t, srv.URL) + stream, err := c.Stream(context.Background(), "GET", "/v1/stream", nil, nil) + if err != nil { + t.Fatalf("Stream() error: %v", err) + } + defer stream.Close() + + var got []string + for stream.Next() { + got = append(got, string(stream.Current())) + } + if err := stream.Err(); err != nil { + t.Fatalf("Err() = %v, want nil", err) + } + checkEvents(t, got) + + if stream.Next() { + t.Error("Next() after exhaustion = true, want false") + } +} + +func TestStreamEventsIterator(t *testing.T) { + srv := sseTestServer(t) + defer srv.Close() + + c := newTestClient(t, srv.URL) + stream, err := c.Stream(context.Background(), "GET", "/v1/stream", nil, nil) + if err != nil { + t.Fatalf("Stream() error: %v", err) + } + defer stream.Close() + + var got []string + for data := range stream.Events() { + got = append(got, string(data)) + } + if err := stream.Err(); err != nil { + t.Fatalf("Err() = %v, want nil", err) + } + checkEvents(t, got) +} + +func TestStreamEventsIteratorBreak(t *testing.T) { + srv := sseTestServer(t) + defer srv.Close() + + c := newTestClient(t, srv.URL) + stream, err := c.Stream(context.Background(), "GET", "/v1/stream", nil, nil) + if err != nil { + t.Fatalf("Stream() error: %v", err) + } + defer stream.Close() + + var got []string + for data := range stream.Events() { + got = append(got, string(data)) + break + } + if len(got) != 1 || got[0] != "hello\nworld" { + t.Fatalf("got %q, want just the first event", got) + } + if err := stream.Err(); err != nil { + t.Fatalf("Err() = %v, want nil", err) + } +} + +func TestStreamNon200(t *testing.T) { + srv := httptest.NewServer(nhttp.HandlerFunc(func(w nhttp.ResponseWriter, r *nhttp.Request) { + w.Header().Set("content-type", "application/json") + w.Header().Set("x-trace-id", "trace-123") + w.WriteHeader(nhttp.StatusBadRequest) + fmt.Fprint(w, `{"code":40001,"message":"bad request"}`) + })) + defer srv.Close() + + c := newTestClient(t, srv.URL) + stream, err := c.Stream(context.Background(), "GET", "/v1/stream", nil, nil) + if err == nil { + stream.Close() + t.Fatal("Stream() error = nil, want *ApiError") + } + ae, ok := err.(*ApiError) + if !ok { + t.Fatalf("Stream() error type = %T, want *ApiError", err) + } + if ae.HttpStatus != nhttp.StatusBadRequest || ae.Code != 40001 || ae.TraceID != "trace-123" { + t.Errorf("unexpected ApiError: %+v", ae) + } +} + +func TestStreamContextCancel(t *testing.T) { + release := make(chan struct{}) + srv := httptest.NewServer(nhttp.HandlerFunc(func(w nhttp.ResponseWriter, r *nhttp.Request) { + w.Header().Set("content-type", "text/event-stream") + w.WriteHeader(nhttp.StatusOK) + w.(nhttp.Flusher).Flush() + <-release // hold the connection open until the test is done + })) + defer srv.Close() + defer close(release) + + c := newTestClient(t, srv.URL) + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + stream, err := c.Stream(ctx, "GET", "/v1/stream", nil, nil) + if err != nil { + t.Fatalf("Stream() error: %v", err) + } + defer stream.Close() + + done := make(chan struct{}) + go func() { + for stream.Next() { + } + close(done) + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("stream did not terminate after context timeout") + } + + if stream.Err() == nil { + t.Fatal("Err() = nil, want context deadline / timeout error") + } +} + +func TestStreamConnectError(t *testing.T) { + // point at a closed server so the request fails to connect + srv := httptest.NewServer(nhttp.HandlerFunc(func(w nhttp.ResponseWriter, r *nhttp.Request) {})) + url := srv.URL + srv.Close() + + c := newTestClient(t, url) + if _, err := c.Stream(context.Background(), "GET", "/v1/stream", nil, nil); err == nil { + t.Fatal("Stream() error = nil, want connection error") + } +} + +func TestStreamClose(t *testing.T) { + srv := sseTestServer(t) + defer srv.Close() + + c := newTestClient(t, srv.URL) + stream, err := c.Stream(context.Background(), "GET", "/v1/stream", nil, nil) + if err != nil { + t.Fatalf("Stream() error: %v", err) + } + if err := stream.Close(); err != nil { + t.Fatalf("Close() error: %v", err) + } + if err := stream.Close(); err != nil { + t.Fatalf("second Close() error: %v", err) + } + if stream.Next() { + t.Error("Next() after Close = true, want false") + } + if err := stream.Err(); err != nil { + t.Errorf("Err() after Close = %v, want nil", err) + } +}