From ff4ceab75fbb135a10d3feb6013d7f150da10d1d Mon Sep 17 00:00:00 2001 From: Roman Volosatovs Date: Mon, 26 Oct 2026 18:01:21 +0100 Subject: [PATCH 1/2] feat(go)!: represent futures and streams as channels A `future` is now a `<-chan T`, which receives the value once the future resolves and is closed afterwards, and a `stream` is a `<-chan []T` of chunks, which is closed once the stream ends. The values cannot fail: as in Rust, a future never resolves after a failed read, and a stream ends. Readers of pending values return a deferred read alongside the value, like writers return a deferred write, and its error is reported on the client's error channel, renamed from `writeErrs__` to `errs__`, or logged by the server. The deferred read takes a `context.Context` and stops once it is done, so an unconsumed channel does not block it forever. The client reads with the invocation context, and the server cancels the context of the deferred parameter reads once the handler returned and the results were written. `IndexReader` gains `Buffered`, which a pending future read uses to reject data received after the value, like Rust does. This also fixes `future`, `future>` and futures held by resources, which did not compile before. Assisted-by: claude:claude-opus-5-5 --- crates/wit-bindgen-go/src/interface.rs | 1102 ++++++----------- crates/wit-bindgen-test/src/go.rs | 4 +- examples/go/streams-client/app.go | 89 +- .../streams/handler/bindings.wrpc.go | 440 ++----- examples/go/streams-server/app.go | 4 +- .../streams/handler/bindings.wrpc.go | 581 ++------- .../wasi/keyvalue/store/bindings.wrpc.go | 33 +- .../wasi/keyvalue/store/bindings.wrpc.go | 33 +- go/frame_reader.go | 4 + go/future.go | 88 +- go/stream.go | 398 +++--- go/wrpc.go | 91 +- tests/go/async.go | 16 +- tests/go/async_test.go | 134 +- tests/go/internal/internal.go | 8 + 15 files changed, 1009 insertions(+), 2016 deletions(-) diff --git a/crates/wit-bindgen-go/src/interface.rs b/crates/wit-bindgen-go/src/interface.rs index c797c40b0..ff70d5403 100644 --- a/crates/wit-bindgen-go/src/interface.rs +++ b/crates/wit-bindgen-go/src/interface.rs @@ -10,7 +10,7 @@ use wit_bindgen_core::wit_parser::{ Variant, WorldKey, }; use wit_bindgen_core::{Source, TypeInfo, uwrite, uwriteln}; -use wrpc_introspect::{async_paths_ty, is_list_of, is_tuple, is_ty, rpc_func_name}; +use wrpc_introspect::{async_paths_ty, is_tuple, is_ty, rpc_func_name}; use crate::{ Deps, GoWrpc, Identifier, InterfaceName, to_go_ident, to_package_ident, to_upper_camel_case, @@ -578,8 +578,11 @@ impl InterfaceGenerator<'_> { let errors = self.deps.errors(); let slog = self.deps.slog(); let wrpc = self.deps.wrpc(); + let deferred = self.is_async(ty); + let d = if deferred { "nil, " } else { "" }; uwrite!(self.src, "func(r {wrpc}.IndexReadCloser, path ...uint32) ("); self.print_list(ty); + self.print_deferred_ret(deferred); uwrite!( self.src, r#", error) {{ @@ -592,41 +595,62 @@ impl InterfaceGenerator<'_> { if i > 0 && err == {io}.EOF {{ err = {io}.ErrUnexpectedEOF }} - return nil, {fmt}.Errorf("failed to read list length byte: %w", err) + return nil, {d}{fmt}.Errorf("failed to read list length byte: %w", err) }} if s == 28 && b > 0x0f {{ - return nil, {errors}.New("list length overflows a 32-bit integer") + return nil, {d}{errors}.New("list length overflows a 32-bit integer") }} if b < 0x80 {{ x = x | uint32(b)< { let fmt = self.deps.fmt(); let slog = self.deps.slog(); let wrpc = self.deps.wrpc(); + let deferred = self.is_async(ty); + let (d, dv) = if deferred { ("nil, ", "d, ") } else { ("", "") }; uwrite!(self.src, "func(r {wrpc}.IndexReadCloser, path ...uint32) ("); self.print_option(ty, true); + self.print_deferred_ret(deferred); uwrite!( self.src, r#", error) {{ {slog}.Debug("reading option status byte") status, err := r.ReadByte() if err != nil {{ - return nil, {fmt}.Errorf("failed to read option status byte: %w", err) + return nil, {d}{fmt}.Errorf("failed to read option status byte: %w", err) }} switch status {{ case 0: - return nil, nil + return nil, {d}nil case 1: {slog}.Debug("reading `option::some` payload") - v, err := "#, + v, {dv}err := "#, ); self.print_read_ty(ty, "r", "path"); self.push_str("\n"); uwrite!( self.src, r#"if err != nil {{ - return nil, {fmt}.Errorf("failed to read `option::some` value: %w", err) + return nil, {d}{fmt}.Errorf("failed to read `option::some` value: %w", err) }} return "#, ); self.print_nillable_ptr(ty, false, false); uwrite!( self.src, - r#"v, nil + r#"v, {dv}nil default: - return nil, {fmt}.Errorf("invalid option status byte %d", status) + return nil, {d}{fmt}.Errorf("invalid option status byte %d", status) }} }}({reader}"#, ); @@ -688,71 +715,65 @@ impl InterfaceGenerator<'_> { let fmt = self.deps.fmt(); let slog = self.deps.slog(); let wrpc = self.deps.wrpc(); + let deferred = ty.ok.iter().chain(&ty.err).any(|ty| self.is_async(ty)); + let d = if deferred { "nil, " } else { "" }; uwrite!( self.src, "func(r {wrpc}.IndexReadCloser, path ...uint32) (*" ); self.print_result(ty); + self.print_deferred_ret(deferred); uwriteln!( self.src, r#", error) {{ {slog}.Debug("reading result status byte") status, err := r.ReadByte() if err != nil {{ - return nil, {fmt}.Errorf("failed to read result status byte: %w", err) + return nil, {d}{fmt}.Errorf("failed to read result status byte: %w", err) }}"#, ); self.push_str("switch status {\n"); - self.push_str("case 0:\n"); - if let Some(ref ok) = ty.ok { - uwriteln!(self.src, r#"{slog}.Debug("reading `result::ok` payload")"#); - self.push_str("v, err := "); - self.print_read_ty(ok, "r", "path"); - self.push_str("\n"); - uwriteln!( - self.src, - r#"if err != nil {{ - return nil, fmt.Errorf("failed to read `result::ok` value: %w", err) - }}"#, - ); - self.push_str("return &"); - self.print_result(ty); - self.push_str("{ Ok: "); - self.print_nillable_ptr(ok, true, false); - } else { - self.push_str("var v struct{}\n"); - self.push_str("return &"); - self.print_result(ty); - self.push_str("{ Ok: &"); - } - self.push_str("v }, nil\n"); - self.push_str("case 1:\n"); - if let Some(ref err) = ty.err { - uwriteln!(self.src, r#"{slog}.Debug("reading `result::err` payload")"#); - self.push_str("v, err := "); - self.print_read_ty(err, "r", "path"); - self.push_str("\n"); - uwriteln!( - self.src, - r#"if err != nil {{ - return nil, {fmt}.Errorf("failed to read `result::err` value: %w", err) + for (i, (case, field, payload)) in [("ok", "Ok", &ty.ok), ("err", "Err", &ty.err)] + .into_iter() + .enumerate() + { + uwriteln!(self.src, "case {i}:"); + let mut ret = d; + if let Some(payload) = payload { + uwriteln!( + self.src, + r#"{slog}.Debug("reading `result::{case}` payload")"# + ); + if self.is_async(payload) { + self.push_str("v, d, err := "); + ret = "d, "; + } else { + self.push_str("v, err := "); + } + self.print_read_ty(payload, "r", "path"); + self.push_str("\n"); + uwriteln!( + self.src, + r#"if err != nil {{ + return nil, {d}{fmt}.Errorf("failed to read `result::{case}` value: %w", err) }}"#, - ); - self.push_str("return &"); - self.print_result(ty); - self.push_str("{ Err: "); - self.print_nillable_ptr(err, true, false); - } else { - self.push_str("var v struct{}\n"); - self.push_str("return &"); - self.print_result(ty); - self.push_str("{ Err: &"); + ); + self.push_str("return &"); + self.print_result(ty); + uwrite!(self.src, "{{ {field}: "); + self.print_nillable_ptr(payload, true, false); + } else { + self.push_str("var v struct{}\n"); + self.push_str("return &"); + self.print_result(ty); + uwrite!(self.src, "{{ {field}: &"); + } + uwriteln!(self.src, "v }}, {ret}nil"); } uwrite!( self.src, - r#"v }}, nil - default: - return nil, {fmt}.Errorf("invalid result status byte %d", status) + r#"default: + return nil, {d}{fmt}.Errorf("invalid result status byte %d", status) }} }}({reader}"# ); @@ -772,32 +793,57 @@ impl InterfaceGenerator<'_> { name: &str, ) { let wrpc = self.deps.wrpc(); - + let deferred = fields.iter().any(|Field { ty, .. }| self.is_async(ty)); + let d = if deferred { "nil, " } else { "" }; + uwrite!( + self.src, + "func(r {wrpc}.IndexReadCloser, path ...uint32) (*{name}" + ); + self.print_deferred_ret(deferred); uwriteln!( self.src, - r#"func(r {wrpc}.IndexReadCloser, path ...uint32) (*{name}, error) {{ + r#", error) {{ v := &{name}{{}} var err error"# ); + if deferred { + let context = self.deps.context(); + uwrite!( + self.src, + "var d func({context}.Context) error\nvar ds []func({context}.Context) error\n" + ); + } for (i, Field { name, ty, .. }) in fields.iter().enumerate() { let fmt = self.deps.fmt(); let slog = self.deps.slog(); let ident = name.to_upper_camel_case(); + let field_deferred = self.is_async(ty); uwrite!( self.src, r#"{slog}.Debug("reading field", "name", "{name}") - v.{ident}, err = "# + v.{ident}, "# ); + if field_deferred { + self.push_str("d, "); + } + self.push_str("err = "); self.print_read_ty(ty, "r", &format!("append(path, {i})")); self.push_str("\n"); uwriteln!( self.src, r#"if err != nil {{ - return nil, {fmt}.Errorf("failed to read `{name}` field: %w", err) + return nil, {d}{fmt}.Errorf("failed to read `{name}` field: %w", err) }}"# ); + if field_deferred { + self.push_str("ds = append(ds, d)\n"); + } + } + if deferred { + uwriteln!(self.src, "return v, {wrpc}.JoinDeferred(ds...), nil"); + } else { + self.push_str("return v, nil\n"); } - self.push_str("return v, nil\n"); uwrite!(self.src, "}}({reader}"); if !path.is_empty() { self.src.push_str(", "); @@ -864,19 +910,30 @@ impl InterfaceGenerator<'_> { fn print_read_variant(&mut self, ty: &Variant, reader: &str, path: &str, name: &str) { let fmt = self.deps.fmt(); let wrpc = self.deps.wrpc(); - + let deferred = ty + .cases + .iter() + .any(|Case { ty, .. }| ty.as_ref().is_some_and(|ty| self.is_async(ty))); + let d = if deferred { "nil, " } else { "" }; uwrite!( self.src, - r#"func(r {wrpc}.IndexReadCloser, path ...uint32) (*{name}, error) {{ + "func(r {wrpc}.IndexReadCloser, path ...uint32) (*{name}" + ); + self.print_deferred_ret(deferred); + uwrite!( + self.src, + r#", error) {{ v := &{name}{{}} n, err := "# ); self.print_read_discriminant(ty.tag(), "r"); self.push_str("\n"); - self.push_str("if err != nil {\n"); - self.push_str("return nil, "); - self.push_str(fmt); - self.push_str(".Errorf(\"failed to read discriminant: %w\", err)\n}\n"); + uwriteln!( + self.src, + r#"if err != nil {{ + return nil, {d}{fmt}.Errorf("failed to read discriminant: %w", err) + }}"# + ); uwriteln!(self.src, "switch {name}Discriminant(n) {{"); for Case { name: case_name, @@ -885,28 +942,29 @@ impl InterfaceGenerator<'_> { } in &ty.cases { let camel = case_name.to_upper_camel_case(); - self.push_str("case "); - self.push_str(name); - self.push_str(&camel); - self.push_str(":\n"); + uwriteln!(self.src, "case {name}{camel}:"); if let Some(ty) = ty { - self.push_str("payload, err := "); + let ret = if self.is_async(ty) { + self.push_str("payload, d, err := "); + "d, " + } else { + self.push_str("payload, err := "); + d + }; self.print_read_ty(ty, "r", "path"); self.push_str("\n"); - self.push_str("if err != nil { return nil, "); - self.push_str(fmt); uwriteln!( self.src, - r#".Errorf("failed to read `{case_name}` payload: %w", err) }}"# + r#"if err != nil {{ return nil, {d}{fmt}.Errorf("failed to read `{case_name}` payload: %w", err) }}"# ); - uwriteln!(self.src, "return v.Set{camel}(payload), nil"); + uwriteln!(self.src, "return v.Set{camel}(payload), {ret}nil"); } else { - uwriteln!(self.src, "return v.Set{camel}(), nil"); + uwriteln!(self.src, "return v.Set{camel}(), {d}nil"); } } uwriteln!( self.src, - r#"default: return nil, {fmt}.Errorf("unknown discriminant value %d", n) }}"# + r#"default: return nil, {d}{fmt}.Errorf("unknown discriminant value %d", n) }}"# ); uwrite!(self.src, "}}({reader}"); if !path.is_empty() { @@ -929,35 +987,56 @@ impl InterfaceGenerator<'_> { } _ => { let wrpc = self.deps.wrpc(); - + let deferred = ty.types.iter().any(|ty| self.is_async(ty)); + let d = if deferred { "nil, " } else { "" }; uwrite!( self.src, "func(r {wrpc}.IndexReadCloser, path ...uint32) (*" ); self.print_tuple(ty, true); + self.print_deferred_ret(deferred); self.push_str(", error) {\n"); self.push_str("v := &"); self.print_tuple(ty, false); self.push_str("{}\n"); self.push_str("var err error\n"); + if deferred { + let context = self.deps.context(); + uwrite!( + self.src, + "var d func({context}.Context) error\nvar ds []func({context}.Context) error\n" + ); + } for (i, ty) in ty.types.iter().enumerate() { let fmt = self.deps.fmt(); let slog = self.deps.slog(); + let elem_deferred = self.is_async(ty); uwrite!( self.src, r#"{slog}.Debug("reading tuple element {i}") - v.V{i}, err = "# + v.V{i}, "# ); + if elem_deferred { + self.push_str("d, "); + } + self.push_str("err = "); self.print_read_ty(ty, "r", &format!("append(path, {i})")); self.push_str("\n"); uwriteln!( self.src, r#"if err != nil {{ - return nil, {fmt}.Errorf("failed to read tuple element {i}: %w", err) + return nil, {d}{fmt}.Errorf("failed to read tuple element {i}: %w", err) }}"# ); + if elem_deferred { + self.push_str("ds = append(ds, d)\n"); + } + } + if deferred { + uwriteln!(self.src, "return v, {wrpc}.JoinDeferred(ds...), nil"); + } else { + self.push_str("return v, nil\n"); } - self.push_str("return v, nil\n"); uwrite!(self.src, "}}({reader}"); if !path.is_empty() { self.src.push_str(", "); @@ -970,286 +1049,56 @@ impl InterfaceGenerator<'_> { } fn print_read_future(&mut self, ty: &Option, reader: &str, path: &str) { - match ty { - Some(ty) if is_list_of(self.resolve, Type::U8, ty) => { - let bytes = self.deps.bytes(); - let io = self.deps.io(); - let fmt = self.deps.fmt(); - let slog = self.deps.slog(); - let wrpc = self.deps.wrpc(); - - uwriteln!( - self.src, - r#"func(r {wrpc}.IndexReadCloser, path ...uint32) ({io}.ReadCloser, error) {{ - {slog}.Debug("reading byte list future status byte") - status, err := r.ReadByte() - if err != nil {{ - return nil, {fmt}.Errorf("failed to read byte list future status byte: %w", err) - }} - switch status {{ - case 0: - {slog}.Debug("indexing pending byte list future reader") - if len(path) > 0 {{ - var err error - r, err = r.Index(path...) - if err != nil {{ - return nil, {fmt}.Errorf("failed to index nested byte list future reader: %w", err) - }} - }} - return {wrpc}.NewByteStreamReader(r), nil - case 1: - {slog}.Debug("reading ready byte list future contents") - buf, err := "# - ); - self.print_read_byte_list("r"); - uwrite!( - self.src, - r#" - if err != nil {{ - return nil, {fmt}.Errorf("failed to read ready byte list future contents: %w", err) - }} - {slog}.Debug("read ready byte list future contents", "len", len(buf)) - return {io}.NopCloser({bytes}.NewReader(buf)), nil - default: - return nil, {fmt}.Errorf("invalid byte list future status byte %d", status) - }} -}}({reader}"# - ); - if !path.is_empty() { - self.src.push_str(", "); - self.src.push_str(path); - self.src.push_str("..."); - } - self.src.push_str(")\n"); - } - Some(ty) => { - let fmt = self.deps.fmt(); - let slog = self.deps.slog(); - let wrpc = self.deps.wrpc(); - - uwrite!( - self.src, - r#"func(r {wrpc}.IndexReadCloser, path ...uint32) ({wrpc}.Receiver["# - ); - self.print_opt_ty(ty, true); - uwrite!( - self.src, - r#"], error) {{ - {slog}.Debug("reading future status byte") - status, err := r.ReadByte() - if err != nil {{ - return nil, {fmt}.Errorf("failed to read future status byte: %w", err) - }} - switch status {{ - case 0: - {slog}.Debug("indexing pending future reader") - if len(path) > 0 {{ - var err error - r, err = r.Index(path...) - if err != nil {{ - return nil, {fmt}.Errorf("failed to index nested future reader: %w", err) - }} - }} - return {wrpc}.NewDecodeReceiver(r, func(r {wrpc}.IndexReadCloser) ("# - ); - self.print_opt_ty(ty, true); - uwrite!( - self.src, - r#", error) {{ - {slog}.Debug("reading pending future element") - v, err := "# - ); - self.print_read_ty(ty, "r", ""); - uwriteln!( - self.src, - r#" - if err != nil {{ - return v, {fmt}.Errorf("failed to read pending future element: %w", err) - }} - return v, nil - }}), nil - case 1: - {slog}.Debug("reading ready future contents") - v, err := "# - ); - self.print_read_ty(ty, "r", "path"); - uwrite!( - self.src, - r#" - if err != nil {{ - return nil, {fmt}.Errorf("failed to read ready future contents: %w", err) - }} - return {wrpc}.NewCompleteReceiver(v), nil - default: - return nil, {fmt}.Errorf("invalid future status byte %d", status) - }} -}}({reader}"# - ); - if !path.is_empty() { - self.src.push_str(", "); - self.src.push_str(path); - self.src.push_str("..."); - } - self.src.push_str(")"); - } - None => panic!("futures with no element types are not supported"), + let Some(ty) = ty else { + panic!("futures with no element types are not supported") + }; + let wrpc = self.deps.wrpc(); + uwrite!(self.src, "{wrpc}.ReadFuture({reader}, "); + self.print_read_element(ty); + if !path.is_empty() { + self.src.push_str(", "); + self.src.push_str(path); + self.src.push_str("..."); } + self.src.push_str(")"); } - fn print_read_stream(&mut self, element: &Option, reader: &str, path: &str) { - match element { - Some(ty) if is_ty(self.resolve, Type::U8, ty) => { - let bytes = self.deps.bytes(); - let io = self.deps.io(); - let fmt = self.deps.fmt(); - let slog = self.deps.slog(); - let wrpc = self.deps.wrpc(); - - uwriteln!( - self.src, - r#"func(r {wrpc}.IndexReadCloser, path ...uint32) ({io}.ReadCloser, error) {{ - {slog}.Debug("reading byte stream status byte") - status, err := r.ReadByte() - if err != nil {{ - return nil, {fmt}.Errorf("failed to read byte stream status byte: %w", err) - }} - switch status {{ - case 0: - if len(path) > 0 {{ - var err error - r, err = r.Index(path...) - if err != nil {{ - return nil, {fmt}.Errorf("failed to index nested byte stream reader: %w", err) - }} - }} - return {wrpc}.NewByteStreamReader(r), nil - case 1: - {slog}.Debug("reading ready byte stream contents") - buf, err := "# - ); - self.print_read_byte_list("r"); - uwrite!( - self.src, - r#" - if err != nil {{ - return nil, {fmt}.Errorf("failed to read ready byte stream contents: %w", err) - }} - {slog}.Debug("read ready byte stream contents", "len", len(buf)) - return {io}.NopCloser({bytes}.NewReader(buf)), nil - default: - return nil, {fmt}.Errorf("invalid stream status byte %d", status) - }} -}}({reader}"# - ); - if !path.is_empty() { - self.src.push_str(", "); - self.src.push_str(path); - self.src.push_str("..."); - } - self.src.push_str(")"); - } - Some(ty) => { - let errors = self.deps.errors(); - let fmt = self.deps.fmt(); - let io = self.deps.io(); - let math = self.deps.math(); - let slog = self.deps.slog(); - let wrpc = self.deps.wrpc(); - - uwrite!( - self.src, - r#"func(r {wrpc}.IndexReadCloser, path ...uint32) ({wrpc}.Receiver["# - ); - self.print_list(ty); - uwrite!( - self.src, - r#"], error) {{ - {slog}.Debug("reading stream status byte") - status, err := r.ReadByte() - if err != nil {{ - return nil, {fmt}.Errorf("failed to read stream status byte: %w", err) - }} - switch status {{ - case 0: - if len(path) > 0 {{ - var err error - r, err = r.Index(path...) - if err != nil {{ - return nil, {fmt}.Errorf("failed to index nested stream reader: %w", err) - }} - }} - var total uint32 - return {wrpc}.NewDecodeReceiver(r, func(r {wrpc}.IndexReadCloser) ("# - ); - self.print_list(ty); - uwrite!( - self.src, - r#", error) {{ - {slog}.Debug("reading pending stream chunk length") - n, err := "# - ); - self.print_read_u32("r"); - uwrite!( - self.src, - r#" - if err != nil {{ - return nil, {fmt}.Errorf("failed to read pending stream chunk length: %w", err) - }} - if n == 0 {{ - return nil, {io}.EOF - }} - if {math}.MaxUint32 - n < total {{ - return nil, {errors}.New("total incoming pending stream element count would overflow a 32-bit unsigned integer") - }} - vs := {wrpc}.NewSlice["# - ); - self.print_list(ty); - uwrite!( - self.src, - r#"](n) - for i := range n {{ - {slog}.Debug("reading pending stream element", "i", total) - v, err := "# - ); - self.print_read_ty(ty, "r", "[]uint32{total}"); - uwriteln!( - self.src, - r#" - if err != nil {{ - return nil, {fmt}.Errorf("failed to read pending stream chunk element %d: %w", i, err) - }} - vs = append(vs, v) - total++ - }} - return vs, nil - }}), nil - case 1: - {slog}.Debug("reading ready stream contents") - vs, err := "# - ); - self.print_read_list(ty, "r", "path"); - uwrite!( - self.src, - r#" - if err != nil {{ - return nil, {fmt}.Errorf("failed to read ready stream contents: %w", err) - }} - {slog}.Debug("read ready stream contents", "len", len(vs)) - return {wrpc}.NewCompleteReceiver(vs), nil - default: - return nil, {fmt}.Errorf("invalid stream status byte %d", status) - }} -}}({reader}"# - ); - if !path.is_empty() { - self.src.push_str(", "); - self.src.push_str(path); - self.src.push_str("..."); - } - self.src.push_str(")"); + fn print_read_element(&mut self, ty: &Type) { + let wrpc = self.deps.wrpc(); + uwrite!(self.src, "func(r {wrpc}.IndexReadCloser, path ...uint32) ("); + self.print_opt_ty(ty, true); + let context = self.deps.context(); + uwrite!(self.src, ", func({context}.Context) error, error) {{\n"); + if self.is_async(ty) { + self.push_str("return "); + self.print_read_ty(ty, "r", "path"); + self.push_str("\n"); + } else { + self.push_str("v, err := "); + self.print_read_ty(ty, "r", "path"); + self.push_str("\nreturn v, nil, err\n"); + } + self.push_str("}"); + } + + fn print_read_stream(&mut self, element: &Option, reader: &str, path: &str) { + let wrpc = self.deps.wrpc(); + match element { + Some(ty) if is_ty(self.resolve, Type::U8, ty) => { + uwrite!(self.src, "{wrpc}.ReadByteStream({reader}"); + } + Some(ty) => { + uwrite!(self.src, "{wrpc}.ReadStream({reader}, "); + self.print_read_element(ty); } None => panic!("streams with no element types are not supported"), } + if !path.is_empty() { + self.src.push_str(", "); + self.src.push_str(path); + self.src.push_str("..."); + } + self.src.push_str(")"); } fn print_read_own(&mut self, reader: &str, id: TypeId) { @@ -1404,17 +1253,27 @@ impl InterfaceGenerator<'_> { TypeDefKind::Stream(ty) => self.print_read_stream(ty, reader, path), TypeDefKind::Type(ty) => { if let Some(name) = name { + let deferred = self.is_async(ty); self.push_str("func() ("); self.print_opt_ptr(ty, true); self.push_str(&name); + self.print_deferred_ret(deferred); self.push_str(", error) {\n"); - self.push_str("v, err :="); + if deferred { + self.push_str("v, d, err :="); + } else { + self.push_str("v, err :="); + } self.print_read_ty(ty, reader, path); self.push_str("\n"); self.push_str("return ("); self.print_opt_ptr(ty, true); self.push_str(&name); - self.push_str(")(v), err\n"); + if deferred { + self.push_str(")(v), d, err\n"); + } else { + self.push_str(")(v), err\n"); + } self.push_str("}()\n"); } else { self.print_read_ty(ty, reader, path); @@ -1734,7 +1593,7 @@ impl InterfaceGenerator<'_> { if len(writes) > 0 {{ return func(w {wrpc}.IndexWriter) error {{ var wg {sync}.WaitGroup - var wgErr {atomic}.Value + var wgErr {atomic}.Pointer[error] for index, write := range writes {{ wg.Add(1) w, err := w.Index(index) @@ -1745,16 +1604,15 @@ impl InterfaceGenerator<'_> { go func() {{ defer wg.Done() if err := write(w); err != nil {{ - wgErr.Store(err) + wgErr.Store(&err) }} }}() }} wg.Wait() - err := wgErr.Load() - if err == nil {{ - return nil + if err := wgErr.Load(); err != nil {{ + return *err }} - return err.(error) + return nil }}, nil }} return nil, nil @@ -1961,7 +1819,7 @@ impl InterfaceGenerator<'_> { r#"if len(writes) > 0 {{ return func(w {wrpc}.IndexWriter) error {{ var wg {sync}.WaitGroup - var wgErr {atomic}.Value + var wgErr {atomic}.Pointer[error] for index, write := range writes {{ wg.Add(1) w, err := w.Index(index) @@ -1972,16 +1830,15 @@ impl InterfaceGenerator<'_> { go func() {{ defer wg.Done() if err := write(w); err != nil {{ - wgErr.Store(err) + wgErr.Store(&err) }} }}() }} wg.Wait() - err := wgErr.Load() - if err == nil {{ - return nil + if err := wgErr.Load(); err != nil {{ + return *err }} - return err.(error) + return nil }}, nil }} return nil, nil @@ -1992,267 +1849,37 @@ impl InterfaceGenerator<'_> { } fn print_write_future(&mut self, ty: &Option, name: &str, writer: &str) { - match ty { - Some(ty) if is_list_of(self.resolve, Type::U8, ty) => { - let fmt = self.deps.fmt(); - let io = self.deps.io(); - let math = self.deps.math(); - let slog = self.deps.slog(); - let wrpc = self.deps.wrpc(); - uwrite!( - self.src, - r#"func(v {io}.Reader, w interface {{ {io}.ByteWriter; {io}.Writer }}) (write func({wrpc}.IndexWriter) error, err error) {{ - {slog}.Debug("writing byte list future `future::pending` status byte") - if err = w.WriteByte(0); err != nil {{ - return nil, fmt.Errorf("failed to write `future::pending` byte: %w", err) - }} - return func(w {wrpc}.IndexWriter) (err error) {{ - defer func() {{ - body, ok := v.({io}.Closer) - if ok {{ - {slog}.Debug("closing byte list future reader") - if cErr := body.Close(); cErr != nil {{ - if err == nil {{ - err = {fmt}.Errorf("failed to close pending byte list future: %w", cErr) - }} else {{ - {slog}.Warn("failed to close pending byte list future", "err", cErr) - }} - }} - }} - }}() - {slog}.Debug("reading pending byte list future contents") - chunk, err := {io}.ReadAll(chunk) - if err != nil {{ - return {fmt}.Errorf("failed to read pending byte list future: %w", err) - }} - if n > {math}.MaxUint32 {{ - return {fmt}.Errorf("pending byte list future length of %d overflows a 32-bit integer", n) - }} - {slog}.Debug("writing pending byte list future length", "len", n) - _, err = {wrpc}.WriteUint32(uint32(n), w) - if err != nil {{ - return {fmt}.Errorf("failed to write pending byte list future length of %d: %w", n, err) - }} - {slog}.Debug("writing pending byte list future contents", "buf", chunk[:n]) - _, err = w.Write(chunk[:n]) - if err != nil {{ - return {fmt}.Errorf("failed to write pending byte list future contents: %w", err) - }} - }}, nil - }}({name}, {writer})"#, - ); - } - Some(ty) => { - let fmt = self.deps.fmt(); - let io = self.deps.io(); - let slog = self.deps.slog(); - let wrpc = self.deps.wrpc(); - uwrite!(self.src, "func(v {wrpc}.Receiver[",); - self.print_opt_ty(ty, true); - uwrite!( - self.src, - r#"], w interface {{ {io}.ByteWriter; {io}.Writer }}) (write func({wrpc}.IndexWriter) error, err error) {{ - {slog}.Debug("writing future `future::pending` status byte") - if err := w.WriteByte(0); err != nil {{ - return nil, fmt.Errorf("failed to write `future::pending` byte: %w", err) - }} - return func(w {wrpc}.IndexWriter) (err error) {{ - defer func() {{ - {slog}.Debug("closing future writer") - if cErr := v.Close(); cErr != nil {{ - if err == nil {{ - err = {fmt}.Errorf("failed to close pending future: %w", cErr) - }} else {{ - {slog}.Warn("failed to close pending future", "err", cErr) - }} - }} - }}() - {slog}.Debug("receiving outgoing pending future contents") - rx, err := v.Receive() - if err != nil {{ - return {fmt}.Errorf("failed to receive outgoing pending future: %w", err) - }} - {slog}.Debug("writing pending future element") - write, err :="#, - ); - self.print_write_ty(ty, "rx", "w"); - uwrite!( - self.src, - r#" - if err != nil {{ - return {fmt}.Errorf("failed to write pending future element: %w", err) - }} - if write != nil {{ - return write(w) - }} - return nil - }}, nil - }}({name}, {writer})"#, - ); - } - None => panic!("streams with no element types are not supported"), - } + let Some(ty) = ty else { + panic!("futures with no element types are not supported") + }; + let wrpc = self.deps.wrpc(); + uwrite!(self.src, "{wrpc}.WriteFuture({name}, {writer}, "); + self.print_write_element(ty); + self.push_str(")"); + } + + fn print_write_element(&mut self, ty: &Type) { + let wrpc = self.deps.wrpc(); + self.push_str("func(v "); + self.print_opt_ty(ty, true); + uwrite!( + self.src, + ", w {wrpc}.ByteWriter) (func({wrpc}.IndexWriter) error, error) {{\nreturn " + ); + self.print_write_ty(ty, "v", "w"); + self.push_str("\n}"); } fn print_write_stream(&mut self, element: &Option, name: &str, writer: &str) { + let wrpc = self.deps.wrpc(); match element { Some(ty) if is_ty(self.resolve, Type::U8, ty) => { - let fmt = self.deps.fmt(); - let io = self.deps.io(); - let math = self.deps.math(); - let slog = self.deps.slog(); - let wrpc = self.deps.wrpc(); - uwrite!( - self.src, - r#"func(v {io}.ReadCloser, w interface {{ {io}.ByteWriter; {io}.Writer }}) (write func({wrpc}.IndexWriter) error, err error) {{ - {slog}.Debug("writing byte stream `stream::pending` status byte") - if err = w.WriteByte(0); err != nil {{ - return nil, fmt.Errorf("failed to write `stream::pending` byte: %w", err) - }} - return func(w {wrpc}.IndexWriter) (err error) {{ - defer func() {{ - {slog}.Debug("closing byte list stream writer") - if cErr := v.Close(); cErr != nil {{ - if err == nil {{ - err = {fmt}.Errorf("failed to close pending byte stream: %w", cErr) - }} else {{ - {slog}.Warn("failed to close pending byte stream", "err", cErr) - }} - }} - }}() - chunk := make([]byte, {wrpc}.ReadChunkSize) - for {{ - var end bool - {slog}.Debug("reading pending byte stream contents") - n, err := v.Read(chunk) - if err == {io}.EOF {{ - end = true - {slog}.Debug("pending byte stream reached EOF") - }} else if err != nil {{ - return {fmt}.Errorf("failed to read pending byte stream chunk: %w", err) - }} - if n > {math}.MaxUint32 {{ - return {fmt}.Errorf("pending byte stream chunk length of %d overflows a 32-bit integer", n) - }} - if n > 0 {{ - {slog}.Debug("writing pending byte stream chunk length", "len", n) - _, err = {wrpc}.WriteUint32(uint32(n), w) - if err != nil {{ - return {fmt}.Errorf("failed to write pending byte stream chunk length of %d: %w", n, err) - }} - _, err = w.Write(chunk[:n]) - if err != nil {{ - return {fmt}.Errorf("failed to write pending byte stream chunk contents: %w", err) - }} - }} - if end {{ - if err := w.WriteByte(0); err != nil {{ - return {fmt}.Errorf("failed to write pending byte stream end byte: %w", err) - }} - return nil - }} - }} - }}, nil - }}({name}, {writer})"#, - ); + uwrite!(self.src, "{wrpc}.WriteByteStream({name}, {writer})"); } Some(ty) => { - let atomic = self.deps.atomic(); - let errors = self.deps.errors(); - let fmt = self.deps.fmt(); - let io = self.deps.io(); - let math = self.deps.math(); - let slog = self.deps.slog(); - let sync = self.deps.sync(); - let wrpc = self.deps.wrpc(); - uwrite!(self.src, "func(v {wrpc}.Receiver[",); - self.print_list(ty); - uwrite!( - self.src, - r#"], w interface {{ {io}.ByteWriter; {io}.Writer }}) (write func({wrpc}.IndexWriter) error, err error) {{ - {slog}.Debug("writing stream `stream::pending` status byte") - if err := w.WriteByte(0); err != nil {{ - return nil, fmt.Errorf("failed to write `stream::pending` byte: %w", err) - }} - return func(w {wrpc}.IndexWriter) (err error) {{ - defer func() {{ - {slog}.Debug("closing outgoing pending stream") - if cErr := v.Close(); cErr != nil {{ - if err == nil {{ - err = {fmt}.Errorf("failed to close outgoing pending stream: %w", cErr) - }} else {{ - {slog}.Warn("failed to close outgoing pending stream", "err", cErr) - }} - }} - }}() - var wg {sync}.WaitGroup - var wgErr {atomic}.Value - var total uint32 - for {{ - var end bool - {slog}.Debug("receiving outgoing pending stream contents") - chunk, err := v.Receive() - n := len(chunk) - if err == {io}.EOF {{ - end = true - {slog}.Debug("outgoing pending stream reached EOF") - }} else if err != nil {{ - return {fmt}.Errorf("failed to receive outgoing pending stream chunk: %w", err) - }} - if n > {math}.MaxUint32 {{ - return {fmt}.Errorf("outgoing pending stream chunk length of %d overflows a 32-bit integer", n) - }} - if {math}.MaxUint32 - uint32(n) < total {{ - return {errors}.New("total outgoing pending stream element count would overflow a 32-bit unsigned integer") - }} - if n > 0 {{ - {slog}.Debug("writing pending stream chunk length", "len", n) - _, err = {wrpc}.WriteUint32(uint32(n), w) - if err != nil {{ - return {fmt}.Errorf("failed to write pending stream chunk length of %d: %w", n, err) - }} - for _, v := range chunk {{ - {slog}.Debug("writing pending stream element", "i", total) - write, err :="#, - ); - self.print_write_ty(ty, "v", "w"); - uwrite!( - self.src, - r#" - if err != nil {{ - return {fmt}.Errorf("failed to write pending stream chunk element %d: %w", total, err) - }} - if write != nil {{ - wg.Add(1) - w, err := w.Index(total) - if err != nil {{ - return {fmt}.Errorf("failed to index nested stream writer: %w", err) - }} - go func() {{ - defer wg.Done() - if err := write(w); err != nil {{ - wgErr.Store(err) - }} - }}() - }} - total++ - }} - }} - if end {{ - if err := w.WriteByte(0); err != nil {{ - return {fmt}.Errorf("failed to write pending stream end byte: %w", err) - }} - wg.Wait() - err := wgErr.Load() - if err == nil {{ - return nil - }} - return err.(error) - }} - }} - }}, nil - }}({name}, {writer})"#, - ); + uwrite!(self.src, "{wrpc}.WriteStream({name}, {writer}, "); + self.print_write_element(ty); + self.push_str(")"); } None => panic!("streams with no element types are not supported"), } @@ -2566,12 +2193,33 @@ func ServeInterface(s {wrpc}.Server, h Handler) (stop func() error, err error) { }} }}()"#, ); + let async_params = func.params.iter().any(|Param { ty, .. }| self.is_async(ty)); + if async_params { + let sync = self.deps.sync(); + uwrite!( + self.src, + r#" + deferredCtx, cancel := {context}.WithCancel(ctx) + var wg {sync}.WaitGroup + defer func() {{ + go func() {{ + wg.Wait() + cancel() + }}() + }}()"#, + ); + } for (i, Param { ty, .. }) in func.params.iter().enumerate() { + let d = if self.is_async(ty) { + format!("d{i}, ") + } else { + String::new() + }; uwrite!( self.src, r#" {slog}.DebugContext(ctx, "reading parameter", "i", {i}) - p{i}, err := "# + p{i}, {d}err := "# ); self.print_read_ty(ty, "r", &format!("[]uint32{{ {i} }}")); self.push_str("\n"); @@ -2587,6 +2235,21 @@ func ServeInterface(s {wrpc}.Server, h Handler) (stop func() error, err error) { }}"#, ); } + for (i, Param { ty, .. }) in func.params.iter().enumerate() { + if self.is_async(ty) { + uwrite!( + self.src, + r#" + if d{i} != nil {{ + go func() {{ + if err := d{i}(deferredCtx); err != nil {{ + {slog}.WarnContext(ctx, "failed to read deferred parameter", "i", {i}, "instance", "{instance}", "name", "{name}", "err", err) + }} + }}() + }}"#, + ); + } + } uwriteln!( self.src, r#" @@ -2678,8 +2341,27 @@ func ServeInterface(s {wrpc}.Server, h Handler) (stop func() error, err error) { {slog}.ErrorContext(ctx, "failed to index result writer", "instance", "{instance}", "name", "{name}", "err", err) return }} - write := write - go func() {{ + write := write"#, + ); + if async_params { + self.push_str( + r#" + wg.Add(1)"#, + ); + } + self.push_str( + r#" + go func() {"#, + ); + if async_params { + self.push_str( + r#" + defer wg.Done()"#, + ); + } + uwrite!( + self.src, + r#" if err := write(w); err != nil {{ {slog}.WarnContext(ctx, "failed to write nested result value", "instance", "{instance}", "name", "{name}", "err", err) }} @@ -2758,12 +2440,10 @@ func ServeInterface(s {wrpc}.Server, h Handler) (stop func() error, err error) { self.src.push_str(", "); } - let async_params = func.params.iter().any(|param| { - let (paths, fut) = async_paths_ty(self.resolve, ¶m.ty); - fut || !paths.is_empty() - }); - if async_params { - self.push_str("writeErrs__ <-chan error, "); + let async_params = func.params.iter().any(|param| self.is_async(¶m.ty)); + let async_results = results.iter().filter(|ty| self.is_async(ty)).count(); + if async_params || async_results > 0 { + self.push_str("errs__ <-chan error, "); } self.push_str("err__ error) {"); if !func.params.is_empty() { @@ -2888,17 +2568,33 @@ func ServeInterface(s {wrpc}.Server, h Handler) (stop func() error, err error) { }}()"#, name = func.name, ); - if async_params { + if async_params || async_results > 0 { let sync = self.deps.sync(); + let count = if async_params { + format!("uint(writeCount__) + {async_results}") + } else { + async_results.to_string() + }; + uwrite!( + self.src, + r#" + errCh__ := make(chan error, {count}) + errs__ = errCh__ + var wg__ {sync}.WaitGroup + defer func() {{ + go func() {{ + wg__.Wait() + close(errCh__) + }}() + }}()"# + ); + } + if async_params { uwrite!( self.src, r#" if writeCount__ > 0 {{ - writeErrCh__ := make(chan error, uint(writeCount__)) - writeErrs__ = writeErrCh__ - var wg__ {sync}.WaitGroup for index, write := range writes__ {{ - wg__.Add(1) w, err := w__.Index(index) if err != nil {{ if cErr := w__.Close(); cErr != nil {{ @@ -2907,18 +2603,15 @@ func ServeInterface(s {wrpc}.Server, h Handler) (stop func() error, err error) { err__ = {fmt}.Errorf("failed to index param writer at index `%v`: %w", index, err) return }} + wg__.Add(1) write := write go func() {{ defer wg__.Done() if err := write(w); err != nil {{ - writeErrCh__ <- err + errCh__ <- err }} }}() }} - go func() {{ - wg__.Wait() - close(writeErrCh__) - }}() }}"#, func.name, ); @@ -2933,13 +2626,25 @@ func ServeInterface(s {wrpc}.Server, h Handler) (stop func() error, err error) { ); let mut idx = 0usize; + let mut deferreds = Vec::default(); for (i, rty) in func.result.iter().enumerate() { for (j, ty) in flatten_ty(self.resolve, rty).enumerate() { - uwrite!( - self.src, - " + let deferred = self.is_async(&ty); + if deferred { + let context = self.deps.context(); + uwrite!( + self.src, + " + var d{idx}__ func({context}.Context) error + r{idx}__, d{idx}__, err__ = " + ); + } else { + uwrite!( + self.src, + " r{idx}__, err__ = " - ); + ); + } let path = if is_tuple(self.resolve, rty) { format!("[]uint32{{ {i}, {j} }}") } else { @@ -2954,9 +2659,27 @@ func ServeInterface(s {wrpc}.Server, h Handler) (stop func() error, err error) { return }}"#, ); + if deferred { + deferreds.push(idx); + } idx = idx.saturating_add(1); } } + for idx in deferreds { + uwrite!( + self.src, + r#" + if d{idx}__ != nil {{ + wg__.Add(1) + go func() {{ + defer wg__.Done() + if err := d{idx}__(ctx__); err != nil {{ + errCh__ <- err + }} + }}() + }}"#, + ); + } if func.result.is_none() { // A function with no results still completes with an (empty) // result transmission; await it so the call is synchronous and @@ -3137,6 +2860,18 @@ func ServeInterface(s {wrpc}.Server, h Handler) (stop func() error, err error) { name } + fn print_deferred_ret(&mut self, deferred: bool) { + if deferred { + let context = self.deps.context(); + uwrite!(self.src, ", func({context}.Context) error"); + } + } + + fn is_async(&self, ty: &Type) -> bool { + let (paths, fut) = async_paths_ty(self.resolve, ty); + fut || !paths.is_empty() + } + fn nillable_ptr(&self, ty: &Type, result: bool, decl: bool) -> &'static str { if let Type::Id(id) = ty { match &self.resolve.types[*id].kind { @@ -3248,7 +2983,7 @@ func ServeInterface(s {wrpc}.Server, h Handler) (stop func() error, err error) { && let TypeDefKind::List(..) | TypeDefKind::Future(..) | TypeDefKind::Stream(..) = self.resolve.types[*id].kind { - // Go slices, channels and interfaces are nillable + // Go slices and channels are nillable self.print_ty(ty, true); return; } @@ -3294,43 +3029,19 @@ func ServeInterface(s {wrpc}.Server, h Handler) (stop func() error, err error) { } fn print_future(&mut self, ty: &Option) { - match ty { - Some(ty) if is_ty(self.resolve, Type::U8, ty) => { - let io = self.deps.io(); - self.push_str(io); - self.push_str(".ReadCloser"); - } - Some(ty) => { - let wrpc = self.deps.wrpc(); - self.push_str(wrpc); - self.push_str(".Receiver["); - self.print_opt_ty(ty, true); - self.push_str("]"); - } - None => { - panic!("futures with no element types are not supported") - } - } + let Some(ty) = ty else { + panic!("futures with no element types are not supported") + }; + self.push_str("<-chan "); + self.print_opt_ty(ty, true); } fn print_stream(&mut self, element: &Option) { - match element { - Some(ty) if is_ty(self.resolve, Type::U8, ty) => { - let io = self.deps.io(); - self.push_str(io); - self.push_str(".ReadCloser"); - } - Some(ty) => { - let wrpc = self.deps.wrpc(); - self.push_str(wrpc); - self.push_str(".Receiver["); - self.print_list(ty); - self.push_str("]"); - } - None => { - panic!("streams with no element types are not supported") - } - } + let Some(ty) = element else { + panic!("streams with no element types are not supported") + }; + self.push_str("<-chan "); + self.print_list(ty); } fn print_own(&mut self, id: TypeId) { @@ -3485,7 +3196,7 @@ func (v *{name}) WriteToIndex(w {wrpc}.ByteWriter) (func({wrpc}.IndexWriter) err if len(writes) > 0 {{ return func(w {wrpc}.IndexWriter) error {{ var wg {sync}.WaitGroup - var wgErr {atomic}.Value + var wgErr {atomic}.Pointer[error] for index, write := range writes {{ wg.Add(1) w, err := w.Index(index) @@ -3496,16 +3207,15 @@ func (v *{name}) WriteToIndex(w {wrpc}.ByteWriter) (func({wrpc}.IndexWriter) err go func() {{ defer wg.Done() if err := write(w); err != nil {{ - wgErr.Store(err) + wgErr.Store(&err) }} }}() }} wg.Wait() - err := wgErr.Load() - if err == nil {{ - return nil + if err := wgErr.Load(); err != nil {{ + return *err }} - return err.(error) + return nil }}, nil }} return nil, nil diff --git a/crates/wit-bindgen-test/src/go.rs b/crates/wit-bindgen-test/src/go.rs index 16a87677b..f85a20c76 100644 --- a/crates/wit-bindgen-test/src/go.rs +++ b/crates/wit-bindgen-test/src/go.rs @@ -35,14 +35,12 @@ impl LanguageMethods for Go { _args: &[String], ) -> bool { // The wRPC Go generator does not yet support bare `stream`/`future` - // types, `error-context`, or futures held by resources. + // types, `error-context`, named fixed-length lists, or `map`. matches!( name, "streams.wit" | "futures.wit" | "error-context.wit" - | "resources-with-futures.wit" - | "future-same-type-different-names.wit" | "named-fixed-length-list.wit" | "map.wit" ) diff --git a/examples/go/streams-client/app.go b/examples/go/streams-client/app.go index dfa3c0d90..dc92731dd 100644 --- a/examples/go/streams-client/app.go +++ b/examples/go/streams-client/app.go @@ -5,7 +5,6 @@ package app import ( "context" "fmt" - "io" "log/slog" "os" "sync" @@ -26,57 +25,24 @@ func init() { }))) } -type ThrottleStream[T any] struct { - tick *time.Ticker - values []T -} - -func (s *ThrottleStream[T]) Receive() ([]T, error) { - if len(s.values) == 0 { - return nil, io.EOF - } - <-s.tick.C - var v T - v, s.values = s.values[0], s.values[1:] - return []T{v}, nil -} - -func (s *ThrottleStream[T]) Close() error { - s.tick.Stop() - return nil -} - -type ThrottleReader struct { - tick *time.Ticker - buf []byte -} - -func (s *ThrottleReader) Read(p []byte) (int, error) { - if len(s.buf) == 0 { - return 0, io.EOF - } - <-s.tick.C - var b byte - b, s.buf = s.buf[0], s.buf[1:] - p[0] = b - return 1, nil -} - -func (s *ThrottleReader) Close() error { - s.tick.Stop() - return nil +func throttle[T any](values []T, d time.Duration) <-chan []T { + ch := make(chan []T) + go func() { + defer close(ch) + tick := time.NewTicker(d) + defer tick.Stop() + for _, v := range values { + <-tick.C + ch <- []T{v} + } + }() + return ch } func Run(prefix string, client wrpc.Invoker) error { numbers, bytes, errCh, err := handler.Echo(context.Background(), client, &handler.Req{ - Numbers: &ThrottleStream[uint64]{ - tick: time.NewTicker(time.Second), - values: []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, - }, - Bytes: &ThrottleReader{ - tick: time.NewTicker(time.Second), - buf: []byte("foo bar baz"), - }, + Numbers: throttle([]uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, time.Second), + Bytes: throttle([]byte("foo bar baz"), time.Second), }) if err != nil { return fmt.Errorf("failed to call `wrpc-examples:streams/handler.echo`: %w", err) @@ -87,38 +53,19 @@ func Run(prefix string, client wrpc.Invoker) error { wg.Add(1) go func() { defer wg.Done() - defer numbers.Close() - for { - chunk, err := numbers.Receive() - if err == io.EOF { - return - } - if err != nil { - slog.Error("failed to receive number chunk", "err", err) - return - } + for chunk := range numbers { fmt.Printf("%s numbers: %v\n", prefix, chunk) } }() wg.Add(1) go func() { defer wg.Done() - defer bytes.Close() - var chunk [128]byte - for { - n, err := bytes.Read(chunk[:]) - if err == io.EOF { - return - } - if err != nil { - slog.Error("failed to receive byte chunk", "err", err) - return - } - fmt.Printf("%s bytes: %s\n", prefix, chunk[:n]) + for chunk := range bytes { + fmt.Printf("%s bytes: %s\n", prefix, chunk) } }() for err := range errCh { - slog.Error("failed to transmit async parameters", "err", err) + slog.Error("failed to exchange async values", "err", err) } return nil } diff --git a/examples/go/streams-client/bindings/wrpc_examples/streams/handler/bindings.wrpc.go b/examples/go/streams-client/bindings/wrpc_examples/streams/handler/bindings.wrpc.go index 881eb6496..fb859f1c8 100644 --- a/examples/go/streams-client/bindings/wrpc_examples/streams/handler/bindings.wrpc.go +++ b/examples/go/streams-client/bindings/wrpc_examples/streams/handler/bindings.wrpc.go @@ -9,15 +9,14 @@ import ( fmt "fmt" io "io" slog "log/slog" - math "math" sync "sync" atomic "sync/atomic" wrpc "wrpc.io/go" ) type Req struct { - Numbers wrpc.Receiver[[]uint64] - Bytes io.ReadCloser + Numbers <-chan []uint64 + Bytes <-chan []uint8 } func (v *Req) String() string { return "Req" } @@ -25,93 +24,15 @@ func (v *Req) String() string { return "Req" } func (v *Req) WriteToIndex(w wrpc.ByteWriter) (func(wrpc.IndexWriter) error, error) { writes := make(map[uint32]func(wrpc.IndexWriter) error, 2) slog.Debug("writing field", "name", "numbers") - write0, err := func(v wrpc.Receiver[[]uint64], w interface { - io.ByteWriter - io.Writer - }) (write func(wrpc.IndexWriter) error, err error) { - slog.Debug("writing stream `stream::pending` status byte") - if err := w.WriteByte(0); err != nil { - return nil, fmt.Errorf("failed to write `stream::pending` byte: %w", err) - } - return func(w wrpc.IndexWriter) (err error) { - defer func() { - slog.Debug("closing outgoing pending stream") - if cErr := v.Close(); cErr != nil { - if err == nil { - err = fmt.Errorf("failed to close outgoing pending stream: %w", cErr) - } else { - slog.Warn("failed to close outgoing pending stream", "err", cErr) - } - } - }() - var wg sync.WaitGroup - var wgErr atomic.Value - var total uint32 - for { - var end bool - slog.Debug("receiving outgoing pending stream contents") - chunk, err := v.Receive() - n := len(chunk) - if err == io.EOF { - end = true - slog.Debug("outgoing pending stream reached EOF") - } else if err != nil { - return fmt.Errorf("failed to receive outgoing pending stream chunk: %w", err) - } - if n > math.MaxUint32 { - return fmt.Errorf("outgoing pending stream chunk length of %d overflows a 32-bit integer", n) - } - if math.MaxUint32-uint32(n) < total { - return errors.New("total outgoing pending stream element count would overflow a 32-bit unsigned integer") - } - if n > 0 { - slog.Debug("writing pending stream chunk length", "len", n) - _, err = wrpc.WriteUint32(uint32(n), w) - if err != nil { - return fmt.Errorf("failed to write pending stream chunk length of %d: %w", n, err) - } - for _, v := range chunk { - slog.Debug("writing pending stream element", "i", total) - write, err := (func(wrpc.IndexWriter) error)(nil), func(v uint64, w io.Writer) (err error) { - b := make([]byte, binary.MaxVarintLen64) - i := binary.PutUvarint(b, uint64(v)) - slog.Debug("writing u64") - _, err = w.Write(b[:i]) - return err - }(v, w) - if err != nil { - return fmt.Errorf("failed to write pending stream chunk element %d: %w", total, err) - } - if write != nil { - wg.Add(1) - w, err := w.Index(total) - if err != nil { - return fmt.Errorf("failed to index nested stream writer: %w", err) - } - go func() { - defer wg.Done() - if err := write(w); err != nil { - wgErr.Store(err) - } - }() - } - total++ - } - } - if end { - if err := w.WriteByte(0); err != nil { - return fmt.Errorf("failed to write pending stream end byte: %w", err) - } - wg.Wait() - err := wgErr.Load() - if err == nil { - return nil - } - return err.(error) - } - } - }, nil - }(v.Numbers, w) + write0, err := wrpc.WriteStream(v.Numbers, w, func(v uint64, w wrpc.ByteWriter) (func(wrpc.IndexWriter) error, error) { + return (func(wrpc.IndexWriter) error)(nil), func(v uint64, w io.Writer) (err error) { + b := make([]byte, binary.MaxVarintLen64) + i := binary.PutUvarint(b, uint64(v)) + slog.Debug("writing u64") + _, err = w.Write(b[:i]) + return err + }(v, w) + }) if err != nil { return nil, fmt.Errorf("failed to write `numbers` field: %w", err) } @@ -119,59 +40,7 @@ func (v *Req) WriteToIndex(w wrpc.ByteWriter) (func(wrpc.IndexWriter) error, err writes[0] = write0 } slog.Debug("writing field", "name", "bytes") - write1, err := func(v io.ReadCloser, w interface { - io.ByteWriter - io.Writer - }) (write func(wrpc.IndexWriter) error, err error) { - slog.Debug("writing byte stream `stream::pending` status byte") - if err = w.WriteByte(0); err != nil { - return nil, fmt.Errorf("failed to write `stream::pending` byte: %w", err) - } - return func(w wrpc.IndexWriter) (err error) { - defer func() { - slog.Debug("closing byte list stream writer") - if cErr := v.Close(); cErr != nil { - if err == nil { - err = fmt.Errorf("failed to close pending byte stream: %w", cErr) - } else { - slog.Warn("failed to close pending byte stream", "err", cErr) - } - } - }() - chunk := make([]byte, wrpc.ReadChunkSize) - for { - var end bool - slog.Debug("reading pending byte stream contents") - n, err := v.Read(chunk) - if err == io.EOF { - end = true - slog.Debug("pending byte stream reached EOF") - } else if err != nil { - return fmt.Errorf("failed to read pending byte stream chunk: %w", err) - } - if n > math.MaxUint32 { - return fmt.Errorf("pending byte stream chunk length of %d overflows a 32-bit integer", n) - } - if n > 0 { - slog.Debug("writing pending byte stream chunk length", "len", n) - _, err = wrpc.WriteUint32(uint32(n), w) - if err != nil { - return fmt.Errorf("failed to write pending byte stream chunk length of %d: %w", n, err) - } - _, err = w.Write(chunk[:n]) - if err != nil { - return fmt.Errorf("failed to write pending byte stream chunk contents: %w", err) - } - } - if end { - if err := w.WriteByte(0); err != nil { - return fmt.Errorf("failed to write pending byte stream end byte: %w", err) - } - return nil - } - } - }, nil - }(v.Bytes, w) + write1, err := wrpc.WriteByteStream(v.Bytes, w) if err != nil { return nil, fmt.Errorf("failed to write `bytes` field: %w", err) } @@ -182,7 +51,7 @@ func (v *Req) WriteToIndex(w wrpc.ByteWriter) (func(wrpc.IndexWriter) error, err if len(writes) > 0 { return func(w wrpc.IndexWriter) error { var wg sync.WaitGroup - var wgErr atomic.Value + var wgErr atomic.Pointer[error] for index, write := range writes { wg.Add(1) w, err := w.Index(index) @@ -193,21 +62,20 @@ func (v *Req) WriteToIndex(w wrpc.ByteWriter) (func(wrpc.IndexWriter) error, err go func() { defer wg.Done() if err := write(w); err != nil { - wgErr.Store(err) + wgErr.Store(&err) } }() } wg.Wait() - err := wgErr.Load() - if err == nil { - return nil + if err := wgErr.Load(); err != nil { + return *err } - return err.(error) + return nil }, nil } return nil, nil } -func Echo(ctx__ context.Context, wrpc__ wrpc.Invoker, r *Req) (r0__ wrpc.Receiver[[]uint64], r1__ io.ReadCloser, writeErrs__ <-chan error, err__ error) { +func Echo(ctx__ context.Context, wrpc__ wrpc.Invoker, r *Req) (r0__ <-chan []uint64, r1__ <-chan []uint8, errs__ <-chan error, err__ error) { var buf__ bytes.Buffer var writeCount__ uint32 write0__, err__ := (r).WriteToIndex(&buf__) @@ -236,12 +104,17 @@ func Echo(ctx__ context.Context, wrpc__ wrpc.Invoker, r *Req) (r0__ wrpc.Receive slog.ErrorContext(ctx__, "failed to close reader", "instance", "wrpc-examples:streams/handler", "name", "echo", "err", err) } }() + errCh__ := make(chan error, uint(writeCount__)+2) + errs__ = errCh__ + var wg__ sync.WaitGroup + defer func() { + go func() { + wg__.Wait() + close(errCh__) + }() + }() if writeCount__ > 0 { - writeErrCh__ := make(chan error, uint(writeCount__)) - writeErrs__ = writeErrCh__ - var wg__ sync.WaitGroup for index, write := range writes__ { - wg__.Add(1) w, err := w__.Index(index) if err != nil { if cErr := w__.Close(); cErr != nil { @@ -250,246 +123,73 @@ func Echo(ctx__ context.Context, wrpc__ wrpc.Invoker, r *Req) (r0__ wrpc.Receive err__ = fmt.Errorf("failed to index param writer at index `%v`: %w", index, err) return } + wg__.Add(1) write := write go func() { defer wg__.Done() if err := write(w); err != nil { - writeErrCh__ <- err + errCh__ <- err } }() } - go func() { - wg__.Wait() - close(writeErrCh__) - }() } if cErr__ := w__.Close(); cErr__ != nil { slog.DebugContext(ctx__, "failed to close outgoing stream", "instance", "wrpc-examples:streams/handler", "name", "echo", "err", cErr__) } - r0__, err__ = func(r wrpc.IndexReadCloser, path ...uint32) (wrpc.Receiver[[]uint64], error) { - slog.Debug("reading stream status byte") - status, err := r.ReadByte() - if err != nil { - return nil, fmt.Errorf("failed to read stream status byte: %w", err) - } - switch status { - case 0: - if len(path) > 0 { - var err error - r, err = r.Index(path...) + var d0__ func(context.Context) error + r0__, d0__, err__ = wrpc.ReadStream(r__, func(r wrpc.IndexReadCloser, path ...uint32) (uint64, func(context.Context) error, error) { + v, err := func(r io.ByteReader) (uint64, error) { + var x uint64 + var s uint8 + for i := 0; i < 10; i++ { + slog.Debug("reading u64 byte", "i", i) + b, err := r.ReadByte() if err != nil { - return nil, fmt.Errorf("failed to index nested stream reader: %w", err) - } - } - var total uint32 - return wrpc.NewDecodeReceiver(r, func(r wrpc.IndexReadCloser) ([]uint64, error) { - slog.Debug("reading pending stream chunk length") - n, err := func(r io.ByteReader) (uint32, error) { - var x uint32 - var s uint8 - for i := 0; i < 5; i++ { - slog.Debug("reading u32 byte", "i", i) - b, err := r.ReadByte() - if err != nil { - if i > 0 && err == io.EOF { - err = io.ErrUnexpectedEOF - } - return x, fmt.Errorf("failed to read u32 byte: %w", err) - } - if s == 28 && b > 0x0f { - return x, errors.New("varint overflows a 32-bit integer") - } - if b < 0x80 { - return x | uint32(b)< 0 && err == io.EOF { + err = io.ErrUnexpectedEOF } - return x, errors.New("varint overflows a 32-bit integer") - }(r) - if err != nil { - return nil, fmt.Errorf("failed to read pending stream chunk length: %w", err) - } - if n == 0 { - return nil, io.EOF + return x, fmt.Errorf("failed to read u64 byte: %w", err) } - if math.MaxUint32-n < total { - return nil, errors.New("total incoming pending stream element count would overflow a 32-bit unsigned integer") + if s == 63 && b > 0x01 { + return x, errors.New("varint overflows a 64-bit integer") } - vs := wrpc.NewSlice[[]uint64](n) - for i := range n { - slog.Debug("reading pending stream element", "i", total) - v, err := func(r io.ByteReader) (uint64, error) { - var x uint64 - var s uint8 - for i := 0; i < 10; i++ { - slog.Debug("reading u64 byte", "i", i) - b, err := r.ReadByte() - if err != nil { - if i > 0 && err == io.EOF { - err = io.ErrUnexpectedEOF - } - return x, fmt.Errorf("failed to read u64 byte: %w", err) - } - if s == 63 && b > 0x01 { - return x, errors.New("varint overflows a 64-bit integer") - } - if b < 0x80 { - return x | uint64(b)< 0 && err == io.EOF { - err = io.ErrUnexpectedEOF - } - return nil, fmt.Errorf("failed to read list length byte: %w", err) - } - if s == 28 && b > 0x0f { - return nil, errors.New("list length overflows a 32-bit integer") - } - if b < 0x80 { - x = x | uint32(b)< 0 && err == io.EOF { - err = io.ErrUnexpectedEOF - } - return x, fmt.Errorf("failed to read u64 byte: %w", err) - } - if s == 63 && b > 0x01 { - return x, errors.New("varint overflows a 64-bit integer") - } - if b < 0x80 { - return x | uint64(b)< 0 { - var err error - r, err = r.Index(path...) - if err != nil { - return nil, fmt.Errorf("failed to index nested byte stream reader: %w", err) - } - } - return wrpc.NewByteStreamReader(r), nil - case 1: - slog.Debug("reading ready byte stream contents") - buf, err := - func(r interface { - io.ByteReader - io.Reader - }) ([]byte, error) { - var x uint32 - var s uint - for i := 0; i < 5; i++ { - slog.Debug("reading byte list length", "i", i) - b, err := r.ReadByte() - if err != nil { - if i > 0 && err == io.EOF { - err = io.ErrUnexpectedEOF - } - return nil, fmt.Errorf("failed to read byte list length byte: %w", err) - } - if s == 28 && b > 0x0f { - return nil, errors.New("byte list length overflows a 32-bit integer") - } - if b < 0x80 { - x = x | uint32(b)< math.MaxUint32 { - return fmt.Errorf("outgoing pending stream chunk length of %d overflows a 32-bit integer", n) - } - if math.MaxUint32-uint32(n) < total { - return errors.New("total outgoing pending stream element count would overflow a 32-bit unsigned integer") - } - if n > 0 { - slog.Debug("writing pending stream chunk length", "len", n) - _, err = wrpc.WriteUint32(uint32(n), w) - if err != nil { - return fmt.Errorf("failed to write pending stream chunk length of %d: %w", n, err) - } - for _, v := range chunk { - slog.Debug("writing pending stream element", "i", total) - write, err := (func(wrpc.IndexWriter) error)(nil), func(v uint64, w io.Writer) (err error) { - b := make([]byte, binary.MaxVarintLen64) - i := binary.PutUvarint(b, uint64(v)) - slog.Debug("writing u64") - _, err = w.Write(b[:i]) - return err - }(v, w) - if err != nil { - return fmt.Errorf("failed to write pending stream chunk element %d: %w", total, err) - } - if write != nil { - wg.Add(1) - w, err := w.Index(total) - if err != nil { - return fmt.Errorf("failed to index nested stream writer: %w", err) - } - go func() { - defer wg.Done() - if err := write(w); err != nil { - wgErr.Store(err) - } - }() - } - total++ - } - } - if end { - if err := w.WriteByte(0); err != nil { - return fmt.Errorf("failed to write pending stream end byte: %w", err) - } - wg.Wait() - err := wgErr.Load() - if err == nil { - return nil - } - return err.(error) - } - } - }, nil - }(v.Numbers, w) + write0, err := wrpc.WriteStream(v.Numbers, w, func(v uint64, w wrpc.ByteWriter) (func(wrpc.IndexWriter) error, error) { + return (func(wrpc.IndexWriter) error)(nil), func(v uint64, w io.Writer) (err error) { + b := make([]byte, binary.MaxVarintLen64) + i := binary.PutUvarint(b, uint64(v)) + slog.Debug("writing u64") + _, err = w.Write(b[:i]) + return err + }(v, w) + }) if err != nil { return nil, fmt.Errorf("failed to write `numbers` field: %w", err) } @@ -119,59 +40,7 @@ func (v *Req) WriteToIndex(w wrpc.ByteWriter) (func(wrpc.IndexWriter) error, err writes[0] = write0 } slog.Debug("writing field", "name", "bytes") - write1, err := func(v io.ReadCloser, w interface { - io.ByteWriter - io.Writer - }) (write func(wrpc.IndexWriter) error, err error) { - slog.Debug("writing byte stream `stream::pending` status byte") - if err = w.WriteByte(0); err != nil { - return nil, fmt.Errorf("failed to write `stream::pending` byte: %w", err) - } - return func(w wrpc.IndexWriter) (err error) { - defer func() { - slog.Debug("closing byte list stream writer") - if cErr := v.Close(); cErr != nil { - if err == nil { - err = fmt.Errorf("failed to close pending byte stream: %w", cErr) - } else { - slog.Warn("failed to close pending byte stream", "err", cErr) - } - } - }() - chunk := make([]byte, wrpc.ReadChunkSize) - for { - var end bool - slog.Debug("reading pending byte stream contents") - n, err := v.Read(chunk) - if err == io.EOF { - end = true - slog.Debug("pending byte stream reached EOF") - } else if err != nil { - return fmt.Errorf("failed to read pending byte stream chunk: %w", err) - } - if n > math.MaxUint32 { - return fmt.Errorf("pending byte stream chunk length of %d overflows a 32-bit integer", n) - } - if n > 0 { - slog.Debug("writing pending byte stream chunk length", "len", n) - _, err = wrpc.WriteUint32(uint32(n), w) - if err != nil { - return fmt.Errorf("failed to write pending byte stream chunk length of %d: %w", n, err) - } - _, err = w.Write(chunk[:n]) - if err != nil { - return fmt.Errorf("failed to write pending byte stream chunk contents: %w", err) - } - } - if end { - if err := w.WriteByte(0); err != nil { - return fmt.Errorf("failed to write pending byte stream end byte: %w", err) - } - return nil - } - } - }, nil - }(v.Bytes, w) + write1, err := wrpc.WriteByteStream(v.Bytes, w) if err != nil { return nil, fmt.Errorf("failed to write `bytes` field: %w", err) } @@ -182,7 +51,7 @@ func (v *Req) WriteToIndex(w wrpc.ByteWriter) (func(wrpc.IndexWriter) error, err if len(writes) > 0 { return func(w wrpc.IndexWriter) error { var wg sync.WaitGroup - var wgErr atomic.Value + var wgErr atomic.Pointer[error] for index, write := range writes { wg.Add(1) w, err := w.Index(index) @@ -193,23 +62,22 @@ func (v *Req) WriteToIndex(w wrpc.ByteWriter) (func(wrpc.IndexWriter) error, err go func() { defer wg.Done() if err := write(w); err != nil { - wgErr.Store(err) + wgErr.Store(&err) } }() } wg.Wait() - err := wgErr.Load() - if err == nil { - return nil + if err := wgErr.Load(); err != nil { + return *err } - return err.(error) + return nil }, nil } return nil, nil } type Handler interface { - Echo(ctx__ context.Context, r *Req) (wrpc.Receiver[[]uint64], io.ReadCloser, error) + Echo(ctx__ context.Context, r *Req) (<-chan []uint64, <-chan []uint8, error) } func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { @@ -229,236 +97,58 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { slog.DebugContext(ctx, "failed to close writer", "instance", "wrpc-examples:streams/handler", "name", "echo", "err", err) } }() + deferredCtx, cancel := context.WithCancel(ctx) + var wg sync.WaitGroup + defer func() { + go func() { + wg.Wait() + cancel() + }() + }() slog.DebugContext(ctx, "reading parameter", "i", 0) - p0, err := func(r wrpc.IndexReadCloser, path ...uint32) (*Req, error) { + p0, d0, err := func(r wrpc.IndexReadCloser, path ...uint32) (*Req, func(context.Context) error, error) { v := &Req{} var err error + var d func(context.Context) error + var ds []func(context.Context) error slog.Debug("reading field", "name", "numbers") - v.Numbers, err = func(r wrpc.IndexReadCloser, path ...uint32) (wrpc.Receiver[[]uint64], error) { - slog.Debug("reading stream status byte") - status, err := r.ReadByte() - if err != nil { - return nil, fmt.Errorf("failed to read stream status byte: %w", err) - } - switch status { - case 0: - if len(path) > 0 { - var err error - r, err = r.Index(path...) + v.Numbers, d, err = wrpc.ReadStream(r, func(r wrpc.IndexReadCloser, path ...uint32) (uint64, func(context.Context) error, error) { + v, err := func(r io.ByteReader) (uint64, error) { + var x uint64 + var s uint8 + for i := 0; i < 10; i++ { + slog.Debug("reading u64 byte", "i", i) + b, err := r.ReadByte() if err != nil { - return nil, fmt.Errorf("failed to index nested stream reader: %w", err) - } - } - var total uint32 - return wrpc.NewDecodeReceiver(r, func(r wrpc.IndexReadCloser) ([]uint64, error) { - slog.Debug("reading pending stream chunk length") - n, err := func(r io.ByteReader) (uint32, error) { - var x uint32 - var s uint8 - for i := 0; i < 5; i++ { - slog.Debug("reading u32 byte", "i", i) - b, err := r.ReadByte() - if err != nil { - if i > 0 && err == io.EOF { - err = io.ErrUnexpectedEOF - } - return x, fmt.Errorf("failed to read u32 byte: %w", err) - } - if s == 28 && b > 0x0f { - return x, errors.New("varint overflows a 32-bit integer") - } - if b < 0x80 { - return x | uint32(b)< 0 && err == io.EOF { + err = io.ErrUnexpectedEOF } - return x, errors.New("varint overflows a 32-bit integer") - }(r) - if err != nil { - return nil, fmt.Errorf("failed to read pending stream chunk length: %w", err) - } - if n == 0 { - return nil, io.EOF + return x, fmt.Errorf("failed to read u64 byte: %w", err) } - if math.MaxUint32-n < total { - return nil, errors.New("total incoming pending stream element count would overflow a 32-bit unsigned integer") + if s == 63 && b > 0x01 { + return x, errors.New("varint overflows a 64-bit integer") } - vs := wrpc.NewSlice[[]uint64](n) - for i := range n { - slog.Debug("reading pending stream element", "i", total) - v, err := func(r io.ByteReader) (uint64, error) { - var x uint64 - var s uint8 - for i := 0; i < 10; i++ { - slog.Debug("reading u64 byte", "i", i) - b, err := r.ReadByte() - if err != nil { - if i > 0 && err == io.EOF { - err = io.ErrUnexpectedEOF - } - return x, fmt.Errorf("failed to read u64 byte: %w", err) - } - if s == 63 && b > 0x01 { - return x, errors.New("varint overflows a 64-bit integer") - } - if b < 0x80 { - return x | uint64(b)< 0 && err == io.EOF { - err = io.ErrUnexpectedEOF - } - return nil, fmt.Errorf("failed to read list length byte: %w", err) - } - if s == 28 && b > 0x0f { - return nil, errors.New("list length overflows a 32-bit integer") - } - if b < 0x80 { - x = x | uint32(b)< 0 && err == io.EOF { - err = io.ErrUnexpectedEOF - } - return x, fmt.Errorf("failed to read u64 byte: %w", err) - } - if s == 63 && b > 0x01 { - return x, errors.New("varint overflows a 64-bit integer") - } - if b < 0x80 { - return x | uint64(b)< 0 { - var err error - r, err = r.Index(path...) - if err != nil { - return nil, fmt.Errorf("failed to index nested byte stream reader: %w", err) - } - } - return wrpc.NewByteStreamReader(r), nil - case 1: - slog.Debug("reading ready byte stream contents") - buf, err := - func(r interface { - io.ByteReader - io.Reader - }) ([]byte, error) { - var x uint32 - var s uint - for i := 0; i < 5; i++ { - slog.Debug("reading byte list length", "i", i) - b, err := r.ReadByte() - if err != nil { - if i > 0 && err == io.EOF { - err = io.ErrUnexpectedEOF - } - return nil, fmt.Errorf("failed to read byte list length byte: %w", err) - } - if s == 28 && b > 0x0f { - return nil, errors.New("byte list length overflows a 32-bit integer") - } - if b < 0x80 { - x = x | uint32(b)< math.MaxUint32 { - return fmt.Errorf("outgoing pending stream chunk length of %d overflows a 32-bit integer", n) - } - if math.MaxUint32-uint32(n) < total { - return errors.New("total outgoing pending stream element count would overflow a 32-bit unsigned integer") - } - if n > 0 { - slog.Debug("writing pending stream chunk length", "len", n) - _, err = wrpc.WriteUint32(uint32(n), w) - if err != nil { - return fmt.Errorf("failed to write pending stream chunk length of %d: %w", n, err) - } - for _, v := range chunk { - slog.Debug("writing pending stream element", "i", total) - write, err := (func(wrpc.IndexWriter) error)(nil), func(v uint64, w io.Writer) (err error) { - b := make([]byte, binary.MaxVarintLen64) - i := binary.PutUvarint(b, uint64(v)) - slog.Debug("writing u64") - _, err = w.Write(b[:i]) - return err - }(v, w) - if err != nil { - return fmt.Errorf("failed to write pending stream chunk element %d: %w", total, err) - } - if write != nil { - wg.Add(1) - w, err := w.Index(total) - if err != nil { - return fmt.Errorf("failed to index nested stream writer: %w", err) - } - go func() { - defer wg.Done() - if err := write(w); err != nil { - wgErr.Store(err) - } - }() - } - total++ - } - } - if end { - if err := w.WriteByte(0); err != nil { - return fmt.Errorf("failed to write pending stream end byte: %w", err) - } - wg.Wait() - err := wgErr.Load() - if err == nil { - return nil - } - return err.(error) - } - } - }, nil - }(r0, &buf) + write0, err := wrpc.WriteStream(r0, &buf, func(v uint64, w wrpc.ByteWriter) (func(wrpc.IndexWriter) error, error) { + return (func(wrpc.IndexWriter) error)(nil), func(v uint64, w io.Writer) (err error) { + b := make([]byte, binary.MaxVarintLen64) + i := binary.PutUvarint(b, uint64(v)) + slog.Debug("writing u64") + _, err = w.Write(b[:i]) + return err + }(v, w) + }) if err != nil { slog.WarnContext(ctx, "failed to write result value", "i", 0, "instance", "wrpc-examples:streams/handler", "name", "echo", "err", err) return @@ -575,59 +194,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { if write0 != nil { writes[0] = write0 } - write1, err := func(v io.ReadCloser, w interface { - io.ByteWriter - io.Writer - }) (write func(wrpc.IndexWriter) error, err error) { - slog.Debug("writing byte stream `stream::pending` status byte") - if err = w.WriteByte(0); err != nil { - return nil, fmt.Errorf("failed to write `stream::pending` byte: %w", err) - } - return func(w wrpc.IndexWriter) (err error) { - defer func() { - slog.Debug("closing byte list stream writer") - if cErr := v.Close(); cErr != nil { - if err == nil { - err = fmt.Errorf("failed to close pending byte stream: %w", cErr) - } else { - slog.Warn("failed to close pending byte stream", "err", cErr) - } - } - }() - chunk := make([]byte, wrpc.ReadChunkSize) - for { - var end bool - slog.Debug("reading pending byte stream contents") - n, err := v.Read(chunk) - if err == io.EOF { - end = true - slog.Debug("pending byte stream reached EOF") - } else if err != nil { - return fmt.Errorf("failed to read pending byte stream chunk: %w", err) - } - if n > math.MaxUint32 { - return fmt.Errorf("pending byte stream chunk length of %d overflows a 32-bit integer", n) - } - if n > 0 { - slog.Debug("writing pending byte stream chunk length", "len", n) - _, err = wrpc.WriteUint32(uint32(n), w) - if err != nil { - return fmt.Errorf("failed to write pending byte stream chunk length of %d: %w", n, err) - } - _, err = w.Write(chunk[:n]) - if err != nil { - return fmt.Errorf("failed to write pending byte stream chunk contents: %w", err) - } - } - if end { - if err := w.WriteByte(0); err != nil { - return fmt.Errorf("failed to write pending byte stream end byte: %w", err) - } - return nil - } - } - }, nil - }(r1, &buf) + write1, err := wrpc.WriteByteStream(r1, &buf) if err != nil { slog.WarnContext(ctx, "failed to write result value", "i", 1, "instance", "wrpc-examples:streams/handler", "name", "echo", "err", err) return @@ -652,7 +219,9 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { return } write := write + wg.Add(1) go func() { + defer wg.Done() if err := write(w); err != nil { slog.WarnContext(ctx, "failed to write nested result value", "instance", "wrpc-examples:streams/handler", "name", "echo", "err", err) } @@ -664,7 +233,9 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { return } write := write + wg.Add(1) go func() { + defer wg.Done() if err := write(w); err != nil { slog.WarnContext(ctx, "failed to write nested result value", "instance", "wrpc-examples:streams/handler", "name", "echo", "err", err) } diff --git a/examples/go/wasi-keyvalue-client/bindings/wasi/keyvalue/store/bindings.wrpc.go b/examples/go/wasi-keyvalue-client/bindings/wasi/keyvalue/store/bindings.wrpc.go index c69933f19..3cd91ae32 100644 --- a/examples/go/wasi-keyvalue-client/bindings/wasi/keyvalue/store/bindings.wrpc.go +++ b/examples/go/wasi-keyvalue-client/bindings/wasi/keyvalue/store/bindings.wrpc.go @@ -228,7 +228,7 @@ func (v *KeyResponse) WriteToIndex(w wrpc.ByteWriter) (func(wrpc.IndexWriter) er if len(writes) > 0 { return func(w wrpc.IndexWriter) error { var wg sync.WaitGroup - var wgErr atomic.Value + var wgErr atomic.Pointer[error] for index, write := range writes { wg.Add(1) w, err := w.Index(index) @@ -239,16 +239,15 @@ func (v *KeyResponse) WriteToIndex(w wrpc.ByteWriter) (func(wrpc.IndexWriter) er go func() { defer wg.Done() if err := write(w); err != nil { - wgErr.Store(err) + wgErr.Store(&err) } }() } wg.Wait() - err := wgErr.Load() - if err == nil { - return nil + if err := wgErr.Load(); err != nil { + return *err } - return err.(error) + return nil }, nil } return nil, nil @@ -312,7 +311,7 @@ func (v *KeyResponse) WriteToIndex(w wrpc.ByteWriter) (func(wrpc.IndexWriter) er if len(writes) > 0 { return func(w wrpc.IndexWriter) error { var wg sync.WaitGroup - var wgErr atomic.Value + var wgErr atomic.Pointer[error] for index, write := range writes { wg.Add(1) w, err := w.Index(index) @@ -323,16 +322,15 @@ func (v *KeyResponse) WriteToIndex(w wrpc.ByteWriter) (func(wrpc.IndexWriter) er go func() { defer wg.Done() if err := write(w); err != nil { - wgErr.Store(err) + wgErr.Store(&err) } }() } wg.Wait() - err := wgErr.Load() - if err == nil { - return nil + if err := wgErr.Load(); err != nil { + return *err } - return err.(error) + return nil }, nil } return nil, nil @@ -886,7 +884,7 @@ func Bucket_Set(ctx__ context.Context, wrpc__ wrpc.Invoker, self wrpc.Borrow[Buc if len(writes) > 0 { return func(w wrpc.IndexWriter) error { var wg sync.WaitGroup - var wgErr atomic.Value + var wgErr atomic.Pointer[error] for index, write := range writes { wg.Add(1) w, err := w.Index(index) @@ -897,16 +895,15 @@ func Bucket_Set(ctx__ context.Context, wrpc__ wrpc.Invoker, self wrpc.Borrow[Buc go func() { defer wg.Done() if err := write(w); err != nil { - wgErr.Store(err) + wgErr.Store(&err) } }() } wg.Wait() - err := wgErr.Load() - if err == nil { - return nil + if err := wgErr.Load(); err != nil { + return *err } - return err.(error) + return nil }, nil } return nil, nil diff --git a/examples/go/wasi-keyvalue-server/bindings/exports/wasi/keyvalue/store/bindings.wrpc.go b/examples/go/wasi-keyvalue-server/bindings/exports/wasi/keyvalue/store/bindings.wrpc.go index 75b45c930..c21689acc 100644 --- a/examples/go/wasi-keyvalue-server/bindings/exports/wasi/keyvalue/store/bindings.wrpc.go +++ b/examples/go/wasi-keyvalue-server/bindings/exports/wasi/keyvalue/store/bindings.wrpc.go @@ -228,7 +228,7 @@ func (v *KeyResponse) WriteToIndex(w wrpc.ByteWriter) (func(wrpc.IndexWriter) er if len(writes) > 0 { return func(w wrpc.IndexWriter) error { var wg sync.WaitGroup - var wgErr atomic.Value + var wgErr atomic.Pointer[error] for index, write := range writes { wg.Add(1) w, err := w.Index(index) @@ -239,16 +239,15 @@ func (v *KeyResponse) WriteToIndex(w wrpc.ByteWriter) (func(wrpc.IndexWriter) er go func() { defer wg.Done() if err := write(w); err != nil { - wgErr.Store(err) + wgErr.Store(&err) } }() } wg.Wait() - err := wgErr.Load() - if err == nil { - return nil + if err := wgErr.Load(); err != nil { + return *err } - return err.(error) + return nil }, nil } return nil, nil @@ -312,7 +311,7 @@ func (v *KeyResponse) WriteToIndex(w wrpc.ByteWriter) (func(wrpc.IndexWriter) er if len(writes) > 0 { return func(w wrpc.IndexWriter) error { var wg sync.WaitGroup - var wgErr atomic.Value + var wgErr atomic.Pointer[error] for index, write := range writes { wg.Add(1) w, err := w.Index(index) @@ -323,16 +322,15 @@ func (v *KeyResponse) WriteToIndex(w wrpc.ByteWriter) (func(wrpc.IndexWriter) er go func() { defer wg.Done() if err := write(w); err != nil { - wgErr.Store(err) + wgErr.Store(&err) } }() } wg.Wait() - err := wgErr.Load() - if err == nil { - return nil + if err := wgErr.Load(); err != nil { + return *err } - return err.(error) + return nil }, nil } return nil, nil @@ -756,7 +754,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { if len(writes) > 0 { return func(w wrpc.IndexWriter) error { var wg sync.WaitGroup - var wgErr atomic.Value + var wgErr atomic.Pointer[error] for index, write := range writes { wg.Add(1) w, err := w.Index(index) @@ -767,16 +765,15 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { go func() { defer wg.Done() if err := write(w); err != nil { - wgErr.Store(err) + wgErr.Store(&err) } }() } wg.Wait() - err := wgErr.Load() - if err == nil { - return nil + if err := wgErr.Load(); err != nil { + return *err } - return err.(error) + return nil }, nil } return nil, nil diff --git a/go/frame_reader.go b/go/frame_reader.go index a691095eb..30273ebff 100644 --- a/go/frame_reader.go +++ b/go/frame_reader.go @@ -207,6 +207,10 @@ func (r *FrameStreamReader) ReadByte() (byte, error) { } } +func (r *FrameStreamReader) Buffered() int { + return len(r.buf) +} + func (r *FrameStreamReader) Index(path ...uint32) (IndexReadCloser, error) { p := r.path + indexPath(path...) readers := r.readers.Add(1) diff --git a/go/future.go b/go/future.go index 40d35f37c..5fd9aa029 100644 --- a/go/future.go +++ b/go/future.go @@ -1,6 +1,9 @@ package wrpc import ( + "bytes" + "context" + "errors" "fmt" "log/slog" ) @@ -23,27 +26,82 @@ func ReadFutureStatus(r ByteReader) (bool, error) { } } -// ReadFuture reads a future from `r` -func ReadFuture[T any](r IndexReadCloser, f func(IndexReadCloser) (T, error), path ...uint32) (Receiver[T], error) { +// ReadFuture reads a `future` from `r` using `f` to read its value. +// The returned channel receives the value once the future resolves and is closed afterwards. +// If the future is pending, the returned function reads it and must be called. If reading fails, +// the channel never receives a value and is never closed. +func ReadFuture[T any](r IndexReadCloser, f func(IndexReadCloser, ...uint32) (T, func(context.Context) error, error), path ...uint32) (<-chan T, func(context.Context) error, error) { slog.Debug("reading future status byte") ok, err := ReadFutureStatus(r) if err != nil { - return nil, err + return nil, nil, err } - if !ok { + ch := make(chan T, 1) + if ok { + slog.Debug("reading ready future") + v, deferred, err := f(r, path...) + if err != nil { + return nil, nil, fmt.Errorf("failed to read ready future: %w", err) + } + ch <- v + close(ch) + return ch, deferred, nil + } + if len(path) > 0 { + slog.Debug("indexing pending future reader") + r, err = r.Index(path...) + if err != nil { + return nil, nil, fmt.Errorf("failed to index nested future reader: %w", err) + } + } + return ch, func(ctx context.Context) error { if len(path) > 0 { - slog.Debug("indexing pending future reader") - r, err = r.Index(path...) - if err != nil { - return nil, fmt.Errorf("failed to index nested future reader: %w", err) - } + defer closeNested(r) + } + slog.Debug("reading pending future") + v, deferred, err := f(r) + if err != nil { + return fmt.Errorf("failed to read pending future: %w", err) + } + ch <- v + close(ch) + if deferred != nil { + return deferred(ctx) + } + if r.Buffered() > 0 { + return errors.New("received data after pending future value end") } - return NewDecodeReceiver(r, f), nil + return nil + }, nil +} + +// WriteFuture writes a `future` received from `v` to `w` using `f` to write its value +func WriteFuture[T any](v <-chan T, w ByteWriter, f func(T, ByteWriter) (func(IndexWriter) error, error)) (func(IndexWriter) error, error) { + if v == nil { + return nil, errors.New("outgoing future channel is nil") } - slog.Debug("reading ready future") - v, err := f(r) - if err != nil { - return nil, err + slog.Debug("writing `future::pending` status byte") + if err := w.WriteByte(0); err != nil { + return nil, fmt.Errorf("failed to write `future::pending` byte: %w", err) } - return NewCompleteReceiver(v), nil + return func(w IndexWriter) error { + slog.Debug("receiving outgoing pending future") + x, ok := <-v + if !ok { + return errors.New("outgoing pending future closed without a value") + } + slog.Debug("writing pending future value") + var buf bytes.Buffer + write, err := f(x, &buf) + if err != nil { + return fmt.Errorf("failed to encode pending future value: %w", err) + } + if _, err := w.Write(buf.Bytes()); err != nil { + return fmt.Errorf("failed to write pending future value: %w", err) + } + if write != nil { + return write(w) + } + return nil + }, nil } diff --git a/go/stream.go b/go/stream.go index a880cb950..c466376d2 100644 --- a/go/stream.go +++ b/go/stream.go @@ -1,113 +1,15 @@ package wrpc import ( - "bufio" "bytes" + "context" "errors" "fmt" - "io" "log/slog" "math" + "sync" ) -// ReadChunkSize is the size of chunks read from byte streams -const ReadChunkSize = 8192 - -type ByteStreamWriter struct { - r io.Reader - chunk []byte -} - -func (v *ByteStreamWriter) WriteTo(w ByteWriter) (err error) { - if len(v.chunk) == 0 { - v.chunk = make([]byte, ReadChunkSize) - } - buf := bufio.NewWriter(w) - defer func() { - if fErr := buf.Flush(); fErr != nil { - if err == nil { - err = fmt.Errorf("failed to flush pending byte stream writer: %w", fErr) - } else { - slog.Warn("failed to flush pending byte stream writer", "err", fErr) - } - } - }() - for { - var end bool - slog.Debug("reading pending byte stream contents") - n, err := v.r.Read(v.chunk) - if err == io.EOF { - end = true - slog.Debug("pending byte stream reached EOF") - } else if err != nil { - return fmt.Errorf("failed to read pending byte stream chunk: %w", err) - } - if n > math.MaxUint32 { - return fmt.Errorf("pending byte stream chunk length of %d overflows a 32-bit integer", n) - } - if n > 0 { - slog.Debug("writing pending byte stream chunk length", "len", n) - _, err = WriteUint32(uint32(n), buf) - if err != nil { - return fmt.Errorf("failed to write pending byte stream chunk length of %d: %w", n, err) - } - _, err = buf.Write(v.chunk[:n]) - if err != nil { - return fmt.Errorf("failed to write pending byte stream chunk contents: %w", err) - } - } - if end { - if err := buf.WriteByte(0); err != nil { - return fmt.Errorf("failed to write pending byte stream end byte: %w", err) - } - return nil - } - } -} - -type ByteStreamReader struct { - r ByteReadCloser - buf uint32 -} - -func (r *ByteStreamReader) Read(p []byte) (int, error) { - n := r.buf - if n == 0 { - slog.Debug("reading pending byte stream chunk length") - var err error - n, err = ReadUint32(r.r) - if err != nil { - return 0, fmt.Errorf("failed to read pending byte stream chunk length: %w", err) - } - if n == 0 { - return 0, io.EOF - } - } - if len(p) > int(n) { - p = p[:n] - } - slog.Debug("reading pending byte stream chunk contents", "len", n) - rn, err := r.r.Read(p) - if err != nil { - return rn, fmt.Errorf("failed to read pending stream chunk bytes: %w", err) - } - if rn > int(n) { - return rn, errors.New("read more bytes than requested") - } - r.buf = n - uint32(rn) - return rn, nil -} - -func (r *ByteStreamReader) Close() error { - return r.r.Close() -} - -func NewByteStreamReader(r ByteReadCloser) *ByteStreamReader { - return &ByteStreamReader{ - r: r, - } -} - // ReadStreamStatus reads a single byte from `r` and returns: // - `true` if stream is "ready" // - `false` if stream is "pending" @@ -126,92 +28,272 @@ func ReadStreamStatus(r ByteReader) (bool, error) { } } -// ReadByteStream reads a stream of bytes from `r` -func ReadByteStream(r IndexReader, path ...uint32) (io.ReadCloser, error) { +// ReadByteStream reads a `stream` from `r`. +// The returned channel receives the chunks of the stream and is closed once the stream ends. +// If the stream is pending, the returned function reads it and must be called. It closes the channel +// if reading fails or `ctx` is done. +func ReadByteStream(r IndexReadCloser, path ...uint32) (<-chan []byte, func(context.Context) error, error) { slog.Debug("reading byte stream status byte") ok, err := ReadStreamStatus(r) if err != nil { - return nil, err + return nil, nil, err } - if !ok { - r, err := r.Index(path...) + if ok { + slog.Debug("reading ready byte stream") + buf, err := ReadByteList(r) if err != nil { - return nil, fmt.Errorf("failed to index reader: %w", err) + return nil, nil, fmt.Errorf("failed to read ready byte stream: %w", err) + } + ch := make(chan []byte, 1) + if len(buf) > 0 { + ch <- buf } - return NewByteStreamReader(r), nil + close(ch) + return ch, nil, nil } - slog.Debug("reading ready byte stream") - buf, err := ReadByteList(r) - if err != nil { - return nil, fmt.Errorf("failed to read bytes: %w", err) + if len(path) > 0 { + slog.Debug("indexing pending byte stream reader") + r, err = r.Index(path...) + if err != nil { + return nil, nil, fmt.Errorf("failed to index nested byte stream reader: %w", err) + } } - slog.Debug("read ready byte stream", "len", len(buf)) - return io.NopCloser(bytes.NewReader(buf)), nil + ch := make(chan []byte) + return ch, func(ctx context.Context) error { + if len(path) > 0 { + defer closeNested(r) + } + defer close(ch) + for { + slog.Debug("reading pending byte stream chunk") + buf, err := ReadByteList(r) + if err != nil { + return fmt.Errorf("failed to read pending byte stream chunk: %w", err) + } + if len(buf) == 0 { + slog.Debug("pending byte stream ended") + return nil + } + select { + case ch <- buf: + case <-ctx.Done(): + return fmt.Errorf("failed to send pending byte stream chunk: %w", context.Cause(ctx)) + } + } + }, nil } -// ReadStream reads a stream from `r` -func ReadStream[T any](r IndexReader, f func(IndexReader) (T, error), path ...uint32) (Receiver[[]T], error) { +// ReadStream reads a `stream` from `r` using `f` to read its elements. +// The returned channel receives the chunks of the stream and is closed once the stream ends. +// If the stream is pending, the returned function reads it and must be called. It closes the channel +// if reading fails or `ctx` is done. +func ReadStream[T any](r IndexReadCloser, f func(IndexReadCloser, ...uint32) (T, func(context.Context) error, error), path ...uint32) (<-chan []T, func(context.Context) error, error) { slog.Debug("reading stream status byte") ok, err := ReadStreamStatus(r) if err != nil { - return nil, err + return nil, nil, err } - if !ok { - r, err := r.Index(path...) + if ok { + slog.Debug("reading ready stream length") + n, err := ReadUint32(r) if err != nil { - return nil, fmt.Errorf("failed to index reader: %w", err) + return nil, nil, fmt.Errorf("failed to read ready stream length: %w", err) } - return NewDecodeReceiver(r, func(r IndexReadCloser) ([]T, error) { - slog.Debug("reading pending stream chunk length") - n, err := ReadUint32(r) + vs := NewSlice[[]T](n) + var ds []func(context.Context) error + for i := range n { + v, deferred, err := f(r, append(path, i)...) if err != nil { - return nil, fmt.Errorf("failed to read pending stream chunk length: %w", err) - } - if n == 0 { - slog.Debug("pending stream EOF chunk received") - return nil, io.EOF + return nil, nil, fmt.Errorf("failed to read ready stream element %d: %w", i, err) } - vs := NewSlice[[]T](n) - for i := range n { - v, err := f(r) + vs = append(vs, v) + ds = append(ds, deferred) + } + ch := make(chan []T, 1) + if len(vs) > 0 { + ch <- vs + } + close(ch) + return ch, JoinDeferred(ds...), nil + } + if len(path) > 0 { + slog.Debug("indexing pending stream reader") + r, err = r.Index(path...) + if err != nil { + return nil, nil, fmt.Errorf("failed to index nested stream reader: %w", err) + } + } + ch := make(chan []T) + return ch, func(ctx context.Context) error { + if len(path) > 0 { + defer closeNested(r) + } + var wg sync.WaitGroup + var errMu sync.Mutex + var errs []error + err := func() error { + defer close(ch) + var total uint32 + for { + slog.Debug("reading pending stream chunk length") + n, err := ReadUint32(r) if err != nil { - return nil, fmt.Errorf("failed to read pending stream chunk element %d: %w", i, err) + return fmt.Errorf("failed to read pending stream chunk length: %w", err) + } + if n == 0 { + slog.Debug("pending stream ended") + return nil + } + if math.MaxUint32-n < total { + return errors.New("total pending stream element count would overflow a 32-bit unsigned integer") + } + vs := NewSlice[[]T](n) + for range n { + v, deferred, err := f(r, total) + if err != nil { + return fmt.Errorf("failed to read pending stream element %d: %w", total, err) + } + if deferred != nil { + wg.Add(1) + go func() { + defer wg.Done() + if err := deferred(ctx); err != nil { + errMu.Lock() + errs = append(errs, err) + errMu.Unlock() + } + }() + } + vs = append(vs, v) + total++ + } + select { + case ch <- vs: + case <-ctx.Done(): + return fmt.Errorf("failed to send pending stream chunk: %w", context.Cause(ctx)) } - vs = append(vs, v) } - return vs, nil - }), nil + }() + wg.Wait() + return errors.Join(append(errs, err)...) + }, nil +} + +// WriteByteStream writes a `stream` with chunks received from `v` to `w` +func WriteByteStream(v <-chan []byte, w ByteWriter) (func(IndexWriter) error, error) { + if v == nil { + return nil, errors.New("outgoing byte stream channel is nil") } - slog.Debug("reading ready stream") - vs, err := ReadList(r, f) - if err != nil { - return nil, fmt.Errorf("failed to read ready stream: %w", err) + slog.Debug("writing `stream::pending` status byte") + if err := w.WriteByte(0); err != nil { + return nil, fmt.Errorf("failed to write `stream::pending` byte: %w", err) } - slog.Debug("read ready stream", "len", len(vs)) - return NewCompleteReceiver(vs), nil + return func(w IndexWriter) error { + var buf bytes.Buffer + for chunk := range v { + if len(chunk) == 0 { + continue + } + slog.Debug("writing pending byte stream chunk", "len", len(chunk)) + buf.Reset() + if _, err := WriteByteList(chunk, &buf); err != nil { + go drain(v) + return fmt.Errorf("failed to encode pending byte stream chunk: %w", err) + } + if _, err := w.Write(buf.Bytes()); err != nil { + go drain(v) + return fmt.Errorf("failed to write pending byte stream chunk: %w", err) + } + } + slog.Debug("writing pending byte stream end") + if err := w.WriteByte(0); err != nil { + return fmt.Errorf("failed to write pending byte stream end byte: %w", err) + } + return nil + }, nil } -func WriteByteStream(r io.Reader, w IndexWriter, chunk []byte, path ...uint32) (err error) { - slog.Debug("writing byte stream `stream::pending` status byte") - if err := w.WriteByte(0); err != nil { - return fmt.Errorf("failed to write `stream::pending` byte: %w", err) +// WriteStream writes a `stream` with chunks received from `v` to `w` using `f` to write its elements +func WriteStream[T any](v <-chan []T, w ByteWriter, f func(T, ByteWriter) (func(IndexWriter) error, error)) (func(IndexWriter) error, error) { + if v == nil { + return nil, errors.New("outgoing stream channel is nil") } - wi, err := w.Index(path...) - if err != nil { - return fmt.Errorf("failed to index reader: %w", err) + slog.Debug("writing `stream::pending` status byte") + if err := w.WriteByte(0); err != nil { + return nil, fmt.Errorf("failed to write `stream::pending` byte: %w", err) } - s := &ByteStreamWriter{r, chunk} - defer func() { - if cErr := wi.Close(); cErr != nil { - if err == nil { - err = fmt.Errorf("failed to close pending byte stream: %w", cErr) - } else { - slog.Warn("failed to close pending byte stream", "err", cErr) + return func(w IndexWriter) error { + var wg sync.WaitGroup + var errMu sync.Mutex + var errs []error + err := func() error { + var total uint32 + var buf bytes.Buffer + writes := make(map[uint32]func(IndexWriter) error) + for chunk := range v { + n := len(chunk) + if n == 0 { + continue + } + if n > math.MaxUint32 || math.MaxUint32-uint32(n) < total { + return errors.New("total pending stream element count would overflow a 32-bit unsigned integer") + } + slog.Debug("writing pending stream chunk", "len", n) + buf.Reset() + if _, err := WriteUint32(uint32(n), &buf); err != nil { + return fmt.Errorf("failed to encode pending stream chunk length of %d: %w", n, err) + } + clear(writes) + for _, v := range chunk { + write, err := f(v, &buf) + if err != nil { + return fmt.Errorf("failed to encode pending stream element %d: %w", total, err) + } + if write != nil { + writes[total] = write + } + total++ + } + if _, err := w.Write(buf.Bytes()); err != nil { + return fmt.Errorf("failed to write pending stream chunk: %w", err) + } + for i, write := range writes { + w, err := w.Index(i) + if err != nil { + return fmt.Errorf("failed to index nested stream writer: %w", err) + } + wg.Add(1) + go func() { + defer wg.Done() + if err := write(w); err != nil { + errMu.Lock() + errs = append(errs, err) + errMu.Unlock() + } + }() + } + } + slog.Debug("writing pending stream end") + if err := w.WriteByte(0); err != nil { + return fmt.Errorf("failed to write pending stream end byte: %w", err) } + return nil + }() + if err != nil { + go drain(v) } - }() - if err := s.WriteTo(wi); err != nil { - return fmt.Errorf("failed to write stream contents: %w", err) + wg.Wait() + return errors.Join(append(errs, err)...) + }, nil +} + +func closeNested(r IndexReadCloser) { + if err := r.Close(); err != nil { + slog.Debug("failed to close nested reader", "err", err) + } +} + +func drain[T any](v <-chan T) { + for range v { } - return nil } diff --git a/go/wrpc.go b/go/wrpc.go index 56bf67fb7..3f9e49b5e 100644 --- a/go/wrpc.go +++ b/go/wrpc.go @@ -3,7 +3,10 @@ package wrpc import ( "bufio" "context" + "errors" "io" + "slices" + "sync" ) // Invoke is the client-side transport handle @@ -81,6 +84,9 @@ type IndexReader interface { io.ByteReader Index[IndexReadCloser] + + // Buffered returns the number of bytes, which were received, but not read yet + Buffered() int } type IndexWriter interface { @@ -110,9 +116,29 @@ type ByteReader interface { io.Reader } -type Receiver[T any] interface { - Receive() (T, error) - io.Closer +// JoinDeferred returns a function, which calls all non-nil `fs` concurrently and returns their joined errors. +// It returns `nil` if all `fs` are `nil`. +func JoinDeferred(fs ...func(context.Context) error) func(context.Context) error { + fs = slices.DeleteFunc(fs, func(f func(context.Context) error) bool { return f == nil }) + switch len(fs) { + case 0: + return nil + case 1: + return fs[0] + } + return func(ctx context.Context) error { + errs := make([]error, len(fs)) + var wg sync.WaitGroup + for i, f := range fs { + wg.Add(1) + go func() { + defer wg.Done() + errs[i] = f(ctx) + }() + } + wg.Wait() + return errors.Join(errs...) + } } type ByteReadCloser interface { @@ -120,65 +146,6 @@ type ByteReadCloser interface { io.Closer } -type CompleteReceiver[T any] struct { - v T - ready bool -} - -func (r *CompleteReceiver[T]) Receive() (T, error) { - defer func() { - r.ready = false - }() - if !r.ready { - return CompleteReceiver[T]{}.v, io.EOF - } - return r.v, nil -} - -func (r *CompleteReceiver[T]) Close() error { - if c, ok := (any)(r.v).(io.Closer); ok { - return c.Close() - } - return nil -} - -func NewCompleteReceiver[T any](v T) *CompleteReceiver[T] { - return &CompleteReceiver[T]{v, true} -} - -type DecodeReceiver[T any] struct { - r IndexReadCloser - decode func(IndexReadCloser) (T, error) -} - -func (r *DecodeReceiver[T]) Receive() (T, error) { - return r.decode(r.r) -} - -func (r *DecodeReceiver[T]) Close() error { - return r.r.Close() -} - -func NewDecodeReceiver[T any](r IndexReadCloser, decode func(IndexReadCloser) (T, error)) *DecodeReceiver[T] { - return &DecodeReceiver[T]{r, decode} -} - -type nestedReceiver[T any, A Receiver[B], B Receiver[T]] struct { - rx A -} - -func (r nestedReceiver[T, A, B]) Receive() (Receiver[T], error) { - return r.rx.Receive() -} - -func (r nestedReceiver[T, A, B]) Close() error { - return r.rx.Close() -} - -func NewNestedReceiver[T any, A Receiver[B], B Receiver[T]](rx A) Receiver[Receiver[T]] { - return nestedReceiver[T, A, B]{rx} -} - type BufReadCloser struct { *bufio.Reader io.Closer diff --git a/tests/go/async.go b/tests/go/async.go index 46a494afc..a4801dcee 100644 --- a/tests/go/async.go +++ b/tests/go/async.go @@ -3,34 +3,30 @@ package integration import ( - "bytes" "context" "errors" - "io" "log/slog" - wrpc "wrpc.io/go" "wrpc.io/tests/go/bindings/async_server/exports/wrpc_test/integration/async" + "wrpc.io/tests/go/internal" ) type AsyncHandler struct{} -func (AsyncHandler) WithStreams(ctx context.Context) (io.ReadCloser, wrpc.Receiver[[][]string], error) { +func (AsyncHandler) WithStreams(ctx context.Context) (<-chan []byte, <-chan [][]string, error) { slog.DebugContext(ctx, "handling `with-streams`") - buf := io.NopCloser(bytes.NewBuffer([]byte("test"))) - str := wrpc.NewCompleteReceiver([][]string{{"foo"}, {"bar", "baz"}}) - return buf, str, nil + return internal.Ready([]byte("test")), internal.Ready([][]string{{"foo"}, {"bar", "baz"}}), nil } -func (AsyncHandler) WithFuture(ctx context.Context, x *async.Something, s io.ReadCloser) (wrpc.Receiver[io.ReadCloser], error) { +func (AsyncHandler) WithFuture(ctx context.Context, x *async.Something, s <-chan []byte) (<-chan <-chan []byte, error) { slog.DebugContext(ctx, "handling `with-future`", "x", x) if x.Foo != "bar" { return nil, errors.New("`foo` is not `bar`") } - return wrpc.NewCompleteReceiver(s), nil + return internal.Ready(s), nil } -func (AsyncHandler) IdentityNestedAsync(ctx context.Context, v wrpc.Receiver[wrpc.Receiver[wrpc.Receiver[wrpc.Receiver[[]string]]]]) (wrpc.Receiver[wrpc.Receiver[wrpc.Receiver[wrpc.Receiver[[]string]]]], error) { +func (AsyncHandler) IdentityNestedAsync(ctx context.Context, v <-chan <-chan <-chan <-chan []string) (<-chan <-chan <-chan <-chan []string, error) { slog.DebugContext(ctx, "handling `identity-nested-async`") return v, nil } diff --git a/tests/go/async_test.go b/tests/go/async_test.go index 2894755bc..aa3a21d45 100644 --- a/tests/go/async_test.go +++ b/tests/go/async_test.go @@ -4,38 +4,18 @@ package integration_test import ( "context" - "io" "log/slog" - "reflect" "testing" "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - wrpc "wrpc.io/go" integration "wrpc.io/tests/go" "wrpc.io/tests/go/bindings/async_client/wrpc_test/integration/async" "wrpc.io/tests/go/bindings/async_server" "wrpc.io/tests/go/internal" ) -type chunkReceiver[T any] struct { - chunks [][]T -} - -func (r *chunkReceiver[T]) Receive() ([]T, error) { - if len(r.chunks) == 0 { - return nil, io.EOF - } - var v []T - v, r.chunks = r.chunks[0], r.chunks[1:] - return v, nil -} - -func (r *chunkReceiver[T]) Close() error { - return nil -} - func TestAsync(t *testing.T) { var cancel func() ctx := context.Background() @@ -58,87 +38,67 @@ func TestAsync(t *testing.T) { t.Run("with-streams", func(t *testing.T) { slog.DebugContext(ctx, "calling `wrpc-test:integration/async.with-streams`") - byteRx, stringListRx, err := async.WithStreams(ctx, client) - if err != nil { - t.Errorf("failed to call `wrpc-test:integration/async.with-streams`: %s", err) - return - } - b, err := io.ReadAll(byteRx) - if err != nil { - t.Errorf("failed to read from stream: %s", err) - return + byteRx, stringListRx, errCh, err := async.WithStreams(ctx, client) + require.NoError(t, err, "failed to call `wrpc-test:integration/async.with-streams`") + var b []byte + for chunk := range byteRx { + b = append(b, chunk...) } - if string(b) != "test" { - t.Errorf("expected: `test`, got: %s", string(b)) - return + assert.Equal(t, "test", string(b)) + var ss [][]string + for chunk := range stringListRx { + ss = append(ss, chunk...) } - if err := byteRx.Close(); err != nil { - t.Errorf("failed to close byte reader: %s", err) - return + assert.Equal(t, [][]string{{"foo"}, {"bar", "baz"}}, ss) + for err := range errCh { + require.NoError(t, err, "failed to receive async value from peer") } + }) - ss, err := stringListRx.Receive() - if err != nil { - t.Errorf("failed to receive ready list stream: %s", err) - return - } - expected := [][]string{{"foo"}, {"bar", "baz"}} - if !reflect.DeepEqual(ss, expected) { - t.Errorf("expected: `%#v`, got: %#v", expected, ss) - return + t.Run("with-streams-cancel", func(t *testing.T) { + callCtx, callCancel := context.WithCancel(ctx) + _, stringListRx, errCh, err := async.WithStreams(callCtx, client) + require.NoError(t, err, "failed to call `wrpc-test:integration/async.with-streams`") + for range stringListRx { } - ss, err = stringListRx.Receive() - if ss != nil || err != io.EOF { - t.Errorf("ready list should have returned (nil, io.EOF), got: (%#v, %v)", ss, err) - return - } - if err := stringListRx.Close(); err != nil { - t.Errorf("failed to close string list receiver: %s", err) - return + callCancel() + timeout := time.After(5 * time.Second) + for { + select { + case _, ok := <-errCh: + if !ok { + return + } + case <-timeout: + t.Fatal("error channel not closed after cancellation") + } } }) t.Run("identity-nested-async", func(t *testing.T) { slog.DebugContext(ctx, "calling `wrpc-test:integration/async.identity-nested-async`") + stream := make(chan []string, 3) + stream <- []string{"foo"} + stream <- []string{} + stream <- []string{"bar", "baz"} + close(stream) fut0, errCh, err := async.IdentityNestedAsync(ctx, client, - wrpc.NewNestedReceiver(wrpc.NewCompleteReceiver( - wrpc.NewCompleteReceiver( - wrpc.NewNestedReceiver(wrpc.NewCompleteReceiver( - &chunkReceiver[string]{[][]string{{"foo"}, {}, {"bar", "baz"}}}, - )), - ), - )), + internal.Ready(internal.Ready(internal.Ready((<-chan []string)(stream)))), ) - if err != nil { - t.Errorf("failed to call `wrpc-test:integration/async.identity-nested-async`: %s", err) - return - } - for err := range errCh { - t.Errorf("failed to send async value to peer: %s", err) - return - } - fut1, err := fut0.Receive() - require.NoError(t, err, "failed to receive future 0") - fut2, err := fut1.Receive() - require.NoError(t, err, "failed to receive future 1") - - stream, err := fut2.Receive() - require.NoError(t, err, "failed to receive future 2") - + require.NoError(t, err, "failed to call `wrpc-test:integration/async.identity-nested-async`") + fut1, ok := <-fut0 + require.True(t, ok, "failed to receive future 0") + fut2, ok := <-fut1 + require.True(t, ok, "failed to receive future 1") + rx, ok := <-fut2 + require.True(t, ok, "failed to receive future 2") var ss []string - for err != io.EOF { - var v []string - v, err = stream.Receive() - if err != io.EOF { - require.NoError(t, err, "failed to receive stream element") - } - ss = append(ss, v...) + for chunk := range rx { + ss = append(ss, chunk...) } - assert.Equal(t, ss, []string{"foo", "bar", "baz"}) - - if err := fut0.Close(); err != nil { - t.Errorf("failed to close async value receiver: %s", err) - return + assert.Equal(t, []string{"foo", "bar", "baz"}, ss) + for err := range errCh { + require.NoError(t, err, "failed to exchange async values with peer") } }) diff --git a/tests/go/internal/internal.go b/tests/go/internal/internal.go index 957fa86b5..d01250941 100644 --- a/tests/go/internal/internal.go +++ b/tests/go/internal/internal.go @@ -64,3 +64,11 @@ func Accept(ctx context.Context, srv *wrpctcp.Server) { } }() } + +// Ready returns a closed channel, which contains `v` +func Ready[T any](v T) <-chan T { + ch := make(chan T, 1) + ch <- v + close(ch) + return ch +} From 595b458620498ba956f4ef4e98dd40198be68b28 Mon Sep 17 00:00:00 2001 From: Roman Volosatovs Date: Mon, 26 Oct 2026 19:22:11 +0100 Subject: [PATCH 2/2] fix(wit-bindgen-go): log the reader close error The served function logged the handler error instead of the error returned by closing the parameter reader. Assisted-by: claude:claude-opus-5-5 --- crates/wit-bindgen-go/src/interface.rs | 2 +- .../wrpc_examples/hello/handler/bindings.wrpc.go | 2 +- .../resources/resources/bindings.wrpc.go | 8 ++++---- .../wrpc_examples/streams/handler/bindings.wrpc.go | 2 +- .../exports/wasi/keyvalue/store/bindings.wrpc.go | 12 ++++++------ 5 files changed, 13 insertions(+), 13 deletions(-) diff --git a/crates/wit-bindgen-go/src/interface.rs b/crates/wit-bindgen-go/src/interface.rs index ff70d5403..f5af12e92 100644 --- a/crates/wit-bindgen-go/src/interface.rs +++ b/crates/wit-bindgen-go/src/interface.rs @@ -2277,7 +2277,7 @@ func ServeInterface(s {wrpc}.Server, h Handler) (stop func() error, err error) { self.src, r#") if cErr := r.Close(); cErr != nil {{ - {slog}.ErrorContext(ctx, "failed to close reader", "instance", "{instance}", "name", "{name}", "err", err) + {slog}.ErrorContext(ctx, "failed to close reader", "instance", "{instance}", "name", "{name}", "err", cErr) }} if err != nil {{ {slog}.WarnContext(ctx, "failed to handle invocation", "instance", "{instance}", "name", "{name}", "err", err) diff --git a/examples/go/hello-server/bindings/exports/wrpc_examples/hello/handler/bindings.wrpc.go b/examples/go/hello-server/bindings/exports/wrpc_examples/hello/handler/bindings.wrpc.go index 91daab8cb..925871ddb 100644 --- a/examples/go/hello-server/bindings/exports/wrpc_examples/hello/handler/bindings.wrpc.go +++ b/examples/go/hello-server/bindings/exports/wrpc_examples/hello/handler/bindings.wrpc.go @@ -36,7 +36,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { slog.DebugContext(ctx, "calling `wrpc-examples:hello/handler.hello` handler") r0, err := h.Hello(ctx) if cErr := r.Close(); cErr != nil { - slog.ErrorContext(ctx, "failed to close reader", "instance", "wrpc-examples:hello/handler", "name", "hello", "err", err) + slog.ErrorContext(ctx, "failed to close reader", "instance", "wrpc-examples:hello/handler", "name", "hello", "err", cErr) } if err != nil { slog.WarnContext(ctx, "failed to handle invocation", "instance", "wrpc-examples:hello/handler", "name", "hello", "err", err) diff --git a/examples/go/resources-server/bindings/exports/wrpc_examples/resources/resources/bindings.wrpc.go b/examples/go/resources-server/bindings/exports/wrpc_examples/resources/resources/bindings.wrpc.go index 9bc0d36a9..0d7441e36 100644 --- a/examples/go/resources-server/bindings/exports/wrpc_examples/resources/resources/bindings.wrpc.go +++ b/examples/go/resources-server/bindings/exports/wrpc_examples/resources/resources/bindings.wrpc.go @@ -41,7 +41,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { slog.DebugContext(ctx, "calling `wrpc-examples:resources/resources.foo` handler") r0, err := h.Foo(ctx) if cErr := r.Close(); cErr != nil { - slog.ErrorContext(ctx, "failed to close reader", "instance", "wrpc-examples:resources/resources", "name", "foo", "err", err) + slog.ErrorContext(ctx, "failed to close reader", "instance", "wrpc-examples:resources/resources", "name", "foo", "err", cErr) } if err != nil { slog.WarnContext(ctx, "failed to handle invocation", "instance", "wrpc-examples:resources/resources", "name", "foo", "err", err) @@ -160,7 +160,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { slog.DebugContext(ctx, "calling `wrpc-examples:resources/resources.foo.baz` handler") r0, err := h.Foo_Baz(ctx, p0) if cErr := r.Close(); cErr != nil { - slog.ErrorContext(ctx, "failed to close reader", "instance", "wrpc-examples:resources/resources", "name", "foo.baz", "err", err) + slog.ErrorContext(ctx, "failed to close reader", "instance", "wrpc-examples:resources/resources", "name", "foo.baz", "err", cErr) } if err != nil { slog.WarnContext(ctx, "failed to handle invocation", "instance", "wrpc-examples:resources/resources", "name", "foo.baz", "err", err) @@ -279,7 +279,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { slog.DebugContext(ctx, "calling `wrpc-examples:resources/resources.foo.bar` handler") r0, err := h.Foo_Bar(ctx, p0) if cErr := r.Close(); cErr != nil { - slog.ErrorContext(ctx, "failed to close reader", "instance", "wrpc-examples:resources/resources", "name", "foo.bar", "err", err) + slog.ErrorContext(ctx, "failed to close reader", "instance", "wrpc-examples:resources/resources", "name", "foo.bar", "err", cErr) } if err != nil { slog.WarnContext(ctx, "failed to handle invocation", "instance", "wrpc-examples:resources/resources", "name", "foo.bar", "err", err) @@ -398,7 +398,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { slog.DebugContext(ctx, "calling `wrpc-examples:resources/resources.bar` handler") r0, err := h.Bar(ctx, p0) if cErr := r.Close(); cErr != nil { - slog.ErrorContext(ctx, "failed to close reader", "instance", "wrpc-examples:resources/resources", "name", "bar", "err", err) + slog.ErrorContext(ctx, "failed to close reader", "instance", "wrpc-examples:resources/resources", "name", "bar", "err", cErr) } if err != nil { slog.WarnContext(ctx, "failed to handle invocation", "instance", "wrpc-examples:resources/resources", "name", "bar", "err", err) diff --git a/examples/go/streams-server/bindings/exports/wrpc_examples/streams/handler/bindings.wrpc.go b/examples/go/streams-server/bindings/exports/wrpc_examples/streams/handler/bindings.wrpc.go index fb30f7724..b055a73de 100644 --- a/examples/go/streams-server/bindings/exports/wrpc_examples/streams/handler/bindings.wrpc.go +++ b/examples/go/streams-server/bindings/exports/wrpc_examples/streams/handler/bindings.wrpc.go @@ -168,7 +168,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { slog.DebugContext(ctx, "calling `wrpc-examples:streams/handler.echo` handler") r0, r1, err := h.Echo(ctx, p0) if cErr := r.Close(); cErr != nil { - slog.ErrorContext(ctx, "failed to close reader", "instance", "wrpc-examples:streams/handler", "name", "echo", "err", err) + slog.ErrorContext(ctx, "failed to close reader", "instance", "wrpc-examples:streams/handler", "name", "echo", "err", cErr) } if err != nil { slog.WarnContext(ctx, "failed to handle invocation", "instance", "wrpc-examples:streams/handler", "name", "echo", "err", err) diff --git a/examples/go/wasi-keyvalue-server/bindings/exports/wasi/keyvalue/store/bindings.wrpc.go b/examples/go/wasi-keyvalue-server/bindings/exports/wasi/keyvalue/store/bindings.wrpc.go index c21689acc..e67f97691 100644 --- a/examples/go/wasi-keyvalue-server/bindings/exports/wasi/keyvalue/store/bindings.wrpc.go +++ b/examples/go/wasi-keyvalue-server/bindings/exports/wasi/keyvalue/store/bindings.wrpc.go @@ -473,7 +473,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { slog.DebugContext(ctx, "calling `wasi:keyvalue/store@0.2.0-draft2.open` handler") r0, err := h.Open(ctx, p0) if cErr := r.Close(); cErr != nil { - slog.ErrorContext(ctx, "failed to close reader", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "open", "err", err) + slog.ErrorContext(ctx, "failed to close reader", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "open", "err", cErr) } if err != nil { slog.WarnContext(ctx, "failed to handle invocation", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "open", "err", err) @@ -678,7 +678,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { slog.DebugContext(ctx, "calling `wasi:keyvalue/store@0.2.0-draft2.bucket.get` handler") r0, err := h.Bucket_Get(ctx, p0, p1) if cErr := r.Close(); cErr != nil { - slog.ErrorContext(ctx, "failed to close reader", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "bucket.get", "err", err) + slog.ErrorContext(ctx, "failed to close reader", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "bucket.get", "err", cErr) } if err != nil { slog.WarnContext(ctx, "failed to handle invocation", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "bucket.get", "err", err) @@ -985,7 +985,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { slog.DebugContext(ctx, "calling `wasi:keyvalue/store@0.2.0-draft2.bucket.set` handler") r0, err := h.Bucket_Set(ctx, p0, p1, p2) if cErr := r.Close(); cErr != nil { - slog.ErrorContext(ctx, "failed to close reader", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "bucket.set", "err", err) + slog.ErrorContext(ctx, "failed to close reader", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "bucket.set", "err", cErr) } if err != nil { slog.WarnContext(ctx, "failed to handle invocation", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "bucket.set", "err", err) @@ -1162,7 +1162,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { slog.DebugContext(ctx, "calling `wasi:keyvalue/store@0.2.0-draft2.bucket.delete` handler") r0, err := h.Bucket_Delete(ctx, p0, p1) if cErr := r.Close(); cErr != nil { - slog.ErrorContext(ctx, "failed to close reader", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "bucket.delete", "err", err) + slog.ErrorContext(ctx, "failed to close reader", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "bucket.delete", "err", cErr) } if err != nil { slog.WarnContext(ctx, "failed to handle invocation", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "bucket.delete", "err", err) @@ -1339,7 +1339,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { slog.DebugContext(ctx, "calling `wasi:keyvalue/store@0.2.0-draft2.bucket.exists` handler") r0, err := h.Bucket_Exists(ctx, p0, p1) if cErr := r.Close(); cErr != nil { - slog.ErrorContext(ctx, "failed to close reader", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "bucket.exists", "err", err) + slog.ErrorContext(ctx, "failed to close reader", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "bucket.exists", "err", cErr) } if err != nil { slog.WarnContext(ctx, "failed to handle invocation", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "bucket.exists", "err", err) @@ -1550,7 +1550,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) { slog.DebugContext(ctx, "calling `wasi:keyvalue/store@0.2.0-draft2.bucket.list-keys` handler") r0, err := h.Bucket_ListKeys(ctx, p0, p1) if cErr := r.Close(); cErr != nil { - slog.ErrorContext(ctx, "failed to close reader", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "bucket.list-keys", "err", err) + slog.ErrorContext(ctx, "failed to close reader", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "bucket.list-keys", "err", cErr) } if err != nil { slog.WarnContext(ctx, "failed to handle invocation", "instance", "wasi:keyvalue/store@0.2.0-draft2", "name", "bucket.list-keys", "err", err)