diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl index 841dd45..492d709 100644 --- a/.beads/issues.jsonl +++ b/.beads/issues.jsonl @@ -1,4 +1,5 @@ {"id":"openapi-generator-dsu","title":"OpenAPI 3.1 type:[X,null] on required properties generates non-Option field","description":"For OpenAPI 3.1 schemas where a property is BOTH listed in 'required' AND declared nullable via the 3.1 type-array form (type: [\"string\",\"null\"]), the generator emits a non-Option field. Deserialization then fails against real responses that send null.\n\nVERIFIED AGAINST THE LIVE RUNPOD v2 API. Two operations fail outright today:\n- GET /v2/catalog/gpus -\u003e GpuType.pool is null in production; spec says type: [\"string\",\"null\"], required. Generated: 'pub pool: String'.\n- GET /v2/pods -\u003e Pod.template is null in production. Generated: 'pub template: String'.\n\nConfirmed affected fields in https://api.runpod.io/v2/openapi.json (4, walking allOf branches):\n GpuType.pool, Pod.dataCenterId, Pod.template, Pod.startedAt\n\nPod.startedAt is the worst latent case: any pod that has not started returns null, which takes out the entire listPods call for that account.\n\nNote Pod composes via allOf, so the required list and properties live in an allOf branch — a fix must traverse composition, not just top-level component properties.\n\nRoot cause reported by live-testing agent (unverified by me): nullability is computed as is_nullable() || is_nullable_pattern() in src/analysis.rs (~1936-1937, ~2506-2514), covering only 3.0 'nullable: true' and the anyOf-with-null shape. Schema::type_array_contains_null() (src/openapi.rs ~649-656) is called from only one site (analysis.rs ~1555). generator.rs (~2497-2517) then yields a plain String.\n\nClosed issue openapi-generator-bgo fixed only the anyOf case; this type-array case is uncovered.\n\nSeverity: 3.1 is the modern dialect and this silently produces a client that cannot read production responses. This is first-run breakage of the worst kind — it compiles, then fails at runtime.","acceptance_criteria":"A required property declared type: [\"string\",\"null\"] generates Option\u003cString\u003e, including when declared inside an allOf branch. Regression test covers both the direct component-schema case and the allOf-composed case. Generating the RunPod v2 spec yields Option for GpuType.pool, Pod.template, Pod.startedAt, and Pod.dataCenterId.","status":"closed","priority":0,"issue_type":"bug","owner":"james@littlebearlabs.io","created_at":"2026-07-26T23:51:03Z","created_by":"James Lal","updated_at":"2026-07-27T00:59:39Z","closed_at":"2026-07-27T00:59:39Z","close_reason":"Fixed in fad3ed3; verified against the live RunPod v2 API (list_gpu_types and list_pods now deserialize, with real nulls present in the responses). 362 tests pass.","dependency_count":0,"dependent_count":0,"comment_count":0} +{"id":"openapi-generator-1lb","title":"Upgrade generated HTTP dependency versions for issue 50","description":"Update generated dependency requirements to the reqwest 0.13 ecosystem and thiserror 2 now that generated SSE no longer depends on reqwest-eventsource, then compile exact generated manifests and run the full validation suite.","acceptance_criteria":"Generated dependencies use mutually compatible current reqwest, reqwest-middleware, reqwest-retry, reqwest-tracing, and thiserror versions; exact generated manifests compile; all-feature tests, formatting, Clippy, and live SSE verification pass; changes are pushed to PR #51.","notes":"Upgraded emitted HTTP stack to reqwest 0.13, reqwest-middleware 0.5, reqwest-retry 0.9, reqwest-tracing 0.7, and thiserror 2; migrated TLS feature to rustls; inferred reqwest/middleware query, form, JSON, multipart, and stream features from generated code. Updated root dependency/lockfile, examples, scratch manifests, install smoke, and benchmark harness. Exact generated dependency fragments compile. Full cargo test --all-features, production Clippy -D warnings, fmt, package/install smoke, example compile, and live OpenAI/Anthropic SSE on reqwest 0.13 pass.","status":"closed","priority":1,"issue_type":"task","assignee":"James Lal","owner":"james@littlebearlabs.io","created_at":"2026-07-29T15:33:31Z","created_by":"James Lal","updated_at":"2026-07-29T15:44:47Z","started_at":"2026-07-29T15:33:35Z","closed_at":"2026-07-29T15:44:47Z","close_reason":"The full generated and root HTTP dependency stack is upgraded and verified across exact manifests, integration tests, packaging, and the live SSE backend.","dependency_count":0,"dependent_count":0,"comment_count":0} {"id":"openapi-generator-1fz","title":"Make scratch Cargo directories portable across CI runners","description":"PR #46 clean Linux runner cannot create hard-coded /private/tmp Cargo target/build directories in generated-crate integration tests. Derive all scratch build paths from each TempDir and verify every affected test plus full CI.","status":"closed","priority":1,"issue_type":"bug","assignee":"James Lal","owner":"james@littlebearlabs.io","created_at":"2026-07-29T00:10:21Z","created_by":"James Lal","updated_at":"2026-07-29T00:17:39Z","started_at":"2026-07-29T00:10:24Z","closed_at":"2026-07-29T00:17:39Z","close_reason":"Replaced all hard-coded /private/tmp Cargo build and target paths with per-test TempDir paths; focused fresh-build tests, full all-features suite, fmt, and clippy pass.","dependency_count":0,"dependent_count":0,"comment_count":0} {"id":"openapi-generator-6vy","title":"Fix clean-runner scratch-crate dependency resolution","description":"PR #46 CI fails because newly added generated-crate integration tests invoke Cargo with --offline before emitted dependencies such as async-trait exist in a clean runner cache. Remove unconditional offline resolution from the new scratch tests and verify with a cold Cargo cache.","status":"closed","priority":1,"issue_type":"bug","assignee":"James Lal","owner":"james@littlebearlabs.io","created_at":"2026-07-28T23:59:37Z","created_by":"James Lal","updated_at":"2026-07-29T00:05:51Z","started_at":"2026-07-28T23:59:42Z","closed_at":"2026-07-29T00:05:51Z","close_reason":"Removed unconditional offline mode from all new scratch-crate Cargo invocations; cold-cache focused suite and full all-features CI tests pass.","dependency_count":0,"dependent_count":0,"comment_count":0} {"id":"openapi-generator-60d","title":"Compile-check all 55 supported corpus specs","description":"Run scripts/spec-compile.sh across the complete supported OpenAPI corpus after the Storyden generator fixes, with generation and isolated cargo check for every supported spec.","status":"closed","priority":1,"issue_type":"task","assignee":"James Lal","owner":"james@littlebearlabs.io","created_at":"2026-07-28T22:51:36Z","created_by":"James Lal","updated_at":"2026-07-28T23:15:21Z","started_at":"2026-07-28T22:51:41Z","closed_at":"2026-07-28T23:15:21Z","close_reason":"All 55 supported OpenAPI corpus documents generated and compiled successfully with isolated manifests; Microsoft Graph was force-compiled despite the standard resource gate. Gitea remains the expected Swagger 2.0 exclusion.","dependency_count":0,"dependent_count":0,"comment_count":0} @@ -53,6 +54,8 @@ {"id":"openapi-generator-cv4","title":"Support exploded query parameters (GH #27)","description":"GH issue 27: query params with style=form, explode=true (OAS defaults) and object schemas currently map to Option\u003cimpl AsRef\u003cstr\u003e\u003e and are sent as one opaque string. Per OAS/RFC6570 form-explode, each object property must become its own query pair (?color=red). Fix: analyzer synthesizes/resolves a typed struct for object query params with form+explode semantics; client generator emits req.query(\u0026struct) so reqwest/serde_urlencoded serializes properties as individual pairs.","notes":"Implemented on branch issue-27-exploded-query-params, draft PR https://github.com/gpu-cli/openapi-to-rust/pull/28. Close when PR merges. Follow-ups: openapi-generator-anu (deepObject/explode=false/arrays), openapi-generator-0jz (server side).","status":"closed","priority":1,"issue_type":"feature","assignee":"James Lal","owner":"james@littlebearlabs.io","created_at":"2026-07-12T23:22:24Z","created_by":"James Lal","updated_at":"2026-07-13T04:21:19Z","started_at":"2026-07-12T23:22:49Z","closed_at":"2026-07-13T04:21:19Z","close_reason":"Released in v0.6.0 (PR #28)","dependency_count":0,"dependent_count":0,"comment_count":0} {"id":"openapi-generator-5q8","title":"DateStrategy::Time emits broken serde codec for time::Date/time::Time (GH #25)","description":"GitHub issue gpu-cli/openapi-to-rust#25: fields with format: date/time under DateStrategy::Time get #[serde(with = \"time::serde::iso8601\")], but that module only supports OffsetDateTime — generated code fails to compile. Fix: emit time::serde::format_description! helper codec modules (time_date_format / time_time_format) into generated code, and correct the time dep requirement features (serde alone doesn't even enable rfc3339).","notes":"Fixed in PR https://github.com/gpu-cli/openapi-to-rust/pull/26 (branch worktree-issue-25-time-date-serde). Close when PR merges.","status":"closed","priority":1,"issue_type":"bug","assignee":"James Lal","owner":"james@littlebearlabs.io","created_at":"2026-07-11T18:36:28Z","created_by":"James Lal","updated_at":"2026-07-11T18:57:02Z","started_at":"2026-07-11T18:36:45Z","closed_at":"2026-07-11T18:57:02Z","close_reason":"Fixed in PR #26, merged to main, released in v0.5.3","dependency_count":0,"dependent_count":0,"comment_count":0} {"id":"openapi-generator-dpd","title":"Hybrid string-or-object discriminated unions deserialize-fail","description":"When an anyOf/oneOf union contains a discriminator AND a non-object branch (e.g. string-enum like ToolChoiceOptions), the generator emits a tagged enum that cannot deserialize the string form. Real-world hit: OpenAI ToolChoiceParam returns 'auto' in Response.tool_choice, but generated type is #[serde(tag=\"type\")] enum with no untagged String variant. Need to fall back to #[serde(untagged)] when the union mixes string/scalar branches with tagged-object branches, OR add a String fallback variant before the tagged variants.","notes":"Live repro: `ToolChoiceParam` from openai.yaml line 52518. anyOf has 8 branches; the first (`ToolChoiceOptions`) is a string-enum (\"none\"|\"auto\"|\"required\"), the rest are objects with discriminator propertyName=type. Generator emits `#[serde(tag=\"type\")] enum ToolChoiceParam { ToolChoiceOptions(ToolChoiceOptions), ... }` which cannot deserialize the string \"auto\" because serde tries to read a \"type\" field from a JSON string. Fix: when an anyOf/oneOf branch is a non-object schema (string/number/etc), the generator must emit `#[serde(untagged)]` with the scalar branch first OR add a String variant before the tagged variants. Hit on real OpenAI Responses API `Response.tool_choice` field.","status":"closed","priority":1,"issue_type":"bug","assignee":"James Lal","owner":"james@littlebearlabs.io","created_at":"2026-05-10T23:47:35Z","created_by":"James Lal","updated_at":"2026-05-11T00:12:46Z","started_at":"2026-05-10T23:54:10Z","closed_at":"2026-05-11T00:12:46Z","close_reason":"Fixed in src/analysis.rs: (dpd) analyze_oneof_union now downgrades to untagged when any branch is non-object — verified live against OpenAI Response.tool_choice='auto' which now deserializes as ToolChoiceParam::ToolChoiceOptions(Auto). (bgo) merge_schema_into_properties now ORs in is_nullable_pattern() for allOf-merged props — verified live against OpenAI Response.incomplete_details which is now Option\u003cResponseIncompleteDetails\u003e and deserializes null cleanly. All 4 smoke tests (OpenAI+Anthropic, sync+stream) pass.","dependency_count":0,"dependent_count":0,"comment_count":0} +{"id":"openapi-generator-ew5","title":"Add typed SSE events and reconnection","description":"Extend the generated standalone SSE client with raw and typed event metadata, SSE retry parsing, automatic retryable reconnection, and Last-Event-ID resumption while preserving existing generated stream APIs.","acceptance_criteria":"Generated sse.rs exposes SseEvent metadata and raw/JSON APIs; retryable disconnects reconnect with Last-Event-ID and server retry delay; explicit completion does not reconnect; deterministic runtime tests and live OpenAI/Anthropic validation pass.","notes":"Generated sse.rs now exposes SseEvent\u003cT\u003e, stream_raw, stream_json, and explicit reconnecting variants while preserving stream\u003cT\u003e. Reconnection parses id/retry, sends Last-Event-ID, applies bounded server/client delays and retryable HTTP/transport retries, stops at [DONE], and is activated for generated clients when reconnection_config is set. Deterministic chunked tests validate metadata and a two-connection resume; exact generated dependencies compile; full cargo test --all-features, production Clippy -D warnings, fmt, and live OpenAI/Anthropic typed streaming all pass.","status":"closed","priority":2,"issue_type":"feature","assignee":"James Lal","owner":"james@littlebearlabs.io","created_at":"2026-07-29T14:51:49Z","created_by":"James Lal","updated_at":"2026-07-29T15:02:31Z","started_at":"2026-07-29T14:51:52Z","closed_at":"2026-07-29T15:02:31Z","close_reason":"Typed SSE metadata and Last-Event-ID reconnection are implemented, tested locally, and validated against the live backend.","dependency_count":0,"dependent_count":0,"comment_count":0} +{"id":"openapi-generator-4bj","title":"Replace reqwest-eventsource with internal SSE implementation","description":"Implement issue #50 groundwork by removing generated reqwest-eventsource usage, parsing SSE internally on a reqwest byte stream, and validating generated OpenAI- and Anthropic-compatible streaming clients against the real backend at http://192.168.68.55:1234.","acceptance_criteria":"Generated SSE code no longer depends on reqwest-eventsource; unit and compile tests cover chunk framing, events, errors, and limits; generated OpenAI and Anthropic streaming requests work against the configured real backend; relevant quality gates pass.","notes":"Implemented conditional generated sse.rs with reusable SseClient, internal incremental SSE framing, bounded non-2xx bodies, content-type validation, ping filtering, and OpenAI [DONE] termination. streaming.rs imports SseClient and re-exports its transport error. Verified chunk/CRLF boundary behavior, exact dependency fragments, full cargo test --all-features, production Clippy, fmt, and live OpenAI + Anthropic streaming at the configured backend (16 OpenAI chunks; Anthropic message_start/content_block/message_delta/message_stop sequence).","status":"closed","priority":2,"issue_type":"feature","assignee":"James Lal","owner":"james@littlebearlabs.io","created_at":"2026-07-29T14:24:22Z","created_by":"James Lal","updated_at":"2026-07-29T14:41:15Z","started_at":"2026-07-29T14:24:29Z","closed_at":"2026-07-29T14:41:15Z","close_reason":"Generated SSE transport is internal, conditionally emitted as sse.rs, covered by local runtime/compile tests, and validated against the live OpenAI/Anthropic-compatible backend.","dependency_count":0,"dependent_count":0,"comment_count":0} {"id":"openapi-generator-d7h","title":"Align wide comparison blocks with narrative column","description":"Comparison headings and prose now align to the standard reading column, but wide tables and other structured blocks still begin at the outer wide-shell edge. Give every article block a shared left edge while preserving extra width to the right for tables and comparison grids.","acceptance_criteria":"On both comparison pages, tables and other structured blocks share the same left edge as their headings and prose; wide blocks continue using available space to the right; mobile overflow remains accessible; build, HTML, SEO, and layout checks pass.","notes":"Applied one responsive --wide-content-inset to every direct child of .prose-wide. Narrative children remain capped at 72ch, while tables, decision grids, command comparisons, code blocks, and pagination use the remaining width to the right. Verified both tables align with their headings in a 2000x3000 local Chrome render. npm run build, audit:html, audit:seo, and the post-change layout detector all pass with zero findings.","status":"closed","priority":2,"issue_type":"bug","assignee":"James Lal","owner":"james@littlebearlabs.io","created_at":"2026-07-29T03:55:54Z","created_by":"James Lal","updated_at":"2026-07-29T03:57:39Z","started_at":"2026-07-29T03:55:59Z","closed_at":"2026-07-29T03:57:39Z","close_reason":"All comparison-page article blocks now share one left edge while structured content retains its wider rightward span; browser and automated checks pass.","dependency_count":0,"dependent_count":0,"comment_count":0} {"id":"openapi-generator-bkv","title":"Align comparison prose with page heading","description":"The wide comparison layout introduced a mismatched left edge: the hero remains on the standard content column while section headings, callouts, and body prose begin at the expanded table edge. Align ordinary prose content with the hero while preserving wide tables and responsive overflow.","acceptance_criteria":"On both comparison pages, hero copy, section headings, callouts, and body prose share a coherent left edge; comparison tables and other wide data structures still use the expanded track; mobile remains readable; build and site audits pass.","notes":"Fixed the wide comparison layout by adding a shared wide-shell token and applying a responsive narrative inset to direct prose children. The inset is half the difference between the 1180px standard shell and the active wide shell, capped at 130px and collapsing to zero on narrow viewports. Tables, decision grids, command comparisons, and pagination remain on the wide article track. Verified with local Chrome screenshots at 2000px, layout detector (0 findings), npm run build, npm run audit:html, and npm run audit:seo.","status":"closed","priority":2,"issue_type":"bug","assignee":"James Lal","owner":"james@littlebearlabs.io","created_at":"2026-07-29T03:44:46Z","created_by":"James Lal","updated_at":"2026-07-29T03:48:26Z","started_at":"2026-07-29T03:44:52Z","closed_at":"2026-07-29T03:48:26Z","close_reason":"Hero, callout, section headings, and body prose now share one left edge while wide comparison content retains the expanded track; browser and automated checks pass.","dependency_count":0,"dependent_count":0,"comment_count":0} {"id":"openapi-generator-qav","title":"Let comparison tables use the available desktop width","description":"Fix comparison-page layout so wide tables use the available content column instead of being constrained to the 72ch prose measure, while preserving readable paragraph line lengths, the desktop TOC, and mobile horizontal scrolling.","acceptance_criteria":"Both comparison tables fit their desktop content area without premature horizontal clipping; prose remains capped at the reading measure; mobile tables remain accessible and scrollable; Astro, HTML, and SEO quality gates pass.","notes":"Implemented opt-in wide documentation layout for comparison pages. Article track now fills up to a 1440px shell while direct prose remains capped at the reading width; desktop TOC collapses at 1280px for comparison pages; tables retain accessible horizontal scrolling and narrower mobile sticky columns. Verified with layout detector (0 findings), npm run build, npm run audit:html, and npm run audit:seo.","status":"closed","priority":2,"issue_type":"bug","assignee":"James Lal","owner":"james@littlebearlabs.io","created_at":"2026-07-29T02:29:03Z","created_by":"James Lal","updated_at":"2026-07-29T02:32:31Z","started_at":"2026-07-29T02:29:10Z","closed_at":"2026-07-29T02:32:31Z","close_reason":"Comparison tables now use available desktop width while prose readability and mobile overflow behavior are preserved; all quality gates pass.","dependency_count":0,"dependent_count":0,"comment_count":0} diff --git a/Cargo.lock b/Cargo.lock index 3b5343f..fdacd52 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -105,6 +105,29 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" +[[package]] +name = "aws-lc-rs" +version = "1.17.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "00bdb5da18dac48ca2cc7cd4a98e533e8635a58e2361d13a1a4ee3888e0d72f1" +dependencies = [ + "aws-lc-sys", + "zeroize", +] + +[[package]] +name = "aws-lc-sys" +version = "0.43.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "43103168cc76fe62678a375e722fc9cb3a0146159ac5828bc4f0dfd755c2224c" +dependencies = [ + "cc", + "cmake", + "dunce", + "fs_extra", + "pkg-config", +] + [[package]] name = "base64" version = "0.22.1" @@ -128,9 +151,9 @@ checksum = "5e764a1d40d510daf35e07be9eb06e75770908c27d411ee6c92109c9840eaaf7" [[package]] name = "bitflags" -version = "2.10.0" +version = "2.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "812e12b5285cc515a9c72a5c1d3b6d46a19dac5acfef5265968c166106e31dd3" +checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" [[package]] name = "borrow-or-share" @@ -163,6 +186,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "aebf35691d1bfb0ac386a69bac2fde4dd276fb618cf8bf4f5318fe285e821bb2" dependencies = [ "find-msvc-tools", + "jobserver", + "libc", "shlex", ] @@ -229,12 +254,31 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a822ea5bc7590f9d40f1ba12c0dc3c2760f3482c6984db1573ad11031420831" +[[package]] +name = "cmake" +version = "0.1.58" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678" +dependencies = [ + "cc", +] + [[package]] name = "colorchoice" version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75" +[[package]] +name = "combine" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd" +dependencies = [ + "bytes", + "memchr", +] + [[package]] name = "console" version = "0.15.11" @@ -257,6 +301,22 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "core-foundation" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6" +dependencies = [ + "core-foundation-sys", + "libc", +] + +[[package]] +name = "core-foundation-sys" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" + [[package]] name = "cpufeatures" version = "0.3.0" @@ -283,6 +343,12 @@ dependencies = [ "syn 2.0.115", ] +[[package]] +name = "dunce" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" + [[package]] name = "email_address" version = "0.2.9" @@ -379,6 +445,12 @@ dependencies = [ "num", ] +[[package]] +name = "fs_extra" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" + [[package]] name = "futures-channel" version = "0.3.31" @@ -577,7 +649,6 @@ dependencies = [ "tokio", "tokio-rustls", "tower-service", - "webpki-roots", ] [[package]] @@ -764,6 +835,65 @@ version = "1.0.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92ecc6618181def0457392ccd0ee51198e065e016d1d527a7ac1b6dc7c1f09d2" +[[package]] +name = "jni" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5efd9a482cf3a427f00d6b35f14332adc7902ce91efb778580e180ff90fa3498" +dependencies = [ + "cfg-if", + "combine", + "jni-macros", + "jni-sys", + "log", + "simd_cesu8", + "thiserror 2.0.18", + "walkdir", + "windows-link", +] + +[[package]] +name = "jni-macros" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a00109accc170f0bdb141fed3e393c565b6f5e072365c3bd58f5b062591560a3" +dependencies = [ + "proc-macro2", + "quote", + "rustc_version", + "simd_cesu8", + "syn 2.0.115", +] + +[[package]] +name = "jni-sys" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6377a88cb3910bee9b0fa88d4f42e1d2da8e79915598f65fb0c7ee14c878af2" +dependencies = [ + "jni-sys-macros", +] + +[[package]] +name = "jni-sys-macros" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "38c0b942f458fe50cdac086d2f946512305e5631e720728f2a61aabcd47a6264" +dependencies = [ + "quote", + "syn 2.0.115", +] + +[[package]] +name = "jobserver" +version = "0.1.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c00acbd29eabad4a2392fa0e921c874934dbbf4194312ad20f04a0ed67a3cb3" +dependencies = [ + "getrandom 0.4.1", + "libc", +] + [[package]] name = "js-sys" version = "0.3.85" @@ -1010,7 +1140,7 @@ dependencies = [ "specta", "syn 2.0.115", "tempfile", - "thiserror 1.0.69", + "thiserror 2.0.18", "toml", "toml_edit", "url", @@ -1028,6 +1158,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "openssl-probe" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" + [[package]] name = "outref" version = "0.5.2" @@ -1075,6 +1211,12 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" +[[package]] +name = "pkg-config" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" + [[package]] name = "potential_utf" version = "0.1.4" @@ -1129,6 +1271,7 @@ version = "0.11.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2f4bfc015262b9df63c8845072ce59068853ff5872180c2ce2f13038b970e560" dependencies = [ + "aws-lc-rs", "bytes", "getrandom 0.4.1", "lru-slab", @@ -1277,9 +1420,9 @@ checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" [[package]] name = "reqwest" -version = "0.12.28" +version = "0.13.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" +checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3" dependencies = [ "base64", "bytes", @@ -1299,8 +1442,8 @@ dependencies = [ "quinn", "rustls", "rustls-pki-types", + "rustls-platform-verifier", "serde", - "serde_json", "serde_urlencoded", "sync_wrapper", "tokio", @@ -1312,7 +1455,6 @@ dependencies = [ "wasm-bindgen", "wasm-bindgen-futures", "web-sys", - "webpki-roots", ] [[package]] @@ -1335,6 +1477,15 @@ version = "2.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" +[[package]] +name = "rustc_version" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92" +dependencies = [ + "semver", +] + [[package]] name = "rustix" version = "1.1.3" @@ -1354,14 +1505,26 @@ version = "0.23.42" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138" dependencies = [ + "aws-lc-rs", "once_cell", - "ring", "rustls-pki-types", "rustls-webpki", "subtle", "zeroize", ] +[[package]] +name = "rustls-native-certs" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d" +dependencies = [ + "openssl-probe", + "rustls-pki-types", + "schannel", + "security-framework", +] + [[package]] name = "rustls-pki-types" version = "1.15.0" @@ -1372,12 +1535,40 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rustls-platform-verifier" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26d1e2536ce4f35f4846aa13bff16bd0ff40157cdb14cc056c7b14ba41233ba0" +dependencies = [ + "core-foundation", + "core-foundation-sys", + "jni", + "log", + "once_cell", + "rustls", + "rustls-native-certs", + "rustls-platform-verifier-android", + "rustls-webpki", + "security-framework", + "security-framework-sys", + "webpki-root-certs", + "windows-sys 0.61.2", +] + +[[package]] +name = "rustls-platform-verifier-android" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" + [[package]] name = "rustls-webpki" version = "0.103.13" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" dependencies = [ + "aws-lc-rs", "ring", "rustls-pki-types", "untrusted", @@ -1395,12 +1586,53 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "same-file" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502" +dependencies = [ + "winapi-util", +] + +[[package]] +name = "schannel" +version = "0.1.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91c1b7e4904c873ef0710c1f407dde2e6287de2bebc1bbbf7d430bb7cbffd939" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "scopeguard" version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "security-framework" +version = "3.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d" +dependencies = [ + "bitflags", + "core-foundation", + "core-foundation-sys", + "libc", + "security-framework-sys", +] + +[[package]] +name = "security-framework-sys" +version = "2.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2691df843ecc5d231c0b14ece2acc3efb62c0a398c7e1d875f3983ce020e3" +dependencies = [ + "core-foundation-sys", + "libc", +] + [[package]] name = "semver" version = "1.0.27" @@ -1501,6 +1733,22 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +[[package]] +name = "simd_cesu8" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11031e251abf8611c80f460e19dbdeb54a66db918e49c65a7065b46ac7aec520" +dependencies = [ + "rustc_version", + "simdutf8", +] + +[[package]] +name = "simdutf8" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e" + [[package]] name = "similar" version = "2.7.0" @@ -1921,6 +2169,16 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5c3082ca00d5a5ef149bb8b555a72ae84c9c59f7250f013ac822ac2e49b19c64" +[[package]] +name = "walkdir" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b" +dependencies = [ + "same-file", + "winapi-util", +] + [[package]] name = "want" version = "0.3.1" @@ -2068,14 +2326,23 @@ dependencies = [ ] [[package]] -name = "webpki-roots" -version = "1.0.8" +name = "webpki-root-certs" +version = "1.0.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bf85cb06032201fa7c6f829d7db5a7e5aa45bcc0655327713065f6f0576731bf" +checksum = "b96554aa2acc8ccdb7e1c9a58a7a68dd5d13bccc69cd124cb09406db612a1c9b" dependencies = [ "rustls-pki-types", ] +[[package]] +name = "winapi-util" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "windows-link" version = "0.2.1" diff --git a/Cargo.toml b/Cargo.toml index c6ef985..6c53529 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -37,13 +37,13 @@ once_cell = "1.19" prettyplease = "0.2" proc-macro2 = "1.0" quote = "1.0" -reqwest = { version = "0.12", default-features = false, features = ["blocking", "rustls-tls"], optional = true } +reqwest = { version = "0.13", default-features = false, features = ["blocking", "query", "rustls"], optional = true } url = "2" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" serde_yaml = "0.9" syn = { version = "2.0", features = ["parsing"] } -thiserror = "1.0" +thiserror = "2.0" tempfile = { version = "3.0", optional = true } insta = { version = "1.41", features = ["yaml"], optional = true } toml = "0.8" diff --git a/benchmarks/progenitor-builds/README.md b/benchmarks/progenitor-builds/README.md index 95e66a5..664bdcf 100644 --- a/benchmarks/progenitor-builds/README.md +++ b/benchmarks/progenitor-builds/README.md @@ -34,7 +34,7 @@ or newer. ## Interpretation limits - Clean-build measurements include each generated client's current dependency - graph (`reqwest` 0.12 for `openapi-to-rust`, 0.13 for Progenitor). They do not + graph (`reqwest` 0.13 for both generators). They do not isolate procedural-macro overhead. - Filesystem and Cargo registry caches are warm, but every clean sample uses fresh compiled-artifact directories. diff --git a/benchmarks/progenitor-builds/run.mjs b/benchmarks/progenitor-builds/run.mjs index 6f5516c..7802696 100644 --- a/benchmarks/progenitor-builds/run.mjs +++ b/benchmarks/progenitor-builds/run.mjs @@ -119,11 +119,11 @@ version = "0.1.0" edition = "2024" [dependencies] -reqwest = { version = "=0.12.28", default-features = false, features = ["rustls-tls"] } -reqwest-middleware = "=0.4.2" +reqwest = { version = "=0.13.4", default-features = false, features = ["rustls"] } +reqwest-middleware = "=0.5.2" serde = { version = "=1.0.229", features = ["derive"] } serde_json = "=1.0.151" -thiserror = "=1.0.69" +thiserror = "=2.0.19" `); fs.writeFileSync(path.join(aotRoot, "src/lib.rs"), `pub mod generated; diff --git a/examples/generate_full_client.rs b/examples/generate_full_client.rs index e754b8f..fdc21fa 100644 --- a/examples/generate_full_client.rs +++ b/examples/generate_full_client.rs @@ -346,13 +346,13 @@ async fn main() -> Result<(), Box> {{ println!(" 1. Review the generated code in examples/generated/"); println!(" 2. Copy the generated files to your project"); println!(" 3. Add the required dependencies to your Cargo.toml:"); - println!(" - reqwest = {{ version = \"0.12\", features = [\"json\"] }}"); + println!(" - reqwest = {{ version = \"0.13\", features = [\"json\"] }}"); println!(" - reqwest-middleware = \"0.3\""); println!(" - reqwest-retry = \"0.6\""); - println!(" - reqwest-tracing = \"0.5\""); + println!(" - reqwest-tracing = \"0.7\""); println!(" - serde = {{ version = \"1.0\", features = [\"derive\"] }}"); println!(" - serde_json = \"1.0\""); - println!(" - thiserror = \"1.0\""); + println!(" - thiserror = \"2.0\""); println!(" - tokio = {{ version = \"1\", features = [\"full\"] }}"); Ok(()) diff --git a/examples/server-openai-responses/Cargo.toml b/examples/server-openai-responses/Cargo.toml index 73baf52..5874ab6 100644 --- a/examples/server-openai-responses/Cargo.toml +++ b/examples/server-openai-responses/Cargo.toml @@ -12,9 +12,9 @@ axum = "0.8" http-body-util = "0.1" jsonschema = { version = "0.49", default-features = false } mime = "0.3" -reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] } -reqwest-middleware = "0.4" -reqwest-tracing = "0.5" +reqwest = { version = "0.13", default-features = false, features = ["rustls"] } +reqwest-middleware = { version = "0.5", features = ["query"] } +reqwest-tracing = "0.7" serde = { version = "1", features = ["derive"] } serde_json = "1" serde_urlencoded = "0.7" diff --git a/scripts/install-smoke.sh b/scripts/install-smoke.sh index 16d28af..27a5982 100755 --- a/scripts/install-smoke.sh +++ b/scripts/install-smoke.sh @@ -49,13 +49,13 @@ dependency_tree="$(cargo tree \ --locked \ --edges normal)" if printf '%s\n' "$dependency_tree" \ - | grep -E 'reqwest v0\.11|hyper v0\.14|insta v|tempfile v' >/dev/null; then + | grep -E 'reqwest v0\.(11|12)|hyper v0\.14|insta v|tempfile v' >/dev/null; then echo "[install-smoke] default dependency tree contains a forbidden dependency" >&2 printf '%s\n' "$dependency_tree" >&2 exit 1 fi -if ! printf '%s\n' "$dependency_tree" | grep -E 'reqwest v0\.12' >/dev/null; then - echo "[install-smoke] expected reqwest 0.12 in the default dependency tree" >&2 +if ! printf '%s\n' "$dependency_tree" | grep -E 'reqwest v0\.13' >/dev/null; then + echo "[install-smoke] expected reqwest 0.13 in the default dependency tree" >&2 exit 1 fi diff --git a/src/generator.rs b/src/generator.rs index f6da442..4bc1655 100644 --- a/src/generator.rs +++ b/src/generator.rs @@ -424,6 +424,12 @@ impl CodeGenerator { .to_string(), )); } + if streaming_config.event_parser_helpers { + files.push(GeneratedFile { + path: "sse.rs".into(), + content: self.generate_sse_runtime()?, + }); + } let streaming_content = self.generate_streaming_client(streaming_config, analysis)?; files.push(GeneratedFile { @@ -762,6 +768,10 @@ impl CodeGenerator { ) -> Result { let mut client_code = TokenStream::new(); let provenance_attribute = self.provenance_attribute(); + let duration_import = streaming_config + .reconnection_config + .as_ref() + .map(|_| quote! { use std::time::Duration; }); // Generate imports let imports = quote! { @@ -776,18 +786,26 @@ impl CodeGenerator { use super::types::*; use async_trait::async_trait; - use futures_util::{Stream, StreamExt}; + use futures_util::Stream; use std::pin::Pin; - use std::time::Duration; use reqwest::header::{HeaderMap, HeaderValue}; - use tracing::{debug, error, info, warn, instrument}; + use tracing::{debug, info, instrument}; + #duration_import }; client_code.extend(imports); - // Generate error types if streaming_config.generate_client { - let error_types = self.generate_streaming_error_types()?; - client_code.extend(error_types); + if streaming_config.reconnection_config.is_some() { + client_code.extend(quote! { + use super::sse::{SseClient, SseReconnectOptions}; + pub use super::sse::StreamingError; + }); + } else { + client_code.extend(quote! { + use super::sse::SseClient; + pub use super::sse::StreamingError; + }); + } } // Generate client trait for each endpoint @@ -802,12 +820,6 @@ impl CodeGenerator { client_code.extend(client_impl); } - // Generate SSE parsing utilities - if streaming_config.event_parser_helpers { - let parser_code = self.generate_sse_parser_utilities(streaming_config)?; - client_code.extend(parser_code); - } - // Generate reconnection utilities if configured if let Some(reconnect_config) = &streaming_config.reconnection_config { let reconnect_code = self.generate_reconnection_utilities(reconnect_config)?; @@ -1285,6 +1297,9 @@ impl CodeGenerator { .collect::>(); let pub_uses = module_names .iter() + // The SSE runtime intentionally keeps transport-level names under + // `sse::` so they cannot collide with API-specific streaming types. + .filter(|name| name.as_str() != "sse") .map(|name| format!("pub use {name}::*;")) .collect::>(); @@ -3682,9 +3697,8 @@ impl CodeGenerator { let mut struct_fields = vec![ quote! { base_url: String }, quote! { api_key: Option }, - quote! { http_client: reqwest::Client }, + quote! { sse_client: SseClient }, quote! { custom_headers: std::collections::BTreeMap }, - quote! { max_response_body_bytes: usize }, ]; let has_optional_headers = !streaming_config @@ -3714,24 +3728,44 @@ impl CodeGenerator { .as_ref() .and_then(|http| http.max_response_body_bytes) .unwrap_or(8 * 1024 * 1024); + let sse_client_initializer = if let Some(reconnect) = &streaming_config.reconnection_config + { + let max_retries = reconnect.max_retries; + let initial_delay_ms = reconnect.initial_delay_ms; + let max_delay_ms = reconnect.max_delay_ms; + let backoff_multiplier = reconnect.backoff_multiplier; + quote! { + SseClient::new() + .with_max_error_body_bytes(#max_response_body_bytes) + .with_reconnect_options(SseReconnectOptions { + max_retries: #max_retries, + initial_retry_delay: std::time::Duration::from_millis(#initial_delay_ms), + max_retry_delay: std::time::Duration::from_millis(#max_delay_ms), + backoff_multiplier: #backoff_multiplier, + }) + } + } else { + quote! { + SseClient::new() + .with_max_error_body_bytes(#max_response_body_bytes) + } + }; // Build constructor fields based on what the struct has let constructor_fields = if has_optional_headers { quote! { base_url: #default_base_url.to_string(), api_key: None, - http_client: reqwest::Client::new(), + sse_client: #sse_client_initializer, custom_headers: std::collections::BTreeMap::new(), - max_response_body_bytes: #max_response_body_bytes, optional_headers: std::collections::BTreeMap::new(), } } else { quote! { base_url: #default_base_url.to_string(), api_key: None, - http_client: reqwest::Client::new(), + sse_client: #sse_client_initializer, custom_headers: std::collections::BTreeMap::new(), - max_response_body_bytes: #max_response_body_bytes, } }; @@ -3770,7 +3804,7 @@ impl CodeGenerator { /// Set the maximum number of error-response bytes buffered in memory. pub fn with_max_response_body_bytes(mut self, limit: usize) -> Self { - self.max_response_body_bytes = limit; + self.sse_client = self.sse_client.with_max_error_body_bytes(limit); self } @@ -3786,7 +3820,7 @@ impl CodeGenerator { /// Set the HTTP client pub fn with_http_client(mut self, client: reqwest::Client) -> Self { - self.http_client = client; + self.sse_client = self.sse_client.with_http_client(client); self } @@ -3987,17 +4021,17 @@ impl CodeGenerator { let url_str = url.to_string(); debug!("Making streaming GET request to: {}", url_str); - let request_builder = self.http_client + let request_builder = self.sse_client .get(url_str) .headers(headers); debug!("Creating SSE stream from request"); - let stream = parse_sse_stream_with_limit::<#event_type>( - request_builder, - self.max_response_body_bytes, - ).await?; + let stream = self + .sse_client + .stream::<#event_type>(request_builder) + .await?; info!("SSE stream created successfully"); - Ok(Box::pin(stream)) + Ok(stream) } } }) @@ -4079,28 +4113,167 @@ impl CodeGenerator { #url_construction debug!("Making streaming POST request to: {}", url); - let request_builder = self.http_client + let request_builder = self.sse_client .post(&url) .headers(headers) .json(&streaming_request); debug!("Creating SSE stream from request"); - let stream = parse_sse_stream_with_limit::<#event_type>( - request_builder, - self.max_response_body_bytes, - ).await?; + let stream = self + .sse_client + .stream::<#event_type>(request_builder) + .await?; info!("SSE stream created successfully"); - Ok(Box::pin(stream)) + Ok(stream) } } }) } - /// Generate SSE parsing utilities using reqwest-eventsource - fn generate_sse_parser_utilities( - &self, - _streaming_config: &crate::streaming::StreamingConfig, - ) -> Result { + /// Generate the reusable SSE transport module emitted as `sse.rs`. + fn generate_sse_runtime(&self) -> Result { + let provenance_attribute = self.provenance_attribute(); + let error_types = self.generate_streaming_error_types()?; + let parser = self.generate_sse_parser_utilities()?; + let tokens = quote! { + //! Generated SSE transport, framing, and JSON decoding support. + //! + //! This module is emitted only when SSE generation is enabled. + #provenance_attribute + #![allow(clippy::format_in_format_args)] + + use futures_util::{Stream, StreamExt}; + use std::pin::Pin; + use std::time::Duration; + use tracing::debug; + + #error_types + + /// Reusable transport client for generated SSE operations. + #[derive(Debug, Clone)] + pub struct SseClient { + http_client: reqwest::Client, + max_error_body_bytes: usize, + reconnect_options: Option, + } + + impl SseClient { + pub fn new() -> Self { + Self { + http_client: reqwest::Client::new(), + max_error_body_bytes: DEFAULT_MAX_SSE_ERROR_BODY_BYTES, + reconnect_options: None, + } + } + + pub fn with_http_client(mut self, client: reqwest::Client) -> Self { + self.http_client = client; + self + } + + pub fn with_max_error_body_bytes(mut self, limit: usize) -> Self { + self.max_error_body_bytes = limit; + self + } + + /// Enable automatic reconnection for [`Self::stream`]. + pub fn with_reconnect_options(mut self, options: SseReconnectOptions) -> Self { + self.reconnect_options = Some(options); + self + } + + pub fn get(&self, url: impl reqwest::IntoUrl) -> reqwest::RequestBuilder { + self.http_client.get(url) + } + + pub fn post(&self, url: impl reqwest::IntoUrl) -> reqwest::RequestBuilder { + self.http_client.post(url) + } + + pub async fn stream( + &self, + request_builder: reqwest::RequestBuilder, + ) -> Result> + Send>>, StreamingError> + where + T: serde::de::DeserializeOwned + Send + 'static, + { + if let Some(options) = self.reconnect_options.clone() { + parse_sse_json_reconnecting_with_limit( + request_builder, + self.max_error_body_bytes, + options, + ).await + } else { + parse_sse_json_stream_with_limit( + request_builder, + self.max_error_body_bytes, + ).await + } + } + + /// Stream raw SSE events from one HTTP connection. + pub async fn stream_raw( + &self, + request_builder: reqwest::RequestBuilder, + ) -> Result, StreamingError>> + Send>>, StreamingError> { + parse_sse_raw_stream_with_limit(request_builder, self.max_error_body_bytes).await + } + + /// Stream typed JSON SSE events from one HTTP connection. + pub async fn stream_json( + &self, + request_builder: reqwest::RequestBuilder, + ) -> Result, StreamingError>> + Send>>, StreamingError> + where + T: serde::de::DeserializeOwned + Send + 'static, + { + parse_sse_json_events_with_limit(request_builder, self.max_error_body_bytes).await + } + + /// Stream raw SSE events and reconnect retryable connections. + pub async fn stream_raw_reconnecting( + &self, + request_builder: reqwest::RequestBuilder, + ) -> Result, StreamingError>> + Send>>, StreamingError> { + parse_sse_raw_reconnecting_with_limit( + request_builder, + self.max_error_body_bytes, + self.reconnect_options.clone().unwrap_or_default(), + ).await + } + + /// Stream typed JSON SSE events and reconnect retryable connections. + pub async fn stream_json_reconnecting( + &self, + request_builder: reqwest::RequestBuilder, + ) -> Result, StreamingError>> + Send>>, StreamingError> + where + T: serde::de::DeserializeOwned + Send + 'static, + { + parse_sse_json_reconnecting_events_with_limit( + request_builder, + self.max_error_body_bytes, + self.reconnect_options.clone().unwrap_or_default(), + ).await + } + } + + impl Default for SseClient { + fn default() -> Self { + Self::new() + } + } + + #parser + }; + let syntax_tree = syn::parse2::(tokens).map_err(|error| { + GeneratorError::CodeGenError(format!("Failed to parse generated sse.rs: {error}")) + })?; + Ok(prettyplease::unparse(&syntax_tree)) + } + + /// Generate the standalone SSE framing and JSON parsing utilities. + fn generate_sse_parser_utilities(&self) -> Result { Ok(quote! { /// Default upper bound for an SSE error response buffered in memory. pub const DEFAULT_MAX_SSE_ERROR_BODY_BYTES: usize = 8 * 1024 * 1024; @@ -4120,124 +4293,562 @@ impl CodeGenerator { Ok(body) } - /// Parse SSE stream from HTTP request using reqwest-eventsource + /// A decoded SSE event with its transport metadata preserved. + #[derive(Debug, Clone, PartialEq, Eq)] + pub struct SseEvent { + /// Event name, or `message` when the server omitted `event:`. + pub event: String, + /// Raw or deserialized event payload. + pub data: T, + /// Most recent event ID, used to resume a reconnected stream. + pub id: Option, + /// Server-supplied reconnection delay on this event, if present. + pub retry: Option, + } + + /// Controls automatic SSE reconnection behavior. + #[derive(Debug, Clone)] + pub struct SseReconnectOptions { + /// Maximum consecutive reconnection attempts. + pub max_retries: u32, + /// Delay before the first reconnection when the server did not send `retry:`. + pub initial_retry_delay: Duration, + /// Upper bound for client-computed and server-supplied delays. + pub max_retry_delay: Duration, + /// Exponential backoff multiplier for consecutive failures. + pub backoff_multiplier: f64, + } + + impl Default for SseReconnectOptions { + fn default() -> Self { + Self { + max_retries: 3, + initial_retry_delay: Duration::from_secs(3), + max_retry_delay: Duration::from_secs(30), + backoff_multiplier: 2.0, + } + } + } + + impl SseReconnectOptions { + fn delay(&self, attempt: u32, server_retry: Option) -> Duration { + if let Some(delay) = server_retry { + return delay.min(self.max_retry_delay); + } + let multiplier = self.backoff_multiplier.max(1.0); + let millis = self.initial_retry_delay.as_millis() as f64 + * multiplier.powi(attempt.min(63) as i32); + Duration::from_millis( + millis.min(self.max_retry_delay.as_millis() as f64) as u64, + ) + } + } + + #[derive(Default)] + struct __SseDecoder { + line: Vec, + event: String, + data: Vec, + last_event_id: Option, + retry_delay: Option, + event_retry: Option, + saw_carriage_return: bool, + } + + impl __SseDecoder { + fn feed( + &mut self, + chunk: &[u8], + ) -> Vec, StreamingError>> { + let mut messages = Vec::new(); + for &byte in chunk { + if self.saw_carriage_return { + self.saw_carriage_return = false; + if byte == b'\n' { + continue; + } + } + + match byte { + b'\n' => self.finish_line(&mut messages), + b'\r' => { + self.finish_line(&mut messages); + self.saw_carriage_return = true; + } + _ => self.line.push(byte), + } + } + messages + } + + fn finish(&mut self) -> Vec, StreamingError>> { + let mut messages = Vec::new(); + if !self.line.is_empty() { + self.finish_line(&mut messages); + } + self.dispatch(&mut messages); + messages + } + + fn finish_line( + &mut self, + messages: &mut Vec, StreamingError>>, + ) { + let line = std::mem::take(&mut self.line); + let line = match String::from_utf8(line) { + Ok(line) => line, + Err(error) => { + messages.push(Err(StreamingError::Parsing(format!( + "SSE line is not valid UTF-8: {}", + error + )))); + return; + } + }; + + if line.is_empty() { + self.dispatch(messages); + return; + } + if line.starts_with(':') { + return; + } + + let (field, value) = line + .split_once(':') + .map_or((line.as_str(), ""), |(field, value)| { + (field, value.strip_prefix(' ').unwrap_or(value)) + }); + match field { + "event" => self.event = value.to_string(), + "data" => self.data.push(value.to_string()), + "id" if !value.contains('\0') => { + self.last_event_id = (!value.is_empty()).then(|| value.to_string()); + } + "retry" if value.bytes().all(|byte| byte.is_ascii_digit()) => { + if let Ok(milliseconds) = value.parse::() { + let delay = Duration::from_millis(milliseconds); + self.retry_delay = Some(delay); + self.event_retry = Some(delay); + } + } + _ => {} + } + } + + fn dispatch( + &mut self, + messages: &mut Vec, StreamingError>>, + ) { + if self.data.is_empty() { + self.event.clear(); + self.event_retry = None; + return; + } + messages.push(Ok(SseEvent { + event: if self.event.is_empty() { + "message".to_string() + } else { + std::mem::take(&mut self.event) + }, + data: std::mem::take(&mut self.data).join("\n"), + id: self.last_event_id.clone(), + retry: self.event_retry.take(), + })); + self.event.clear(); + } + + fn reset_for_reconnect(&mut self) { + self.line.clear(); + self.event.clear(); + self.data.clear(); + self.event_retry = None; + self.saw_carriage_return = false; + } + } + + fn __deserialize_sse_event( + event: SseEvent, + ) -> Option, StreamingError>> + where + T: serde::de::DeserializeOwned, + { + if event.data.trim() == "[DONE]" { + return None; + } + if event.event == "ping" { + debug!("Received SSE ping event, skipping"); + return None; + } + if event.data.trim().is_empty() { + debug!("Empty SSE data, skipping"); + return None; + } + + let json_value = match serde_json::from_str::(&event.data) { + Ok(value) => value, + Err(error) => { + return Some(Err(StreamingError::Parsing(format!( + "SSE event is not valid JSON: {} ({})", + event.data, error + )))); + } + }; + let is_ping = json_value + .get("event") + .or_else(|| json_value.get("type")) + .and_then(serde_json::Value::as_str) + .is_some_and(|event| event == "ping"); + if is_ping { + debug!("Received ping event in JSON data, skipping"); + return None; + } + + Some( + serde_json::from_value::(json_value) + .map(|data| SseEvent { + event: event.event.clone(), + data, + id: event.id.clone(), + retry: event.retry, + }) + .map_err(|error| StreamingError::Parsing(format!( + "Failed to parse SSE event: {} (raw: {}, event: {})", + error, event.data, event.event + ))), + ) + } + + /// Parse an SSE response without an external EventSource wrapper. pub async fn parse_sse_stream( request_builder: reqwest::RequestBuilder - ) -> Result>, StreamingError> + ) -> Result> + Send>>, StreamingError> where T: serde::de::DeserializeOwned + Send + 'static, { - parse_sse_stream_with_limit(request_builder, DEFAULT_MAX_SSE_ERROR_BODY_BYTES).await + parse_sse_json_stream_with_limit( + request_builder, + DEFAULT_MAX_SSE_ERROR_BODY_BYTES, + ).await } - async fn parse_sse_stream_with_limit( + struct __SseOpenError { + error: StreamingError, + retryable: bool, + } + + async fn __open_sse_response( request_builder: reqwest::RequestBuilder, max_response_body_bytes: usize, - ) -> Result>, StreamingError> - where - T: serde::de::DeserializeOwned + Send + 'static, - { - let mut event_source = reqwest_eventsource::EventSource::new(request_builder).map_err(|e| { - StreamingError::Connection(format!("Failed to create event source: {}", e)) + ) -> Result { + let response = request_builder.send().await.map_err(|error| __SseOpenError { + error: error.into(), + retryable: true, })?; + if !response.status().is_success() { + let status = response.status(); + let retryable = status.as_u16() == 429 || status.is_server_error(); + let error = match __read_bounded_streaming_error_body( + response, + max_response_body_bytes, + ).await { + Ok(body) => StreamingError::Connection(format!( + "HTTP {} error: {}", + status.as_u16(), + String::from_utf8_lossy(&body) + )), + Err(error) => error, + }; + return Err(__SseOpenError { error, retryable }); + } - let stream = event_source.filter_map(move |event_result| async move { - match event_result { - Ok(reqwest_eventsource::Event::Open) => { - debug!("SSE connection opened"); - None - } - Ok(reqwest_eventsource::Event::Message(message)) => { - // Check if this is a ping event by SSE event type - if message.event == "ping" { - debug!("Received SSE ping event, skipping"); + let content_type = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .unwrap_or_default(); + if !content_type + .split(';') + .next() + .is_some_and(|value| value.trim().eq_ignore_ascii_case("text/event-stream")) + { + let error = StreamingError::Parsing(format!( + "Expected text/event-stream response, received {}", + if content_type.is_empty() { "no Content-Type" } else { content_type } + )); + return Err(__SseOpenError { error, retryable: false }); + } + + debug!("SSE connection opened"); + Ok(response) + } + + fn __raw_response_stream( + response: reqwest::Response, + ) -> Pin, StreamingError>> + Send>> { + let stream = futures_util::stream::unfold( + ( + response, + __SseDecoder::default(), + std::collections::VecDeque::, StreamingError>>::new(), + false, + ), + |(mut response, mut decoder, mut pending, mut done)| async move { + loop { + if let Some(item) = pending.pop_front() { + return Some((item, (response, decoder, pending, done))); + } + if done { + debug!("SSE stream completed normally"); return None; } - // Special handling for empty data - if message.data.trim().is_empty() { - debug!("Empty SSE data, skipping"); + match response.chunk().await { + Ok(Some(chunk)) => { + for event in decoder.feed(&chunk) { + let is_done = event + .as_ref() + .is_ok_and(|event| event.data.trim() == "[DONE]"); + pending.push_back(event); + if is_done { + done = true; + break; + } + } + } + Err(error) => { + done = true; + pending.push_back(Err(error.into())); + } + Ok(None) => { + done = true; + for event in decoder.finish() { + pending.push_back(event); + } + } + } + } + } + ); + + Box::pin(stream) + } + + async fn parse_sse_raw_stream_with_limit( + request_builder: reqwest::RequestBuilder, + max_response_body_bytes: usize, + ) -> Result, StreamingError>> + Send>>, StreamingError> { + Ok(match __open_sse_response(request_builder, max_response_body_bytes).await { + Ok(response) => __raw_response_stream(response), + Err(error) => Box::pin(futures_util::stream::once(async move { Err(error.error) })), + }) + } + + fn __json_event_stream( + raw: Pin, StreamingError>> + Send>>, + ) -> Pin, StreamingError>> + Send>> + where + T: serde::de::DeserializeOwned + Send + 'static, + { + Box::pin(raw.filter_map(|event| async move { + match event { + Ok(event) => __deserialize_sse_event(event), + Err(error) => Some(Err(error)), + } + })) + } + + async fn parse_sse_json_events_with_limit( + request_builder: reqwest::RequestBuilder, + max_response_body_bytes: usize, + ) -> Result, StreamingError>> + Send>>, StreamingError> + where + T: serde::de::DeserializeOwned + Send + 'static, + { + Ok(__json_event_stream( + parse_sse_raw_stream_with_limit(request_builder, max_response_body_bytes).await?, + )) + } + + async fn parse_sse_json_stream_with_limit( + request_builder: reqwest::RequestBuilder, + max_response_body_bytes: usize, + ) -> Result> + Send>>, StreamingError> + where + T: serde::de::DeserializeOwned + Send + 'static, + { + let events = parse_sse_json_events_with_limit(request_builder, max_response_body_bytes).await?; + Ok(Box::pin(events.map(|event| event.map(|event| event.data)))) + } + + struct __ReconnectState { + request: reqwest::RequestBuilder, + response: Option, + decoder: __SseDecoder, + pending: std::collections::VecDeque, StreamingError>>, + options: SseReconnectOptions, + max_response_body_bytes: usize, + attempts: u32, + wait_before_open: bool, + done: bool, + } + + async fn parse_sse_raw_reconnecting_with_limit( + request_builder: reqwest::RequestBuilder, + max_response_body_bytes: usize, + options: SseReconnectOptions, + ) -> Result, StreamingError>> + Send>>, StreamingError> { + if request_builder.try_clone().is_none() { + return Err(StreamingError::Connection( + "SSE reconnection requires a cloneable request body".to_string(), + )); + } + + let stream = futures_util::stream::unfold( + __ReconnectState { + request: request_builder, + response: None, + decoder: __SseDecoder::default(), + pending: std::collections::VecDeque::new(), + options, + max_response_body_bytes, + attempts: 0, + wait_before_open: false, + done: false, + }, + |mut state| async move { + loop { + if let Some(item) = state.pending.pop_front() { + return Some((item, state)); + } + if state.done { return None; } - // Check if this is a ping event in the JSON data - if let Ok(json_value) = serde_json::from_str::(&message.data) { - if let Some(event_type) = json_value.get("event").and_then(|v| v.as_str()) { - if event_type == "ping" { - debug!("Received ping event in JSON data, skipping"); - return None; - } + if state.response.is_none() { + if state.wait_before_open { + let delay = state.options.delay( + state.attempts.saturating_sub(1), + state.decoder.retry_delay, + ); + debug!(?delay, attempt = state.attempts, "Reconnecting SSE stream"); + futures_timer::Delay::new(delay).await; + state.wait_before_open = false; } - // Try to parse the full event normally - match serde_json::from_value::(json_value) { - Ok(parsed_event) => { - Some(Ok(parsed_event)) + let mut request = state.request.try_clone().expect("request clone checked"); + if let Some(last_event_id) = state.decoder.last_event_id.as_deref() { + request = request.header("Last-Event-ID", last_event_id); + } + match __open_sse_response(request, state.max_response_body_bytes).await { + Ok(response) => state.response = Some(response), + Err(error) if error.retryable && state.attempts < state.options.max_retries => { + state.attempts += 1; + state.wait_before_open = true; + continue; } - Err(e) => { - if message.data.contains("ping") || message.event.contains("ping") { - debug!("Ignoring ping-related event: {}", message.data); - None - } else { - Some(Err(StreamingError::Parsing( - format!("Failed to parse SSE event: {} (raw: {})", e, message.data) - ))) - } + Err(error) => { + state.done = true; + state.pending.push_back(Err(error.error)); + continue; } } - } else { - // Not valid JSON at all - Some(Err(StreamingError::Parsing( - format!("SSE event is not valid JSON: {}", message.data) - ))) } - } - Err(e) => { - // Check if this is a normal stream end vs actual error - match e { - reqwest_eventsource::Error::StreamEnded => { - debug!("SSE stream completed normally"); - None // Normal stream end, not an error - } - reqwest_eventsource::Error::InvalidStatusCode(status, response) => { - // We have access to the response body for error details - let status_code = status.as_u16(); - - // Read the response body to get error details - let error_body = match __read_bounded_streaming_error_body( - response, - max_response_body_bytes, - ).await { - Ok(body) => String::from_utf8_lossy(&body).into_owned(), - Err(error) => return Some(Err(error)), - }; - - error!("SSE connection error - HTTP {}: {}", status_code, error_body); - - let detailed_error = format!( - "HTTP {} error: {}", - status_code, - error_body - ); - Some(Err(StreamingError::Connection(detailed_error))) + let next = state.response.as_mut().expect("response opened").chunk().await; + match next { + Ok(Some(chunk)) => { + let events = state.decoder.feed(&chunk); + if !events.is_empty() { + state.attempts = 0; + } + for event in events { + let is_done = event + .as_ref() + .is_ok_and(|event| event.data.trim() == "[DONE]"); + state.pending.push_back(event); + if is_done { + state.done = true; + state.response = None; + break; + } + } + } + Ok(None) => { + let events = state.decoder.finish(); + if !events.is_empty() { + state.attempts = 0; + } + for event in events { + let is_done = event + .as_ref() + .is_ok_and(|event| event.data.trim() == "[DONE]"); + state.pending.push_back(event); + if is_done { + state.done = true; + break; + } + } + state.response = None; + state.decoder.reset_for_reconnect(); + if !state.done { + if state.attempts < state.options.max_retries { + state.attempts += 1; + state.wait_before_open = true; + } else { + state.done = true; + } + } } - _ => { - let error_str = e.to_string(); - if error_str.contains("stream closed") { - debug!("SSE stream closed"); - None + Err(error) => { + state.response = None; + state.decoder.reset_for_reconnect(); + if state.attempts < state.options.max_retries { + state.attempts += 1; + state.wait_before_open = true; } else { - error!("SSE connection error: {}", e); - Some(Err(StreamingError::Connection(error_str))) + state.done = true; + state.pending.push_back(Err(error.into())); } } } } - } - }); + }, + ); + Ok(Box::pin(stream)) + } + + async fn parse_sse_json_reconnecting_events_with_limit( + request_builder: reqwest::RequestBuilder, + max_response_body_bytes: usize, + options: SseReconnectOptions, + ) -> Result, StreamingError>> + Send>>, StreamingError> + where + T: serde::de::DeserializeOwned + Send + 'static, + { + Ok(__json_event_stream( + parse_sse_raw_reconnecting_with_limit( + request_builder, + max_response_body_bytes, + options, + ).await?, + )) + } - Ok(stream) + async fn parse_sse_json_reconnecting_with_limit( + request_builder: reqwest::RequestBuilder, + max_response_body_bytes: usize, + options: SseReconnectOptions, + ) -> Result> + Send>>, StreamingError> + where + T: serde::de::DeserializeOwned + Send + 'static, + { + let events = parse_sse_json_reconnecting_events_with_limit( + request_builder, + max_response_body_bytes, + options, + ).await?; + Ok(Box::pin(events.map(|event| event.map(|event| event.data)))) } }) } diff --git a/src/type_mapping.rs b/src/type_mapping.rs index a2226c5..fe7226e 100644 --- a/src/type_mapping.rs +++ b/src/type_mapping.rs @@ -310,10 +310,16 @@ pub fn collect_generated_dep_requirements<'a>( } if uses("reqwest::") { - let mut features = vec!["rustls-tls"]; + let mut features = vec!["rustls"]; if uses(".json(&") { features.push("json"); } + if uses(".query(&") { + features.push("query"); + } + if uses(".form(&") { + features.push("form"); + } // Multipart operations reference `reqwest::multipart::Form` directly. // reqwest is pinned with default-features off, so the feature has to be // requested explicitly or the generated client fails to compile with @@ -323,27 +329,36 @@ pub fn collect_generated_dep_requirements<'a>( if uses("reqwest::multipart") { features.push("multipart"); } - // `Response::bytes_stream()`, used by generated SSE operations, is - // gated behind reqwest's `stream` feature. - if uses(".bytes_stream()") { + // Incremental response streaming APIs used by generated SSE operations + // are gated behind reqwest's `stream` feature. + if uses(".bytes_stream()") || uses(".chunk().await") { features.push("stream"); } dependencies.push( - DepRequirement::new("reqwest", "0.12") + DepRequirement::new("reqwest", "0.13") .without_default_features() .with_features(&features), ); } if uses("reqwest_middleware::") { - let dependency = if uses(".multipart(form)") { - DepRequirement::new("reqwest-middleware", "0.4").with_features(&["multipart"]) - } else { - DepRequirement::new("reqwest-middleware", "0.4") - }; - dependencies.push(dependency); + let mut features = Vec::new(); + if uses(".json(&") { + features.push("json"); + } + if uses(".query(&") { + features.push("query"); + } + if uses(".form(&") { + features.push("form"); + } + if uses(".multipart(form)") { + features.push("multipart"); + } + dependencies + .push(DepRequirement::new("reqwest-middleware", "0.5").with_features(&features)); } if uses("reqwest_retry::") { - let dependency = DepRequirement::new("reqwest-retry", "0.7"); + let dependency = DepRequirement::new("reqwest-retry", "0.9"); dependencies.push(if uses("reqwest_tracing::") { dependency } else { @@ -351,13 +366,10 @@ pub fn collect_generated_dep_requirements<'a>( }); } if uses("reqwest_tracing::") { - dependencies.push(DepRequirement::new("reqwest-tracing", "0.5")); - } - if uses("reqwest_eventsource::") { - dependencies.push(DepRequirement::new("reqwest-eventsource", "0.6")); + dependencies.push(DepRequirement::new("reqwest-tracing", "0.7")); } if uses("thiserror::") || uses("use thiserror::") { - dependencies.push(DepRequirement::new("thiserror", "1")); + dependencies.push(DepRequirement::new("thiserror", "2")); } if uses("async_trait::") { dependencies.push(DepRequirement::new("async-trait", "0.1")); @@ -365,6 +377,9 @@ pub fn collect_generated_dep_requirements<'a>( if uses("futures_util::") { dependencies.push(DepRequirement::new("futures-util", "0.3")); } + if uses("futures_timer::") { + dependencies.push(DepRequirement::new("futures-timer", "3")); + } if uses("futures_core::") { dependencies.push(DepRequirement::new("futures-core", "0.3")); } diff --git a/tests/client_response_body_limit_test.rs b/tests/client_response_body_limit_test.rs index 439fb46..8fda654 100644 --- a/tests/client_response_body_limit_test.rs +++ b/tests/client_response_body_limit_test.rs @@ -138,8 +138,13 @@ fn generated_clients_bound_chunked_responses_without_content_length() { .find(|file| file.path == std::path::Path::new("streaming.rs")) .unwrap(); assert!(streaming.content.contains("with_max_response_body_bytes")); - assert!(streaming.content.contains("ResponseTooLarge")); - assert!(!streaming.content.contains("response.text().await")); + let sse = result + .files + .iter() + .find(|file| file.path == std::path::Path::new("sse.rs")) + .unwrap(); + assert!(sse.content.contains("ResponseTooLarge")); + assert!(!sse.content.contains("response.text().await")); std::fs::write( temp.path().join("src/lib.rs"), @@ -151,14 +156,16 @@ mod tests { use super::generated::streaming::{ StreamEventsStreamingClient, StreamingClient, StreamingError, }; + use super::generated::sse::{SseClient, SseReconnectOptions}; use futures_util::StreamExt; + use std::time::Duration; use tokio::io::{AsyncReadExt, AsyncWriteExt}; async fn spawn_chunked_server() -> std::net::SocketAddr { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); tokio::spawn(async move { - for _ in 0..8 { + for _ in 0..10 { let (mut socket, _) = listener.accept().await.unwrap(); let mut request = Vec::new(); loop { @@ -192,6 +199,17 @@ mod tests { "application/json", vec![b"12345", b"67890"], ), + "/sse-chunks" => ( + "200 OK", + "text/event-stream; charset=utf-8", + vec![ + b": keepalive\r", + b"\nevent: ping\r\ndata: {\"type\":\"ping\"}\r\n\r\n", + b"event: message\ndata: {\"message\":\"fir", + b"st\"}\n\ndata: [DONE]\n\n", + b"data: {\"message\":\"ignored\"}\n\n", + ], + ), "/oversized-text" => ("200 OK", "text/plain", vec![b"12345", b"67890"]), "/oversized-binary" => ( "200 OK", @@ -218,6 +236,46 @@ mod tests { address } + async fn spawn_reconnect_server() -> (std::net::SocketAddr, tokio::task::JoinHandle<()>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let task = tokio::spawn(async move { + for attempt in 0..2 { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + loop { + let mut buffer = [0_u8; 1024]; + let read = socket.read(&mut buffer).await.unwrap(); + request.extend_from_slice(&buffer[..read]); + if read == 0 || request.windows(4).any(|window| window == b"\r\n\r\n") { + break; + } + } + let request = String::from_utf8_lossy(&request).to_ascii_lowercase(); + if attempt == 0 { + assert!(!request.contains("last-event-id:")); + } else { + assert!(request.contains("last-event-id: 42"), "{request}"); + } + + let body = if attempt == 0 { + b"id: 42\nretry: 1\nevent: update\ndata: {\"sequence\":1}\n\n".as_slice() + } else { + b"id: 43\nevent: update\ndata: {\"sequence\":2}\n\ndata: [DONE]\n\n".as_slice() + }; + let headers = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n"; + socket.write_all(headers.as_bytes()).await.unwrap(); + socket + .write_all(format!("{:X}\r\n", body.len()).as_bytes()) + .await + .unwrap(); + socket.write_all(body).await.unwrap(); + socket.write_all(b"\r\n0\r\n\r\n").await.unwrap(); + } + }); + (address, task) + } + macro_rules! assert_too_large { ($future:expr) => { match $future.await.unwrap_err() { @@ -252,6 +310,48 @@ mod tests { StreamingError::ResponseTooLarge { limit } => assert_eq!(limit, 8), other => panic!("unexpected streaming error: {other:?}"), } + + let sse_client = SseClient::new(); + let request = sse_client.get(format!("http://{address}/sse-chunks")); + let mut stream = sse_client + .stream::(request) + .await + .unwrap(); + assert_eq!( + stream.next().await.unwrap().unwrap(), + serde_json::json!({ "message": "first" }) + ); + assert!(stream.next().await.is_none()); + + let request = sse_client.get(format!("http://{address}/sse-chunks")); + let mut raw = sse_client.stream_raw(request).await.unwrap(); + assert_eq!(raw.next().await.unwrap().unwrap().event, "ping"); + assert_eq!(raw.next().await.unwrap().unwrap().event, "message"); + assert_eq!(raw.next().await.unwrap().unwrap().data, "[DONE]"); + assert!(raw.next().await.is_none()); + + let (reconnect_address, reconnect_task) = spawn_reconnect_server().await; + let reconnecting = SseClient::new().with_reconnect_options(SseReconnectOptions { + max_retries: 1, + initial_retry_delay: Duration::ZERO, + max_retry_delay: Duration::from_millis(10), + backoff_multiplier: 1.0, + }); + let request = reconnecting.get(format!("http://{reconnect_address}/events")); + let mut events = reconnecting + .stream_json_reconnecting::(request) + .await + .unwrap(); + let first = events.next().await.unwrap().unwrap(); + assert_eq!(first.event, "update"); + assert_eq!(first.id.as_deref(), Some("42")); + assert_eq!(first.retry, Some(Duration::from_millis(1))); + assert_eq!(first.data, serde_json::json!({ "sequence": 1 })); + let second = events.next().await.unwrap().unwrap(); + assert_eq!(second.id.as_deref(), Some("43")); + assert_eq!(second.data, serde_json::json!({ "sequence": 2 })); + assert!(events.next().await.is_none()); + reconnect_task.await.unwrap(); } } "#, diff --git a/tests/generation_requirements_test.rs b/tests/generation_requirements_test.rs index 4f6c115..191ee23 100644 --- a/tests/generation_requirements_test.rs +++ b/tests/generation_requirements_test.rs @@ -1,5 +1,5 @@ use openapi_to_rust::config::{ServerSection, ServerValidationSection}; -use openapi_to_rust::streaming::{StreamingConfig, StreamingEndpoint}; +use openapi_to_rust::streaming::{ReconnectionConfig, StreamingConfig, StreamingEndpoint}; use openapi_to_rust::type_mapping::{BinaryStrategy, DurationStrategy, TypeMappingConfig}; use openapi_to_rust::{CodeGenerator, GeneratorConfig, RetryConfig, SchemaAnalyzer, TypeMapper}; use serde_json::json; @@ -141,6 +141,18 @@ fn assert_names( assert_eq!(dependency_names(result), expected.into_iter().collect()); } +fn assert_version(result: &openapi_to_rust::GenerationResult, crate_name: &str, expected: &str) { + let dependency = result + .required_deps + .iter() + .find(|dependency| dependency.crate_name == crate_name) + .unwrap_or_else(|| panic!("missing {crate_name} dependency")); + assert_eq!( + dependency.version, expected, + "unexpected {crate_name} version" + ); +} + fn compile_case(name: &str, mut config: GeneratorConfig) -> openapi_to_rust::GenerationResult { let manifest_dir = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")); let temp = tempfile::TempDir::new().expect("scratch crate"); @@ -221,6 +233,7 @@ fn streaming_config() -> StreamingConfig { event_union_type: "StreamEvent".into(), ..Default::default() }], + reconnection_config: Some(ReconnectionConfig::default()), ..Default::default() } } @@ -268,6 +281,12 @@ fn disabled_sse_feature_does_not_emit_streaming_code_or_dependencies() { .iter() .all(|file| file.path != std::path::Path::new("streaming.rs")) ); + assert!( + result + .files + .iter() + .all(|file| file.path != std::path::Path::new("sse.rs")) + ); assert!(!dependency_names(&result).contains("reqwest-eventsource")); assert!(!dependency_names(&result).contains("futures-util")); } @@ -391,7 +410,8 @@ fn every_generation_mode_compiles_from_its_exact_dependency_fragment() { .iter() .find(|dependency| dependency.crate_name == "reqwest-middleware") .expect("middleware dependency"); - assert_eq!(middleware.features, vec!["multipart"]); + assert_eq!(middleware.version, "0.5"); + assert_eq!(middleware.features, vec!["multipart", "query"]); let sse = compile_case( "sse", @@ -410,9 +430,9 @@ fn every_generation_mode_compiles_from_its_exact_dependency_fragment() { "base64", "bytes", "chrono", + "futures-timer", "futures-util", "reqwest", - "reqwest-eventsource", "serde", "serde_json", "thiserror", @@ -426,8 +446,37 @@ fn every_generation_mode_compiles_from_its_exact_dependency_fragment() { .iter() .find(|dependency| dependency.crate_name == "reqwest") .expect("reqwest dependency"); - assert_eq!(reqwest.features, vec!["json", "rustls-tls"]); + assert_eq!(reqwest.version, "0.13"); + assert_eq!(reqwest.features, vec!["json", "rustls", "stream"]); assert!(!reqwest.default_features); + assert_version(&sse, "thiserror", "2"); + let sse_runtime = sse + .files + .iter() + .find(|file| file.path == std::path::Path::new("sse.rs")) + .expect("SSE runtime module"); + assert!(sse_runtime.content.contains("pub struct SseClient")); + assert!(sse_runtime.content.contains("pub struct SseEvent")); + assert!( + sse_runtime + .content + .contains("pub struct SseReconnectOptions") + ); + assert!(sse_runtime.content.contains("pub async fn stream_raw")); + assert!(sse_runtime.content.contains("Last-Event-ID")); + let streaming = sse + .files + .iter() + .find(|file| file.path == std::path::Path::new("streaming.rs")) + .expect("API-specific streaming module"); + assert!( + streaming + .content + .contains("use super::sse::{SseClient, SseReconnectOptions}") + ); + assert!(streaming.content.contains("with_reconnect_options")); + assert!(sse.mod_file.content.contains("pub mod sse;")); + assert!(!sse.mod_file.content.contains("pub use sse::*;")); let server = compile_case( "server", @@ -499,12 +548,12 @@ fn every_generation_mode_compiles_from_its_exact_dependency_fragment() { "bytes", "chrono", "futures-core", + "futures-timer", "futures-util", "http-body-util", "jsonschema", "mime", "reqwest", - "reqwest-eventsource", "reqwest-middleware", "reqwest-retry", "reqwest-tracing", @@ -517,4 +566,9 @@ fn every_generation_mode_compiles_from_its_exact_dependency_fragment() { "uuid", ], ); + assert_version(&combined, "reqwest", "0.13"); + assert_version(&combined, "reqwest-middleware", "0.5"); + assert_version(&combined, "reqwest-retry", "0.9"); + assert_version(&combined, "reqwest-tracing", "0.7"); + assert_version(&combined, "thiserror", "2"); } diff --git a/tests/live_sse_backend_test.rs b/tests/live_sse_backend_test.rs new file mode 100644 index 0000000..0d1cc71 --- /dev/null +++ b/tests/live_sse_backend_test.rs @@ -0,0 +1,201 @@ +use openapi_to_rust::streaming::{HttpMethod, StreamingConfig, StreamingEndpoint}; +use openapi_to_rust::{CodeGenerator, GeneratorConfig, SchemaAnalyzer}; +use serde_json::json; +use std::process::Command; + +fn live_sse_spec() -> serde_json::Value { + json!({ + "openapi": "3.1.0", + "info": { "title": "live SSE transport", "version": "1.0.0" }, + "paths": { + "/events": { + "post": { + "operationId": "streamEvents", + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { "$ref": "#/components/schemas/StreamRequest" } + } + } + }, + "responses": { + "200": { + "description": "events", + "content": { + "text/event-stream": { + "schema": { "$ref": "#/components/schemas/StreamEvent" } + } + } + } + } + } + } + }, + "components": { + "schemas": { + "StreamRequest": { + "type": "object", + "additionalProperties": true + }, + "StreamEvent": { + "type": "object", + "additionalProperties": true + } + } + } + }) +} + +#[test] +#[ignore = "requires OPENAPI_TO_RUST_SSE_BASE_URL and a live OpenAI/Anthropic-compatible backend"] +fn generated_sse_transport_streams_openai_and_anthropic_protocols() { + let base_url = std::env::var("OPENAPI_TO_RUST_SSE_BASE_URL") + .expect("OPENAPI_TO_RUST_SSE_BASE_URL must point to the live backend"); + let model = std::env::var("OPENAPI_TO_RUST_SSE_MODEL") + .unwrap_or_else(|_| "qwen/qwen3.5-9b".to_string()); + + let temp = tempfile::TempDir::new().unwrap(); + let output_dir = temp.path().join("src/generated"); + let mut analysis = SchemaAnalyzer::new(live_sse_spec()) + .unwrap() + .analyze() + .unwrap(); + let generator = CodeGenerator::new(GeneratorConfig { + output_dir: output_dir.clone(), + enable_async_client: false, + enable_sse_client: true, + tracing_enabled: false, + streaming_config: Some(StreamingConfig { + endpoints: vec![StreamingEndpoint { + operation_id: "streamEvents".into(), + path: "/events".into(), + http_method: HttpMethod::Post, + event_union_type: "StreamEvent".into(), + ..Default::default() + }], + ..Default::default() + }), + ..Default::default() + }); + let result = generator.generate_all(&mut analysis).unwrap(); + generator.write_files(&result).unwrap(); + + assert!(output_dir.join("sse.rs").is_file()); + let streaming = std::fs::read_to_string(output_dir.join("streaming.rs")).unwrap(); + assert!(streaming.contains("use super::sse::SseClient")); + + std::fs::write( + temp.path().join("src/main.rs"), + r#"mod generated; + +use futures_util::StreamExt; +use generated::sse::SseClient; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let base_url = std::env::var("OPENAPI_TO_RUST_SSE_BASE_URL")?; + let model = std::env::var("OPENAPI_TO_RUST_SSE_MODEL")?; + let client = SseClient::new(); + + let openai_request = client + .post(format!("{base_url}/v1/chat/completions")) + .bearer_auth("local-test") + .json(&serde_json::json!({ + "model": model, + "messages": [{ "role": "user", "content": "Reply with exactly: SSE works" }], + "stream": true, + "max_tokens": 16 + })); + let mut openai = client.stream::(openai_request).await?; + let mut openai_chunks = 0_usize; + while let Some(event) = openai.next().await { + let event = event?; + if event.get("object").and_then(serde_json::Value::as_str) + == Some("chat.completion.chunk") + { + openai_chunks += 1; + } + } + if openai_chunks == 0 { + return Err("OpenAI stream produced no chat.completion.chunk events".into()); + } + + let anthropic_request = client + .post(format!("{base_url}/v1/messages")) + .header("x-api-key", "local-test") + .header("anthropic-version", "2023-06-01") + .json(&serde_json::json!({ + "model": model, + "messages": [{ "role": "user", "content": "Reply with exactly: SSE works" }], + "stream": true, + "max_tokens": 16 + })); + let mut anthropic = client + .stream_json::(anthropic_request) + .await?; + let mut event_types = std::collections::BTreeSet::new(); + while let Some(event) = anthropic.next().await { + let event = event?; + if let Some(event_type) = event + .data + .get("type") + .and_then(serde_json::Value::as_str) + { + if event.event != event_type { + return Err(format!( + "Anthropic SSE event name {:?} did not match typed payload {event_type:?}", + event.event + ) + .into()); + } + event_types.insert(event_type.to_string()); + } + } + for expected in ["message_start", "content_block_delta", "message_stop"] { + if !event_types.contains(expected) { + return Err(format!("Anthropic stream omitted {expected}; saw {event_types:?}").into()); + } + } + + println!("OpenAI chunks: {openai_chunks}; Anthropic event types: {event_types:?}"); + Ok(()) +} +"#, + ) + .unwrap(); + + let dependencies = std::fs::read_to_string(output_dir.join("REQUIRED_DEPS.toml")).unwrap(); + std::fs::write( + temp.path().join("Cargo.toml"), + format!( + r#"[package] +name = "live-generated-sse-client" +version = "0.0.0" +edition = "2024" +publish = false + +{dependencies} +tokio = {{ version = "1", features = ["macros", "rt-multi-thread"] }} +"# + ), + ) + .unwrap(); + + let output = Command::new("cargo") + .args(["run", "--quiet"]) + .current_dir(temp.path()) + .env("CARGO_BUILD_BUILD_DIR", temp.path().join("cargo-build")) + .env("CARGO_TARGET_DIR", temp.path().join("cargo-target")) + .env("OPENAPI_TO_RUST_SSE_BASE_URL", base_url) + .env("OPENAPI_TO_RUST_SSE_MODEL", model) + .output() + .unwrap(); + assert!( + output.status.success(), + "live generated SSE client failed:\nstdout: {}\nstderr: {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + eprintln!("{}", String::from_utf8_lossy(&output.stdout)); +} diff --git a/tests/multi_response_client_test.rs b/tests/multi_response_client_test.rs index 572b536..7499e00 100644 --- a/tests/multi_response_client_test.rs +++ b/tests/multi_response_client_test.rs @@ -22,9 +22,9 @@ edition = "2021" [dependencies] serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" -reqwest = { version = "0.12", features = ["json", "multipart"] } -reqwest-middleware = { version = "0.4", features = ["multipart"] } -thiserror = "1.0" +reqwest = { version = "0.13", features = ["json", "multipart"] } +reqwest-middleware = { version = "0.5", features = ["multipart", "query"] } +thiserror = "2.0" tokio = { version = "1.0", features = ["full"] } validator = { version = "0.20", features = ["derive"] } # Q2 typed-scalar deps (default-on; harmless when unused). diff --git a/tests/operation_builder_test.rs b/tests/operation_builder_test.rs index 7e73a67..5261035 100644 --- a/tests/operation_builder_test.rs +++ b/tests/operation_builder_test.rs @@ -324,8 +324,8 @@ edition = "2024" serde = { version = "1", features = ["derive"] } serde_json = "1" thiserror = "2" -reqwest = { version = "0.12", features = ["json", "multipart"] } -reqwest-middleware = { version = "0.4", features = ["multipart"] } +reqwest = { version = "0.13", features = ["json", "multipart"] } +reqwest-middleware = { version = "0.5", features = ["multipart", "query"] } "#, ) .unwrap(); diff --git a/tests/server_query_roundtrip_test.rs b/tests/server_query_roundtrip_test.rs index 95f06e7..feb95d4 100644 --- a/tests/server_query_roundtrip_test.rs +++ b/tests/server_query_roundtrip_test.rs @@ -316,8 +316,8 @@ axum = "0.8" jsonschema = { version = "0.49", default-features = false } mime = "0.3" http-body-util = "0.1" -reqwest = { version = "0.12", features = ["json", "multipart"] } -reqwest-middleware = { version = "0.4", features = ["multipart"] } +reqwest = { version = "0.13", features = ["json", "multipart"] } +reqwest-middleware = { version = "0.5", features = ["multipart", "query"] } serde = { version = "1", features = ["derive"] } serde_json = "1" serde_urlencoded = "0.7" diff --git a/tests/server_validation_roundtrip_test.rs b/tests/server_validation_roundtrip_test.rs index b41d497..a95de77 100644 --- a/tests/server_validation_roundtrip_test.rs +++ b/tests/server_validation_roundtrip_test.rs @@ -131,12 +131,12 @@ axum = { version = "0.8", default-features = false, features = ["http1", "json", http-body-util = "0.1" jsonschema = { version = "0.49", default-features = false } mime = "0.3" -reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] } -reqwest-middleware = "0.4" +reqwest = { version = "0.13", default-features = false, features = ["rustls"] } +reqwest-middleware = { version = "0.5", features = ["query"] } serde = { version = "1", features = ["derive"] } serde_json = "1" serde_urlencoded = "0.7" -thiserror = "1" +thiserror = "2" tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "sync", "time"] } url = "2" "#,