From 93d64addd2343fdb2e6c0087ceab7eb4af122e4b Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Fri, 21 Aug 2026 15:29:16 -0500 Subject: [PATCH] feat(generator): a text/event-stream response is read as it arrives Every response was read whole before anything was decoded, so an endpoint that streams was reachable only as one lump once the server finished, which for a chat completion is the opposite of the point. The ACTIVATE spec offers exactly one such response, its OpenAI-compatible chat endpoint, and its client could not stream it. An operation whose success response offers text/event-stream now gets a second method returning EventStream[T], typed to the schema the stream declares, read event by event. The buffered method stays, so an endpoint offering both JSON and events has one method for each. do and doStream now share send, which runs the request and hands back the response with its body unread. A non-2xx comes back as an APIError carrying the body, so a body is left open only for a response that succeeded, and the retry path drains rather than reads. The parser follows the event stream format: several data lines join, blank lines end an event, comments are the keep-alives servers send, and a stream that ends without the blank line that would have dispatched its last event still delivers it. A [DONE] payload ends iteration rather than failing to decode, which is how OpenAI-compatible APIs close a stream. --- README.md | 32 +++ internal/analyzer/operations.go | 23 +++ internal/analyzer/operations_test.go | 60 ++++++ internal/generator/e2e_streaming_test.go | 244 +++++++++++++++++++++++ internal/generator/e2e_test.go | 1 + internal/generator/funcmap.go | 6 + internal/generator/generator.go | 1 + internal/ir/operations.go | 5 +- internal/templates/client.go.tmpl | 96 +++++---- internal/templates/operations.go.tmpl | 47 +++-- internal/templates/reserved.go | 1 + internal/templates/streaming.go.tmpl | 135 +++++++++++++ 12 files changed, 601 insertions(+), 50 deletions(-) create mode 100644 internal/generator/e2e_streaming_test.go create mode 100644 internal/templates/streaming.go.tmpl diff --git a/README.md b/README.md index 74a965e..c250e56 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,7 @@ Given any OpenAPI 3.1 (or 3.0) spec, it outputs a complete, idiomatic Go client - **Webhooks and callbacks**: typed payloads and a dispatcher for the requests the API sends you - **Server URLs**: `DefaultBaseURL` from the spec, with a builder for templated servers - **Response headers**: status and headers captured through the context, with declared headers parsed per operation +- **Streaming**: `text/event-stream` responses get a typed `EventStream[T]` read as the server writes it - **Authentication**: `AuthProvider` interface with built-in Bearer, API key, and Basic auth, skipped for operations the spec marks as needing none - **Error handling** — `APIError` with sentinel errors (`errors.Is`), typed error wrappers with parsed response bodies (`errors.As`), readable messages via `x-ms-primary-error-message` - **Pagination**: auto-detected cursor, offset, and page pagination with a generic `PageIterator[T]` @@ -163,6 +164,37 @@ client := petstore.NewClient(petstore.ServerURL("eu-west-1", "")) A relative server URL (`/api/v3`) gets neither, since it resolves against wherever the spec is served and the generated package cannot know that host. +### Streaming responses + +A response that offers `text/event-stream` gets a second method that reads the +events as the server writes them, typed to the schema the stream declares: + +```go +stream, err := client.ChatCompletionStream(ctx, request) +if err != nil { + return err +} +defer stream.Close() + +for { + chunk, err := stream.Next() + if errors.Is(err, io.EOF) { + break + } + if err != nil { + return err + } + fmt.Print(chunk.Choices[0].Delta.Content) +} +``` + +The buffered method stays as it is, so an endpoint offering both JSON and events +has one method for each. `stream.EventName()` and `stream.EventID()` carry the +`event` and `id` fields of the event just returned, for a stream that names them. +A payload of `[DONE]`, which OpenAI-compatible APIs use to close a stream, ends +iteration rather than failing to decode, and a stream whose schema is a string +hands back each event's text rather than parsing it. + ### Response headers A method returns the decoded body, so what a response says outside its body is diff --git a/internal/analyzer/operations.go b/internal/analyzer/operations.go index d8b382b..807a9e2 100644 --- a/internal/analyzer/operations.go +++ b/internal/analyzer/operations.go @@ -570,6 +570,11 @@ func (a *Analyzer) convertResponses(responses *v3high.Responses, opDef *ir.Opera if opDef.SuccessResponse == nil { opDef.SuccessResponse = rd } + // A response offering text/event-stream carries its payload one + // event at a time, which the buffered path cannot hand back. + if opDef.EventType == "" { + opDef.EventType = a.eventStreamType(resp, hint) + } } else if isErrorCode(code) { rd.IsError = true rd.ErrorWrapper = a.errorWrapperName(rd.TypeName, hint) @@ -651,6 +656,24 @@ func isJSONContent(contentType string) bool { return strings.Contains(contentType, "json") } +// eventStreamType returns the Go type of one event's payload for a response that +// offers text/event-stream, or "" for one that does not. +func (a *Analyzer) eventStreamType(resp *v3high.Response, nameHint string) string { + if resp.Content == nil { + return "" + } + mediaType, ok := resp.Content.Get("text/event-stream") + if !ok || mediaType == nil { + return "" + } + goType := a.resolveMediaTypeSchema(mediaType, nameHint+"Event") + if goType == "" { + // Events with no schema are still events; their data arrives as text. + return "string" + } + return goType +} + // convertResponseHeaders lowers the headers a response declares. A header value // arrives as text, so only the kinds text parses into unambiguously are typed; // everything else, dates and lists included, stays the raw string. diff --git a/internal/analyzer/operations_test.go b/internal/analyzer/operations_test.go index 94cc8e4..a5f5f1b 100644 --- a/internal/analyzer/operations_test.go +++ b/internal/analyzer/operations_test.go @@ -849,3 +849,63 @@ func TestOperationNames_CollisionsAreNumbered(t *testing.T) { } _ = typeMap } + +const eventStreamSpec = `openapi: 3.1.0 +info: { title: t, version: "1" } +paths: + /chat: + post: + operationId: chat + responses: + "200": + description: whole or in parts + content: + application/json: { schema: { $ref: "#/components/schemas/Reply" } } + text/event-stream: { schema: { $ref: "#/components/schemas/Chunk" } } + /logs: + get: + operationId: tailLogs + responses: + "200": + description: events with no schema + content: + text/event-stream: {} + /plain: + get: + operationId: plain + responses: + "200": + description: ok + content: + application/json: { schema: { $ref: "#/components/schemas/Reply" } } +components: + schemas: + Reply: { type: object, properties: { text: { type: string } } } + Chunk: { type: object, properties: { delta: { type: string } } } +` + +// A response offering text/event-stream carries its payload one event at a +// time, which the buffered path cannot hand back. +func TestEventStream_PayloadTypeIsTheEventSchema(t *testing.T) { + pkg, _ := analyzeSpec(t, eventStreamSpec) + + byName := map[string]*ir.OperationDef{} + for _, op := range pkg.Operations { + byName[op.Name] = op + } + + if chat := byName["Chat"]; chat == nil || chat.EventType != "Chunk" { + t.Errorf("Chat EventType = %+v, want Chunk", chat) + } + // The JSON alternative still drives the buffered method. + if chat := byName["Chat"]; chat == nil || chat.SuccessResponse.TypeName != "Reply" { + t.Errorf("Chat success type = %+v, want Reply", chat.SuccessResponse) + } + // Events with no schema are still events; their data arrives as text. + if logs := byName["TailLogs"]; logs == nil || logs.EventType != "string" { + t.Errorf("TailLogs EventType = %+v, want string", logs) + } + if plain := byName["Plain"]; plain == nil || plain.EventType != "" { + t.Errorf("Plain EventType = %+v, want none", plain) + } +} diff --git a/internal/generator/e2e_streaming_test.go b/internal/generator/e2e_streaming_test.go new file mode 100644 index 0000000..c7192ce --- /dev/null +++ b/internal/generator/e2e_streaming_test.go @@ -0,0 +1,244 @@ +package generator + +import "testing" + +const streamingSpec = `openapi: 3.1.0 +info: { title: chat, version: "1" } +paths: + /chat: + post: + operationId: chat + requestBody: + required: true + content: + application/json: { schema: { $ref: "#/components/schemas/Request" } } + responses: + "200": + description: whole or in parts + content: + application/json: { schema: { $ref: "#/components/schemas/Reply" } } + text/event-stream: { schema: { $ref: "#/components/schemas/Chunk" } } + /logs: + get: + operationId: tailLogs + responses: + "200": + description: text events + content: + text/event-stream: { schema: { type: string } } +components: + schemas: + Request: + type: object + properties: + prompt: { type: string } + Reply: + type: object + properties: + text: { type: string } + Chunk: + type: object + properties: + delta: { type: string } + index: { type: integer } +` + +// TestE2E_EventStreams covers a response offering text/event-stream. The client +// read every response whole, so a stream was reachable only as one lump after +// the server finished, which for a chat completion is the opposite of the point. +func TestE2E_EventStreams(t *testing.T) { + files, _ := generateFromSpec(t, streamingSpec, "chatapi") + + runGeneratedWireTest(t, files, "eventstream", `package chatapi + +import ( + "errors" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +// frames serves the given SSE text, flushing each write so the client sees the +// events as they are written rather than at the end. +func frames(t *testing.T, chunks ...string) *Client { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("Accept"); got != "text/event-stream" { + t.Errorf("Accept = %q, want text/event-stream", got) + } + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + for _, chunk := range chunks { + w.Write([]byte(chunk)) + w.(http.Flusher).Flush() + } + })) + t.Cleanup(srv.Close) + return NewClient(srv.URL) +} + +func TestChunksArriveTyped(t *testing.T) { + client := frames(t, + ": keep-alive\n\n", + "data: {\"delta\":\"He\",\"index\":0}\n\n", + "data: {\"delta\":\"llo\",\"index\":1}\n\n", + "data: [DONE]\n\n", + ) + + stream, err := client.ChatStream(t.Context(), Request{}) + if err != nil { + t.Fatalf("ChatStream: %v", err) + } + defer stream.Close() + + var text string + var count int + for { + chunk, err := stream.Next() + if errors.Is(err, io.EOF) { + break + } + if err != nil { + t.Fatalf("Next: %v", err) + } + count++ + if chunk.Delta != nil { + text += *chunk.Delta + } + } + + // The comment is a keep-alive and the sentinel is not an event, so neither + // is delivered as one. + if count != 2 || text != "Hello" { + t.Errorf("got %d chunks spelling %q, want 2 spelling Hello", count, text) + } +} + +// Events arrive as the server writes them, which is the whole reason for the +// method: a reader blocked until the response completed would time out here. +func TestEventsArriveBeforeTheStreamEnds(t *testing.T) { + release := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + w.Write([]byte("data: {\"delta\":\"first\"}\n\n")) + w.(http.Flusher).Flush() + <-release + w.Write([]byte("data: [DONE]\n\n")) + w.(http.Flusher).Flush() + })) + defer srv.Close() + defer close(release) + + stream, err := NewClient(srv.URL).ChatStream(t.Context(), Request{}) + if err != nil { + t.Fatalf("ChatStream: %v", err) + } + defer stream.Close() + + done := make(chan string, 1) + go func() { + chunk, err := stream.Next() + if err != nil { + done <- "error: " + err.Error() + return + } + done <- *chunk.Delta + }() + + select { + case got := <-done: + if got != "first" { + t.Errorf("first event = %q", got) + } + case <-time.After(5 * time.Second): + t.Fatal("the first event did not arrive while the server was still writing") + } +} + +func TestMultilineDataAndEventFields(t *testing.T) { + client := frames(t, "event: tick\nid: 7\ndata: line one\ndata: line two\n\n") + + stream, err := client.TailLogsStream(t.Context()) + if err != nil { + t.Fatalf("TailLogsStream: %v", err) + } + defer stream.Close() + + // A stream of text declares string, where the data is the value rather than + // a document to decode. + line, err := stream.Next() + if err != nil { + t.Fatalf("Next: %v", err) + } + if line != "line one\nline two" { + t.Errorf("payload = %q, want the data lines joined", line) + } + if stream.EventName() != "tick" || stream.EventID() != "7" { + t.Errorf("event name = %q id = %q, want tick and 7", stream.EventName(), stream.EventID()) + } +} + +// A stream that ends without the blank line that would dispatch its last event +// still delivers it. +func TestUnterminatedFinalEventIsDelivered(t *testing.T) { + client := frames(t, "data: {\"delta\":\"tail\"}") + + stream, err := client.ChatStream(t.Context(), Request{}) + if err != nil { + t.Fatalf("ChatStream: %v", err) + } + defer stream.Close() + + chunk, err := stream.Next() + if err != nil { + t.Fatalf("Next: %v", err) + } + if chunk.Delta == nil || *chunk.Delta != "tail" { + t.Errorf("chunk = %+v", chunk) + } + if _, err := stream.Next(); !errors.Is(err, io.EOF) { + t.Errorf("second Next = %v, want io.EOF", err) + } +} + +func TestErrorStatusIsAnErrorRatherThanAStream(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusTooManyRequests) + w.Write([]byte("slow down")) + })) + defer srv.Close() + + _, err := NewClient(srv.URL).ChatStream(t.Context(), Request{}) + var apiErr *APIError + if !errors.As(err, &apiErr) || apiErr.StatusCode != http.StatusTooManyRequests { + t.Fatalf("err = %v, want the 429", err) + } + if string(apiErr.Body) != "slow down" { + t.Errorf("body = %q, want the error payload", apiErr.Body) + } +} + +// The buffered method is still there, and still decodes the JSON alternative. +func TestBufferedMethodIsUnaffected(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("Accept"); got != "application/json" { + t.Errorf("Accept = %q, want application/json", got) + } + w.Header().Set("Content-Type", "application/json") + w.Write([]byte("{\"text\":\"whole\"}")) + })) + defer srv.Close() + + reply, err := NewClient(srv.URL).Chat(t.Context(), Request{}) + if err != nil { + t.Fatalf("Chat: %v", err) + } + if reply == nil || reply.Text == nil || *reply.Text != "whole" { + t.Errorf("reply = %+v", reply) + } +} +`) +} diff --git a/internal/generator/e2e_test.go b/internal/generator/e2e_test.go index 8377377..8ff291c 100644 --- a/internal/generator/e2e_test.go +++ b/internal/generator/e2e_test.go @@ -69,6 +69,7 @@ func TestE2E_PetstoreGeneration(t *testing.T) { "errors.go": false, "responses.go": false, "webhooks.go": false, + "streaming.go": false, } for _, f := range files { if _, ok := expectedFiles[f.Name]; ok { diff --git a/internal/generator/funcmap.go b/internal/generator/funcmap.go index f4e5c49..fb1ac81 100644 --- a/internal/generator/funcmap.go +++ b/internal/generator/funcmap.go @@ -46,6 +46,7 @@ func FuncMap() template.FuncMap { "declaredNamesLiteral": declaredNamesLiteral, "discriminatorFieldName": discriminatorFieldName, "hasPaginatedOps": hasPaginatedOps, + "hasEventStreams": hasEventStreams, "paginationItemType": paginationItemType, "paginationCursorField": paginationCursorField, "paginationOffsetParam": paginationOffsetParam, @@ -742,6 +743,11 @@ func discriminatorFieldName(propertyName string) string { return naming.Exported(propertyName) } +// hasEventStreams reports whether any operation's payload arrives as events. +func hasEventStreams(ops []*ir.OperationDef) bool { + return slices.ContainsFunc(ops, func(op *ir.OperationDef) bool { return op.EventType != "" }) +} + // hasPaginatedOps returns true if any operation has pagination configured. func hasPaginatedOps(ops []*ir.OperationDef) bool { for _, op := range ops { diff --git a/internal/generator/generator.go b/internal/generator/generator.go index 3c161d8..f0e7c4d 100644 --- a/internal/generator/generator.go +++ b/internal/generator/generator.go @@ -53,6 +53,7 @@ func (g *Generator) Generate() ([]GeneratedFile, error) { {"errors.go.tmpl", "errors.go"}, {"responses.go.tmpl", "responses.go"}, {"webhooks.go.tmpl", "webhooks.go"}, + {"streaming.go.tmpl", "streaming.go"}, } var files []GeneratedFile diff --git a/internal/ir/operations.go b/internal/ir/operations.go index 3d12b0c..3178eb5 100644 --- a/internal/ir/operations.go +++ b/internal/ir/operations.go @@ -19,7 +19,10 @@ type OperationDef struct { SecurityReqs [][]SecurityReq // OR of (AND of scheme refs) // NoAuth records that the operation declares an empty security requirement, // which overrides the document's to say it takes no credential. - NoAuth bool + NoAuth bool + // EventType is the Go type of one server-sent event's payload, set when a + // success response offers text/event-stream. + EventType string Deprecated bool Pagination *PaginationDef // nil if not paginated } diff --git a/internal/templates/client.go.tmpl b/internal/templates/client.go.tmpl index af6b01d..67342cd 100644 --- a/internal/templates/client.go.tmpl +++ b/internal/templates/client.go.tmpl @@ -66,6 +66,50 @@ func NewClient(baseURL string, opts ...ClientOption) *Client { // whose spec declares an empty security requirement passes authenticated false, // since it says it takes no credential. func (c *Client) do(ctx context.Context, method string, path string, body any, contentType string, result any, accept string, authenticated bool, headers ...http.Header) error { + resp, err := c.send(ctx, method, path, body, contentType, accept, authenticated, headers...) + if err != nil { + return err + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(resp.Body) + if err != nil { + return fmt.Errorf("reading response body: %w", err) + } + + if result != nil && len(respBody) > 0 { + ct := resp.Header.Get("Content-Type") + switch { + case strings.Contains(ct, "json") || ct == "": + if err := json.Unmarshal(respBody, result); err != nil { + return fmt.Errorf("decoding response body: %w", err) + } + case isRawResult(result, respBody): + // A body the operation types as bytes or text is itself the value, + // so it is taken verbatim rather than parsed. + default: + // A server that labels JSON as something else is common enough to + // try anyway, and the media types name each other when it fails. + if err := json.Unmarshal(respBody, result); err != nil { + return fmt.Errorf("decoding response body: asked for %s and the server sent %s: %w", accept, ct, err) + } + } + } + + return nil +} +{{ if hasEventStreams .Operations }} +// doStream executes an HTTP request and hands back the response with its body +// unread, for an operation whose payload arrives an event at a time. The caller +// closes it. +func (c *Client) doStream(ctx context.Context, method string, path string, body any, contentType string, accept string, authenticated bool, headers ...http.Header) (*http.Response, error) { + return c.send(ctx, method, path, body, contentType, accept, authenticated, headers...) +} +{{ end }} +// send runs the request, retrying as the config allows, and returns the response +// with its body still unread. A status outside 2xx comes back as an *APIError +// carrying the body, so a body is left open only for a response that succeeded. +func (c *Client) send(ctx context.Context, method string, path string, body any, contentType string, accept string, authenticated bool, headers ...http.Header) (*http.Response, error) { fullURL := c.baseURL + path var payload []byte @@ -73,7 +117,7 @@ func (c *Client) do(ctx context.Context, method string, path string, body any, c var err error payload, contentType, err = encodeRequestBody(body, contentType) if err != nil { - return err + return nil, err } } @@ -99,7 +143,7 @@ func (c *Client) do(ctx context.Context, method string, path string, body any, c for attempt := 0; attempt < maxAttempts; attempt++ { // Check context cancellation before each attempt. if err := ctx.Err(); err != nil { - return err + return nil, err } var bodyReader io.Reader @@ -109,7 +153,7 @@ func (c *Client) do(ctx context.Context, method string, path string, body any, c req, err := http.NewRequestWithContext(ctx, method, fullURL, bodyReader) if err != nil { - return fmt.Errorf("creating request: %w", err) + return nil, fmt.Errorf("creating request: %w", err) } if payload != nil { @@ -134,7 +178,7 @@ func (c *Client) do(ctx context.Context, method string, path string, body any, c {{ if .AuthSchemes }} if c.auth != nil && authenticated { if err := c.auth.Apply(req); err != nil { - return fmt.Errorf("applying auth: %w", err) + return nil, fmt.Errorf("applying auth: %w", err) } } {{ end }} @@ -150,28 +194,26 @@ func (c *Client) do(ctx context.Context, method string, path string, body any, c select { case <-ctx.Done(): timer.Stop() - return ctx.Err() + return nil, ctx.Err() case <-timer.C: } continue } - return fmt.Errorf("executing request: %w", err) - } - - respBody, err := io.ReadAll(resp.Body) - resp.Body.Close() - if err != nil { - return fmt.Errorf("reading response body: %w", err) + return nil, fmt.Errorf("executing request: %w", err) } // Check if we should retry. if c.retryConfig != nil && attempt < maxAttempts-1 && shouldRetryStatus(method, resp.StatusCode, *c.retryConfig) && withinRetryBudget(resp, *c.retryConfig) { delay := retryDelay(attempt, *c.retryConfig, resp) + // The next attempt replaces this response, so its body is drained + // rather than read: draining lets the connection be reused. + io.Copy(io.Discard, resp.Body) + resp.Body.Close() timer := time.NewTimer(delay) select { case <-ctx.Done(): timer.Stop() - return ctx.Err() + return nil, ctx.Err() case <-timer.C: } continue @@ -180,30 +222,16 @@ func (c *Client) do(ctx context.Context, method string, path string, body any, c captureResponse(ctx, resp) if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return &APIError{StatusCode: resp.StatusCode, Status: resp.Status, Body: respBody} - } - - if result != nil && len(respBody) > 0 { - ct := resp.Header.Get("Content-Type") - switch { - case strings.Contains(ct, "json") || ct == "": - if err := json.Unmarshal(respBody, result); err != nil { - return fmt.Errorf("decoding response body: %w", err) - } - case isRawResult(result, respBody): - // A body the operation types as bytes or text is itself the value, - // so it is taken verbatim rather than parsed. - default: - // A server that labels JSON as something else is common enough to - // try anyway, and the media types name each other when it fails. - if err := json.Unmarshal(respBody, result); err != nil { - return fmt.Errorf("decoding response body: asked for %s and the server sent %s: %w", accept, ct, err) - } + respBody, readErr := io.ReadAll(resp.Body) + resp.Body.Close() + if readErr != nil { + return nil, fmt.Errorf("reading response body: %w", readErr) } + return nil, &APIError{StatusCode: resp.StatusCode, Status: resp.Status, Body: respBody} } - return nil + return resp, nil } - return fmt.Errorf("max retries exceeded") + return nil, fmt.Errorf("max retries exceeded") } diff --git a/internal/templates/operations.go.tmpl b/internal/templates/operations.go.tmpl index ad28840..6c4a2ee 100644 --- a/internal/templates/operations.go.tmpl +++ b/internal/templates/operations.go.tmpl @@ -8,23 +8,10 @@ import ( "net/http" "net/url" ) -{{ range .Operations }} +{{ define "requestSetup" }} {{- $hasParams := or .QueryParams .HeaderParams .CookieParams }} {{- $hasReqParam := or (hasRequiredQueryParams .) (hasRequiredHeaderParams .) (hasRequiredCookieParams .) }} {{- $needHeaders := or .HeaderParams .CookieParams }} -{{ if $hasParams }}// {{ .Name }}Params contains the parameters for the {{ .Name }} operation. -// Required parameters are value fields; optional parameters are pointers. -type {{ .Name }}Params struct { -{{ range .QueryParams }}{{ $pDoc := paramDocComment . }}{{ if $pDoc }}{{ indent $pDoc }} -{{ end }} {{ .FieldName }} {{ paramType . }} `json:"{{ .OrigName }}{{ if not .Required }},omitempty{{ end }}"` -{{ end }}{{ range .HeaderParams }}{{ $pDoc := paramDocComment . }}{{ if $pDoc }}{{ indent $pDoc }} -{{ end }} {{ .FieldName }} {{ paramType . }} `json:"{{ .OrigName }}{{ if not .Required }},omitempty{{ end }}"` -{{ end }}{{ range .CookieParams }}{{ $pDoc := paramDocComment . }}{{ if $pDoc }}{{ indent $pDoc }} -{{ end }} {{ .FieldName }} {{ paramType . }} `json:"{{ .OrigName }}{{ if not .Required }},omitempty{{ end }}"` -{{ end }}} -{{ end }} -{{ $opDoc := opDocComment . }}{{ if $opDoc }}{{ $opDoc }} -{{ end }}func (c *Client) {{ .Name }}(ctx context.Context{{ range .PathParams }}, {{ .Name }} {{ .Type }}{{ end }}{{ if hasBody . }}, body {{ if .RequestBody }}{{ if not .RequestBody.Required }}*{{ end }}{{ .RequestBody.TypeName }}{{ else }}any{{ end }}{{ end }}{{ if $hasReqParam }}, params {{ .Name }}Params{{ else if $hasParams }}, opts ...{{ .Name }}Params{{ end }}) {{ if successType . }}(*{{ successType . }}, error){{ else }}error{{ end }} { path := "{{ .Path }}" {{ range .PathParams }} path = pathReplace(path, "{{ .OrigName }}", "{{ .Style }}", {{ .Explode }}, {{ .Name }}) {{ end }} @@ -48,6 +35,25 @@ type {{ .Name }}Params struct { {{ else }} addCookieHeader(headers, "{{ .OrigName }}", params.{{ .FieldName }}) {{ end }} {{ end }}{{ end }} +{{ end }} +{{ range .Operations }} +{{- $hasParams := or .QueryParams .HeaderParams .CookieParams }} +{{- $hasReqParam := or (hasRequiredQueryParams .) (hasRequiredHeaderParams .) (hasRequiredCookieParams .) }} +{{- $needHeaders := or .HeaderParams .CookieParams }} +{{ if $hasParams }}// {{ .Name }}Params contains the parameters for the {{ .Name }} operation. +// Required parameters are value fields; optional parameters are pointers. +type {{ .Name }}Params struct { +{{ range .QueryParams }}{{ $pDoc := paramDocComment . }}{{ if $pDoc }}{{ indent $pDoc }} +{{ end }} {{ .FieldName }} {{ paramType . }} `json:"{{ .OrigName }}{{ if not .Required }},omitempty{{ end }}"` +{{ end }}{{ range .HeaderParams }}{{ $pDoc := paramDocComment . }}{{ if $pDoc }}{{ indent $pDoc }} +{{ end }} {{ .FieldName }} {{ paramType . }} `json:"{{ .OrigName }}{{ if not .Required }},omitempty{{ end }}"` +{{ end }}{{ range .CookieParams }}{{ $pDoc := paramDocComment . }}{{ if $pDoc }}{{ indent $pDoc }} +{{ end }} {{ .FieldName }} {{ paramType . }} `json:"{{ .OrigName }}{{ if not .Required }},omitempty{{ end }}"` +{{ end }}} +{{ end }} +{{ $opDoc := opDocComment . }}{{ if $opDoc }}{{ $opDoc }} +{{ end }}func (c *Client) {{ .Name }}(ctx context.Context{{ range .PathParams }}, {{ .Name }} {{ .Type }}{{ end }}{{ if hasBody . }}, body {{ if .RequestBody }}{{ if not .RequestBody.Required }}*{{ end }}{{ .RequestBody.TypeName }}{{ else }}any{{ end }}{{ end }}{{ if $hasReqParam }}, params {{ .Name }}Params{{ else if $hasParams }}, opts ...{{ .Name }}Params{{ end }}) {{ if successType . }}(*{{ successType . }}, error){{ else }}error{{ end }} { +{{ template "requestSetup" . }} {{- $errType := errorType . -}} {{ if successType . }} var result {{ successType . }} if err := c.do(ctx, "{{ .HTTPMethod }}", path, {{ if hasBody . }}body{{ else }}nil{{ end }}, {{ printf "%q" (requestContentType .) }}, &result, {{ printf "%q" (successContentType .) }}, {{ not .NoAuth }}{{ if $needHeaders }}, headers{{ end }}); err != nil { @@ -59,4 +65,15 @@ type {{ .Name }}Params struct { } return nil {{ end }}} -{{ end }} +{{ if .EventType }} +// {{ .Name }}Stream is {{ .Name }} with the response taken as it arrives, one +// server-sent event at a time. Close the stream when finished with it. +func (c *Client) {{ .Name }}Stream(ctx context.Context{{ range .PathParams }}, {{ .Name }} {{ .Type }}{{ end }}{{ if hasBody . }}, body {{ if .RequestBody }}{{ if not .RequestBody.Required }}*{{ end }}{{ .RequestBody.TypeName }}{{ else }}any{{ end }}{{ end }}{{ if $hasReqParam }}, params {{ .Name }}Params{{ else if $hasParams }}, opts ...{{ .Name }}Params{{ end }}) (*EventStream[{{ .EventType }}], error) { +{{ template "requestSetup" . }} + resp, err := c.doStream(ctx, "{{ .HTTPMethod }}", path, {{ if hasBody . }}body{{ else }}nil{{ end }}, {{ printf "%q" (requestContentType .) }}, "text/event-stream", {{ not .NoAuth }}{{ if $needHeaders }}, headers{{ end }}) + if err != nil { + return nil, {{ if $errType }}parse{{ $errType }}(err){{ else }}err{{ end }} + } + return newEventStream[{{ .EventType }}](resp), nil +} +{{ end }}{{ end }} diff --git a/internal/templates/reserved.go b/internal/templates/reserved.go index ac7d1ef..d175389 100644 --- a/internal/templates/reserved.go +++ b/internal/templates/reserved.go @@ -27,6 +27,7 @@ var ReservedIdentifiers = []string{ "ErrServiceUnavailable", "ErrTooManyRequests", "ErrUnauthorized", + "EventStream", "FormFile", "Middleware", "NewClient", diff --git a/internal/templates/streaming.go.tmpl b/internal/templates/streaming.go.tmpl new file mode 100644 index 0000000..a16e4f1 --- /dev/null +++ b/internal/templates/streaming.go.tmpl @@ -0,0 +1,135 @@ +// Code generated by openapi-client-generator. DO NOT EDIT. + +package {{ .Name }} +{{ if hasEventStreams .Operations }} +import ( + "bufio" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" +) + +// eventStreamDone is the payload OpenAI-compatible APIs send to close a stream. +// It is a marker rather than an event, so it ends iteration instead of failing +// to decode as one. +const eventStreamDone = "[DONE]" + +// EventStream reads server-sent events as the server writes them. Close it when +// finished, whether or not the stream ran to its end. +// +// stream, err := client.Whatever(ctx) +// defer stream.Close() +// +// for { +// event, err := stream.Next() +// if errors.Is(err, io.EOF) { +// break +// } +// if err != nil { +// return err +// } +// } +type EventStream[T any] struct { + body io.ReadCloser + reader *bufio.Reader + name string + id string +} + +func newEventStream[T any](resp *http.Response) *EventStream[T] { + return &EventStream[T]{body: resp.Body, reader: bufio.NewReader(resp.Body)} +} + +// Next returns the next event's payload, decoded as the spec declares it, and +// io.EOF once the server closes the stream. +func (s *EventStream[T]) Next() (T, error) { + var zero T + var data strings.Builder + var name, id string + + deliver := func() (T, error) { + payload := data.String() + if payload == eventStreamDone { + return zero, io.EOF + } + s.name, s.id = name, id + return decodeEvent[T](payload) + } + + for { + line, err := s.reader.ReadString('\n') + atEOF := errors.Is(err, io.EOF) + if err != nil && !atEOF { + return zero, err + } + + switch line = strings.TrimRight(line, "\r\n"); { + case line == "": + // A blank line ends an event. Between events, and for the keep-alives + // some servers send, there is nothing to end. + if !atEOF && data.Len() > 0 { + return deliver() + } + case strings.HasPrefix(line, ":"): + // A comment, which is how keep-alives are written. + default: + field, value, _ := strings.Cut(line, ":") + value = strings.TrimPrefix(value, " ") + switch field { + case "data": + // Several data lines are one payload, joined by newlines. + if data.Len() > 0 { + data.WriteByte('\n') + } + data.WriteString(value) + case "event": + name = value + case "id": + id = value + } + } + + if atEOF { + // A stream may end without the blank line that would have dispatched + // what it already sent, so what arrived is delivered before EOF. + if data.Len() == 0 { + return zero, io.EOF + } + return deliver() + } + } +} + +// EventName returns the event field of the event Next last returned, or "" for +// a stream that does not name its events. +func (s *EventStream[T]) EventName() string { + return s.name +} + +// EventID returns the id field of the event Next last returned, or "". +func (s *EventStream[T]) EventID() string { + return s.id +} + +// Close ends the stream and releases the connection. +func (s *EventStream[T]) Close() error { + return s.body.Close() +} + +// decodeEvent renders one event's data as the type the spec declares. A stream +// of text declares string, where the data is the value rather than a document. +func decodeEvent[T any](data string) (T, error) { + var value T + if text, ok := any(&value).(*string); ok { + *text = data + return value, nil + } + if err := json.Unmarshal([]byte(data), &value); err != nil { + return value, fmt.Errorf("decoding event: %w", err) + } + return value, nil +} +{{ end }}