Skip to content

Commit 68e7fdb

Browse files
rvolosatovsrvolosatovs
committed
feat(go)!: always assume io.Closer for async
Signed-off-by: Roman Volosatovs <rvolosatovs@riseup.net>
1 parent 2ac2a0e commit 68e7fdb

14 files changed

Lines changed: 159 additions & 161 deletions

File tree

‎crates/wit-bindgen-go/src/interface.rs‎

Lines changed: 35 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -939,7 +939,7 @@ impl InterfaceGenerator<'_> {
939939

940940
uwriteln!(
941941
self.src,
942-
r#"func(r {wrpc}.IndexReader, path ...uint32) ({io}.Reader, error) {{
942+
r#"func(r {wrpc}.IndexReader, path ...uint32) ({io}.ReadCloser, error) {{
943943
{slog}.Debug("reading byte list future status byte")
944944
status, err := r.ReadByte()
945945
if err != nil {{
@@ -965,7 +965,7 @@ impl InterfaceGenerator<'_> {
965965
return nil, {fmt}.Errorf("failed to read ready byte list future contents: %w", err)
966966
}}
967967
{slog}.Debug("read ready byte list future contents", "len", len(buf))
968-
return {bytes}.NewReader(buf), nil
968+
return {io}.NopCloser({bytes}.NewReader(buf)), nil
969969
default:
970970
return nil, {fmt}.Errorf("invalid byte list future status byte %d", status)
971971
}}
@@ -985,7 +985,7 @@ impl InterfaceGenerator<'_> {
985985

986986
uwrite!(
987987
self.src,
988-
r#"func(r {wrpc}.IndexReader, path ...uint32) ({wrpc}.Receiver["#
988+
r#"func(r {wrpc}.IndexReader, path ...uint32) ({wrpc}.ReceiveCloser["#
989989
);
990990
self.print_opt_ty(ty, true);
991991
uwrite!(
@@ -1060,7 +1060,7 @@ impl InterfaceGenerator<'_> {
10601060

10611061
uwriteln!(
10621062
self.src,
1063-
r#"func(r {wrpc}.IndexReader, path ...uint32) ({io}.Reader, error) {{
1063+
r#"func(r {wrpc}.IndexReader, path ...uint32) ({io}.ReadCloser, error) {{
10641064
{slog}.Debug("reading byte stream status byte")
10651065
status, err := r.ReadByte()
10661066
if err != nil {{
@@ -1085,7 +1085,7 @@ impl InterfaceGenerator<'_> {
10851085
return nil, {fmt}.Errorf("failed to read ready byte stream contents: %w", err)
10861086
}}
10871087
{slog}.Debug("read ready byte stream contents", "len", len(buf))
1088-
return {bytes}.NewReader(buf), nil
1088+
return {io}.NopCloser({bytes}.NewReader(buf)), nil
10891089
default:
10901090
return nil, {fmt}.Errorf("invalid stream status byte %d", status)
10911091
}}
@@ -1108,7 +1108,7 @@ impl InterfaceGenerator<'_> {
11081108

11091109
uwrite!(
11101110
self.src,
1111-
r#"func(r {wrpc}.IndexReader, path ...uint32) ({wrpc}.Receiver["#
1111+
r#"func(r {wrpc}.IndexReader, path ...uint32) ({wrpc}.ReceiveCloser["#
11121112
);
11131113
self.print_list(ty);
11141114
uwrite!(
@@ -1990,7 +1990,7 @@ impl InterfaceGenerator<'_> {
19901990
let io = self.deps.io();
19911991
let slog = self.deps.slog();
19921992
let wrpc = self.deps.wrpc();
1993-
uwrite!(self.src, "func(v {wrpc}.Receiver[",);
1993+
uwrite!(self.src, "func(v {wrpc}.ReceiveCloser[",);
19941994
self.print_opt_ty(ty, true);
19951995
uwrite!(
19961996
self.src,
@@ -2001,17 +2001,14 @@ impl InterfaceGenerator<'_> {
20012001
}}
20022002
return func(w {wrpc}.IndexWriter) (err error) {{
20032003
defer func() {{
2004-
body, ok := v.({io}.Closer)
2005-
if ok {{
2006-
{slog}.Debug("closing future writer")
2007-
if cErr := body.Close(); cErr != nil {{
2008-
if err == nil {{
2009-
err = {fmt}.Errorf("failed to close pending future: %w", cErr)
2010-
}} else {{
2011-
{slog}.Warn("failed to close pending future", "err", cErr)
2012-
}}
2013-
}}
2014-
}}
2004+
{slog}.Debug("closing future writer")
2005+
if cErr := v.Close(); cErr != nil {{
2006+
if err == nil {{
2007+
err = {fmt}.Errorf("failed to close pending future: %w", cErr)
2008+
}} else {{
2009+
{slog}.Warn("failed to close pending future", "err", cErr)
2010+
}}
2011+
}}
20152012
}}()
20162013
{slog}.Debug("receiving outgoing pending future contents")
20172014
rx, err := v.Receive()
@@ -2050,24 +2047,21 @@ impl InterfaceGenerator<'_> {
20502047
let wrpc = self.deps.wrpc();
20512048
uwrite!(
20522049
self.src,
2053-
r#"func(v {io}.Reader, w interface {{ {io}.ByteWriter; {io}.Writer }}) (write func({wrpc}.IndexWriter) error, err error) {{
2050+
r#"func(v {io}.ReadCloser, w interface {{ {io}.ByteWriter; {io}.Writer }}) (write func({wrpc}.IndexWriter) error, err error) {{
20542051
{slog}.Debug("writing byte stream `stream::pending` status byte")
20552052
if err = w.WriteByte(0); err != nil {{
20562053
return nil, fmt.Errorf("failed to write `stream::pending` byte: %w", err)
20572054
}}
20582055
return func(w {wrpc}.IndexWriter) (err error) {{
20592056
defer func() {{
2060-
body, ok := v.({io}.Closer)
2061-
if ok {{
2062-
{slog}.Debug("closing byte list stream writer")
2063-
if cErr := body.Close(); cErr != nil {{
2064-
if err == nil {{
2065-
err = {fmt}.Errorf("failed to close pending byte stream: %w", cErr)
2066-
}} else {{
2067-
{slog}.Warn("failed to close pending byte stream", "err", cErr)
2068-
}}
2069-
}}
2070-
}}
2057+
{slog}.Debug("closing byte list stream writer")
2058+
if cErr := v.Close(); cErr != nil {{
2059+
if err == nil {{
2060+
err = {fmt}.Errorf("failed to close pending byte stream: %w", cErr)
2061+
}} else {{
2062+
{slog}.Warn("failed to close pending byte stream", "err", cErr)
2063+
}}
2064+
}}
20712065
}}()
20722066
chunk := make([]byte, 8096)
20732067
for {{
@@ -2111,7 +2105,7 @@ impl InterfaceGenerator<'_> {
21112105
let slog = self.deps.slog();
21122106
let sync = self.deps.sync();
21132107
let wrpc = self.deps.wrpc();
2114-
uwrite!(self.src, "func(v {wrpc}.Receiver[",);
2108+
uwrite!(self.src, "func(v {wrpc}.ReceiveCloser[",);
21152109
self.print_list(ty);
21162110
uwrite!(
21172111
self.src,
@@ -2122,15 +2116,12 @@ impl InterfaceGenerator<'_> {
21222116
}}
21232117
return func(w {wrpc}.IndexWriter) (err error) {{
21242118
defer func() {{
2125-
body, ok := v.({io}.Closer)
2126-
if ok {{
2127-
{slog}.Debug("closing stream writer")
2128-
if cErr := body.Close(); cErr != nil {{
2129-
if err == nil {{
2130-
err = {fmt}.Errorf("failed to close pending stream: %w", cErr)
2131-
}} else {{
2132-
{slog}.Warn("failed to close pending stream", "err", cErr)
2133-
}}
2119+
{slog}.Debug("closing stream writer")
2120+
if cErr := v.Close(); cErr != nil {{
2121+
if err == nil {{
2122+
err = {fmt}.Errorf("failed to close pending stream: %w", cErr)
2123+
}} else {{
2124+
{slog}.Warn("failed to close pending stream", "err", cErr)
21342125
}}
21352126
}}
21362127
}}()
@@ -3183,12 +3174,12 @@ func ServeInterface(s {wrpc}.Server, h Handler) (stop func() error, err error) {
31833174
Some(ty) if is_ty(self.resolve, Type::U8, ty) => {
31843175
let io = self.deps.io();
31853176
self.push_str(io);
3186-
self.push_str(".Reader");
3177+
self.push_str(".ReadCloser");
31873178
}
31883179
Some(ty) => {
31893180
let wrpc = self.deps.wrpc();
31903181
self.push_str(wrpc);
3191-
self.push_str(".Receiver[");
3182+
self.push_str(".ReceiveCloser[");
31923183
self.print_opt_ty(ty, true);
31933184
self.push_str("]");
31943185
}
@@ -3203,12 +3194,12 @@ func ServeInterface(s {wrpc}.Server, h Handler) (stop func() error, err error) {
32033194
Some(ty) if is_ty(self.resolve, Type::U8, ty) => {
32043195
let io = self.deps.io();
32053196
self.push_str(io);
3206-
self.push_str(".Reader");
3197+
self.push_str(".ReadCloser");
32073198
}
32083199
Some(ty) => {
32093200
let wrpc = self.deps.wrpc();
32103201
self.push_str(wrpc);
3211-
self.push_str(".Receiver[");
3202+
self.push_str(".ReceiveCloser[");
32123203
self.print_list(ty);
32133204
self.push_str("]");
32143205
}

‎examples/go/hello-server/bindings/exports/wrpc_examples/hello/handler/bindings.wrpc.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) {
8484
for index, write := range writes {
8585
w, err := w.Index(index)
8686
if err != nil {
87-
slog.ErrorContext(ctx, "failed to index writer", "index", index, "instance", "wrpc-examples:hello/handler", "name", "hello", "err", err)
87+
slog.ErrorContext(ctx, "failed to index result writer", "index", index, "instance", "wrpc-examples:hello/handler", "name", "hello", "err", err)
8888
return
8989
}
9090
index := index

‎examples/go/resources-server/bindings/exports/wrpc_examples/resources/resources/bindings.wrpc.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) {
8989
for index, write := range writes {
9090
w, err := w.Index(index)
9191
if err != nil {
92-
slog.ErrorContext(ctx, "failed to index writer", "index", index, "instance", "wrpc-examples:resources/resources", "name", "foo", "err", err)
92+
slog.ErrorContext(ctx, "failed to index result writer", "index", index, "instance", "wrpc-examples:resources/resources", "name", "foo", "err", err)
9393
return
9494
}
9595
index := index
@@ -206,7 +206,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) {
206206
for index, write := range writes {
207207
w, err := w.Index(index)
208208
if err != nil {
209-
slog.ErrorContext(ctx, "failed to index writer", "index", index, "instance", "wrpc-examples:resources/resources", "name", "foo.foo", "err", err)
209+
slog.ErrorContext(ctx, "failed to index result writer", "index", index, "instance", "wrpc-examples:resources/resources", "name", "foo.foo", "err", err)
210210
return
211211
}
212212
index := index
@@ -323,7 +323,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) {
323323
for index, write := range writes {
324324
w, err := w.Index(index)
325325
if err != nil {
326-
slog.ErrorContext(ctx, "failed to index writer", "index", index, "instance", "wrpc-examples:resources/resources", "name", "foo.bar", "err", err)
326+
slog.ErrorContext(ctx, "failed to index result writer", "index", index, "instance", "wrpc-examples:resources/resources", "name", "foo.bar", "err", err)
327327
return
328328
}
329329
index := index
@@ -440,7 +440,7 @@ func ServeInterface(s wrpc.Server, h Handler) (stop func() error, err error) {
440440
for index, write := range writes {
441441
w, err := w.Index(index)
442442
if err != nil {
443-
slog.ErrorContext(ctx, "failed to index writer", "index", index, "instance", "wrpc-examples:resources/resources", "name", "bar", "err", err)
443+
slog.ErrorContext(ctx, "failed to index result writer", "index", index, "instance", "wrpc-examples:resources/resources", "name", "bar", "err", err)
444444
return
445445
}
446446
index := index

‎examples/go/streams-client/app.go‎

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,11 @@ func (s *ThrottleStream[T]) Receive() ([]T, error) {
4141
return []T{v}, nil
4242
}
4343

44+
func (s *ThrottleStream[T]) Close() error {
45+
s.tick.Stop()
46+
return nil
47+
}
48+
4449
type ThrottleReader struct {
4550
tick *time.Ticker
4651
buf []byte
@@ -57,6 +62,11 @@ func (s *ThrottleReader) Read(p []byte) (int, error) {
5762
return 1, nil
5863
}
5964

65+
func (s *ThrottleReader) Close() error {
66+
s.tick.Stop()
67+
return nil
68+
}
69+
6070
func Run(prefix string, client wrpc.Invoker) error {
6171
numbers, bytes, errCh, err := handler.Echo(context.Background(), client, &handler.Req{
6272
Numbers: &ThrottleStream[uint64]{
@@ -77,9 +87,7 @@ func Run(prefix string, client wrpc.Invoker) error {
7787
wg.Add(1)
7888
go func() {
7989
defer wg.Done()
80-
if c, ok := numbers.(io.Closer); ok {
81-
defer c.Close()
82-
}
90+
defer numbers.Close()
8391
for {
8492
chunk, err := numbers.Receive()
8593
if err == io.EOF {
@@ -95,9 +103,7 @@ func Run(prefix string, client wrpc.Invoker) error {
95103
wg.Add(1)
96104
go func() {
97105
defer wg.Done()
98-
if c, ok := bytes.(io.Closer); ok {
99-
defer c.Close()
100-
}
106+
defer bytes.Close()
101107
var chunk [128]byte
102108
for {
103109
n, err := bytes.Read(chunk[:])

0 commit comments

Comments
 (0)