feat(generator): a text/event-stream response is read as it arrives

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

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.

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

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)
+	}
+}
+`)
+}

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 {