Skip to content

decode streamed bodies in the typescript client #293

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
eburgos merged 1 commit into from
Sep 6, 2026
Merged

decode streamed bodies in the typescript client #293

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2201,9 +2201,9 @@ impl<H: FaultHandler + Sync> http_rest_client::Transport for HttpLoop<'_, H> {

A client adapter is the mirror: a small hand-written `Transport` implementation over a real HTTP stack (reqwest in Rust; a service-agnostic `fetch` helper in TypeScript; a service-agnostic `send` implementation in Dart), living with whichever codebase calls the service, doing exactly one job -- carry the plain-terms request across a real connection and hand the plain-terms response back. Timeouts, cancellation, retries, connection handling, mutual TLS and authentication are that adapter's own concerns, never the generated seam's, exactly as `amqp_rpc`'s own adapter split already works: the generator emits the seam, the application implements it against its own libraries, and a workspace that wants to share one adapter across services keeps it in its own crate under its own name -- never a `tixschema`-branded runtime dependency.

**The TypeScript and Dart clients.** `<Service>Schema::ts_http_client()` publishes the `http_rest` half beside `ts_client()`'s AMQP-shaped one: a service-agnostic `{Service}HttpTransport` seam (`send(request): Promise<response>`, both the request and the response carrying `method`/`path`/`query`/`headers`/`body` as plain strings, plus `parts` where the service declares a multipart operation), the `{Service}HttpClient` interface, and `create{Service}HttpClient(transport)`. It needs the `zod` feature exactly as `ts_client()` and `ts_service()` do -- outbound validation before a byte goes out is what a `safeParse` against the message's own `$Schema` gives it, and a build without Zod cannot write that check truthfully, so it publishes none of the three rather than one without it. `<Service>Schema::dart_http_client()` is the Dart sibling -- the same seam and per-operation client, over the `dart` feature's own generated types and JSON codec rather than Zod, needing no separate outbound check because a Dart message is a real class with `required` constructor parameters and cannot be built malformed in the first place. Where TypeScript answers a reply with an `{ ok, value | error }` union, Dart throws: a reply method answers `Future<Success>` directly and throws `{Service}HttpError<Declared>` (the declared error, or a fault behind `isServiceFault`), and a one-way method answers `Future<void>` and throws the fault-only `{Service}HttpRefusal` -- Dart's own idiom for a `Future`, mirroring exactly how its own one-way AMQP methods already throw.
**The TypeScript and Dart clients.** `<Service>Schema::ts_http_client()` publishes the `http_rest` half beside `ts_client()`'s AMQP-shaped one: a service-agnostic `{Service}HttpTransport` seam (`send(request): Promise<response>`, both the request and the response carrying `method`/`path`/`query`/`headers`/`body` as plain strings, plus `parts` on the request where the service declares a multipart operation and `bodyStream: ReadableStream<Uint8Array>` on the response where the service declares a streamed operation -- the platform's own stream type, never a naming of `fetch`), the `{Service}HttpClient` interface, and `create{Service}HttpClient(transport)`. It needs the `zod` feature exactly as `ts_client()` and `ts_service()` do -- outbound validation before a byte goes out is what a `safeParse` against the message's own `$Schema` gives it, and a build without Zod cannot write that check truthfully, so it publishes none of the three rather than one without it. `<Service>Schema::dart_http_client()` is the Dart sibling -- the same seam and per-operation client, over the `dart` feature's own generated types and JSON codec rather than Zod, needing no separate outbound check because a Dart message is a real class with `required` constructor parameters and cannot be built malformed in the first place. Where TypeScript answers a reply with an `{ ok, value | error }` union, Dart throws: a reply method answers `Future<Success>` directly and throws `{Service}HttpError<Declared>` (the declared error, or a fault behind `isServiceFault`), and a one-way method answers `Future<void>` and throws the fault-only `{Service}HttpRefusal` -- Dart's own idiom for a `Future`, mirroring exactly how its own one-way AMQP methods already throw.

Both language backends cover `body = "json"` and `body = "bytes"` in full. Neither covers `body = "stream"`: only the Rust dispatcher and the Rust client understand `StreamedAnswer`, so a streamed operation's generated TypeScript or Dart client method is not part of what this version writes correctly -- keep a streamed operation's callers in Rust, or hand-write that one call against the plain-terms transport seam directly, until a later version teaches these two backends the same seam. `body = "multipart"` is covered on the TypeScript side (the client builds `parts` from the message's own fields and the declared `part` bindings) but not on the Dart side, which always JSON-encodes the whole message as the body regardless of the declared body kind -- a multipart operation reached through `dart_http_client()` is the same open gap as a streamed one.
Both language backends cover `body = "json"`, `body = "bytes"`, `body = "stream"` and `body = "multipart"` in full. A streamed operation's TypeScript client answers `{ contentRange: string | undefined; body: ReadableStream<Uint8Array> }` -- `contentRange` left `undefined` at the operation's own `ok_status`, read back off the response ahead of naming the body and set to the range text at `206` -- off the seam's own `bodyStream` field; its Dart client answers the same pairing as a `({String? contentRange, Stream<List<int>> body})` record off `bodyStream` there too. A declared `header_out` composes onto either answer exactly as it does for `json` and `bytes`. A multipart operation's TypeScript client builds `parts` from the message's own fields and the declared `part` bindings; its Dart client builds the same list, the file handles crossing as `dynamic` through the same path an unknown type already renders by.

## Field Validation (`model_schema_prop`)

Expand Down
206 changes: 156 additions & 50 deletions src/features/service_schema/http_client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,15 @@
//! fixed fault statuses (400 validation, 404 unmatched route, 500 panic) all become the same
//! [`crate::service_schema::support`]-published fault type every other surface answers faults
//! through.
//!
//! # `body = "stream"` reads two success statuses, not one
//!
//! `reply_decode_stmt` peels `BodyKind::Stream` off first into its own status ladder (`206` reads
//! `content-range` back before the body is ever named; the declared `ok_status` leaves it
//! `undefined`; either way the answer carries `response.bodyStream` — the seam's own
//! `ReadableStream<Uint8Array>`, present only where the service declares a streamed operation) —
//! mirroring the Rust and Dart clients' own split. What is left (`success_decode_block`) only ever
//! sees `Bytes`, `Json` and `Multipart`.

use super::fault;
use super::message;
Expand All @@ -45,15 +54,16 @@ use crate::rename_rule::RenameRule;
use crate::service_schema::parse::{
BodyKind, DEFAULT_BINDING_ERROR_STATUS, HttpShape, OperationDef, OperationInputs,
OperationOutcome, PathSegment, ScalarKind, ServiceDef, is_unit_type, option_inner, scalar_kind,
service_declares_multipart, tuple_elements, vec_inner, wire_key,
service_declares_a_stream, service_declares_multipart, tuple_elements, vec_inner, wire_key,
};
use core::fmt::Write as _;
use syn::Type;

pub fn emit(service: &ServiceDef) -> Vec<String> {
let has_stream = service_declares_a_stream(service);
let has_multipart = service_declares_multipart(service);
let mut published = vec![
transport_type(&service.ident.to_string(), has_multipart),
transport_type(&service.ident.to_string(), has_stream, has_multipart),
client_type(service),
];
published.extend(fault_helpers(service));
Expand All @@ -68,13 +78,22 @@ pub fn emit(service: &ServiceDef) -> Vec<String> {
/// The one seam type a `{service}` client sends through: a request record in, a response record
/// out, nothing service-specific in either and nothing here naming the library that finally
/// carries the call. The request record carries `parts` only where the service declares a
/// multipart operation - every other body kind still carries its content as `body`.
fn transport_type(service: &str, has_multipart: bool) -> String {
/// multipart operation - every other body kind still carries its content as `body`. The response
/// record carries `bodyStream` only where the service declares a streamed operation - a real
/// implementation can fill it a chunk at a time rather than buffering the whole answer first, while
/// `body` keeps answering eagerly for every other operation. `ReadableStream` is the platform's own
/// type, not a naming of `fetch`: the seam stays library-agnostic either way.
fn transport_type(service: &str, has_stream: bool, has_multipart: bool) -> String {
let parts_field = if has_multipart {
"\n parts: ReadonlyArray<readonly [string, unknown]>;"
} else {
""
};
let stream_field = if has_stream {
"\n bodyStream: ReadableStream<Uint8Array>;"
} else {
""
};
format!(
"/**\n \
* What binds a `{service}` client to a real HTTP stack.\n \
Expand All @@ -95,7 +114,7 @@ fn transport_type(service: &str, has_multipart: bool) -> String {
}}): Promise<{{\n \
status: number;\n \
headers: ReadonlyArray<readonly [string, string]>;\n \
body: string;\n \
body: string;{stream_field}\n \
}}>;\n\
}};"
)
Expand Down Expand Up @@ -535,6 +554,9 @@ fn reply_decode_stmt(
error: &Type,
success: &Type,
) -> String {
if matches!(shape.body_kind, BodyKind::Stream) {
return stream_reply_decode_stmt(prefix, shape, wire, error, success);
}
let ok_status = shape.ok_status;
let error_condition = error_condition_expr(shape);
let error_ty = get_field_def("error", error, "").typescript_typename();
Expand Down Expand Up @@ -584,7 +606,9 @@ fn reply_decode_stmt(
/// The statements that read a success answer off the response, once its status has already
/// matched: nothing at all for a unit reply, the body alone for an ordinary reply, the response
/// bytes and their content type for a `body = "bytes"` operation, or the body plus every
/// `header_out` element read back from the response's own headers.
/// `header_out` element read back from the response's own headers. `body = "stream"` never reaches
/// here — [`reply_decode_stmt`] answers it through [`stream_reply_decode_stmt`] instead, since a
/// streamed answer has two success statuses (`200` and `206`), not one.
fn success_decode_block(prefix: &str, wire: &str, shape: &HttpShape, success: &Type) -> String {
if matches!(shape.body_kind, BodyKind::Bytes) {
return bytes_success_decode_block(prefix, wire, shape, success);
Expand Down Expand Up @@ -629,41 +653,8 @@ fn success_decode_block(prefix: &str, wire: &str, shape: &HttpShape, success: &T
}};\n \
}}\n"
);
let mut header_idents = Vec::new();
for (index, (name, element_ty)) in shape
.header_out
.iter()
.zip(elements.iter().skip(1))
.enumerate()
{
let raw_ident = format!("rawHeaderOut{index}");
let ident = format!("headerOut{index}");
let element_typename = get_field_def("value", element_ty, "").typescript_typename();
let _ = write!(
stmt,
" const {raw_ident} = response.headers.find(\n \
([name]) => name.toLowerCase() === \"{name}\",\n \
);\n \
if ({raw_ident} === undefined) {{\n \
return {{\n \
ok: false,\n \
error: {{\n \
isServiceFault: true,\n \
fault: {prefix}HttpUndeserializablePayload(\n \
\"{wire}\",\n \
\"a declared response header was missing\",\n \
),\n \
}},\n \
}};\n \
}}\n"
);
let value_expr = header_out_value_expr(element_ty, &format!("{raw_ident}[1]"));
let _ = writeln!(
stmt,
" const {ident} = {value_expr} as {element_typename};"
);
header_idents.push(ident);
}
let (header_stmts, header_idents) = header_out_read_stmts(prefix, wire, shape, &elements, 1);
stmt.push_str(&header_stmts);
let _ = writeln!(
stmt,
" return {{ ok: true, value: [value, {}] }};",
Expand Down Expand Up @@ -694,11 +685,131 @@ fn bytes_success_decode_block(
return stmt;
}
let elements: Vec<&Type> = tuple_elements(success).into_iter().flatten().collect();
let mut header_idents = Vec::new();
let (header_stmts, header_idents) = header_out_read_stmts(prefix, wire, shape, &elements, 2);
stmt.push_str(&header_stmts);
let _ = writeln!(
stmt,
" return {{ ok: true, value: [response.body, contentType, {}] }};",
header_idents.join(", ")
);
stmt
}

/// A `body = "stream"` operation's own decode: `206` answers the streamed record with its own
/// `contentRange` read back off the response ahead of the body; the declared `ok_status` answers
/// the same record with `contentRange` left `undefined`; everything else falls through the same
/// declared-error, fixed-fault and unexpected-status ladder every other kind answers through —
/// mirrors the Rust and Dart clients' own stream decode.
fn stream_reply_decode_stmt(
prefix: &str,
shape: &HttpShape,
wire: &str,
error: &Type,
success: &Type,
) -> String {
let ok_status = shape.ok_status;
let error_condition = error_condition_expr(shape);
let error_ty = get_field_def("error", error, "").typescript_typename();
let partial = stream_success_arm(prefix, wire, shape, success, true);
let full = stream_success_arm(prefix, wire, shape, success, false);
format!(
" const status = response.status;\n \
if (status === 206) {{\n\
{partial} \
}}\n \
if (status === {ok_status}) {{\n\
{full} \
}}\n \
if ({error_condition}) {{\n \
let declared: {error_ty};\n \
try {{\n \
declared = JSON.parse(response.body) as {error_ty};\n \
}} catch (rejected) {{\n \
return {{\n \
ok: false,\n \
error: {{\n \
isServiceFault: true,\n \
fault: {prefix}HttpUndeserializablePayload(\"{wire}\", String(rejected)),\n \
}},\n \
}};\n \
}}\n \
return {{ ok: false, error: declared }};\n \
}}\n \
if (status === 400 || status === 404 || status === 500) {{\n \
return {{\n \
ok: false,\n \
error: {{\n \
isServiceFault: true,\n \
fault: {prefix}HttpFaultFromBody(\"{wire}\", response.body),\n \
}},\n \
}};\n \
}}\n \
return {{\n \
ok: false,\n \
error: {{\n \
isServiceFault: true,\n \
fault: {prefix}HttpUndeserializablePayload(\n \
\"{wire}\",\n \
`an unexpected status (${{status}}) answered`,\n \
),\n \
}},\n \
}};\n"
)
}

/// One status arm of [`stream_reply_decode_stmt`]: `contentRange` read back off the response for a
/// `206` partial answer, left `undefined` for the declared `ok_status`'s whole-body answer — both
/// paired with `response.bodyStream`, the seam's own lazily-pulled source — then every declared
/// `header_out` element read back exactly as the bytes and JSON paths do.
fn stream_success_arm(
prefix: &str,
wire: &str,
shape: &HttpShape,
success: &Type,
partial: bool,
) -> String {
let mut stmt = if partial {
" const contentRange =\n \
response.headers.find(\n \
([name]) => name.toLowerCase() === \"content-range\",\n \
)?.[1] ?? \"\";\n"
.to_owned()
} else {
" const contentRange: string | undefined = undefined;\n".to_owned()
};
stmt.push_str(" const answer = { contentRange, body: response.bodyStream };\n");
if shape.header_out.is_empty() {
stmt.push_str(" return { ok: true, value: answer };\n");
return stmt;
}
let elements: Vec<&Type> = tuple_elements(success).into_iter().flatten().collect();
let (header_stmts, header_idents) = header_out_read_stmts(prefix, wire, shape, &elements, 1);
stmt.push_str(&header_stmts);
let _ = writeln!(
stmt,
" return {{ ok: true, value: [answer, {}] }};",
header_idents.join(", ")
);
stmt
}

/// Reads every declared `header_out` element back off the response headers, in declaration order,
/// skipping `body_elements` positions in `elements` to reach the first header slot — 1 for the
/// ordinary JSON and streamed shapes (the value or answer alone), 2 for `body = "bytes"` (the bytes
/// and their content type). Shared by every success-decoding arm so the copies cannot drift.
fn header_out_read_stmts(
prefix: &str,
wire: &str,
shape: &HttpShape,
elements: &[&Type],
body_elements: usize,
) -> (String, Vec<String>) {
let mut stmt = String::new();
let mut idents = Vec::new();
for (index, (name, element_ty)) in shape
.header_out
.iter()
.zip(elements.iter().skip(2))
.zip(elements.iter().skip(body_elements))
.enumerate()
{
let raw_ident = format!("rawHeaderOut{index}");
Expand Down Expand Up @@ -727,14 +838,9 @@ fn bytes_success_decode_block(
stmt,
" const {ident} = {value_expr} as {element_typename};"
);
header_idents.push(ident);
idents.push(ident);
}
let _ = writeln!(
stmt,
" return {{ ok: true, value: [response.body, contentType, {}] }};",
header_idents.join(", ")
);
stmt
(stmt, idents)
}

/// The expression that reads a `header_out` element's declared type back off `raw` — a `string`
Expand Down
Loading
Loading