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 }}