client: use stop function to deregister context.AfterFunc

Signed-off-by: Michael Smithhisler <smithhisler.mike@gmail.com>
This commit is contained in:
Michael Smithhisler
2026-01-05 07:32:53 -05:00
parent 92a240fd42
commit 06704ef904
4 changed files with 24 additions and 8 deletions

View File

@@ -45,12 +45,15 @@ func (r stream) Close() error {
// JSONMessages decodes the response stream as a sequence of JSONMessages.
// if stream ends or context is cancelled, the underlying [io.Reader] is closed.
func (r stream) JSONMessages(ctx context.Context) iter.Seq2[jsonstream.Message, error] {
context.AfterFunc(ctx, func() {
stop := context.AfterFunc(ctx, func() {
_ = r.Close()
})
dec := json.NewDecoder(r)
return func(yield func(jsonstream.Message, error) bool) {
defer r.Close()
defer func() {
stop() // unregister AfterFunc
r.Close()
}()
for {
var jm jsonstream.Message
err := dec.Decode(&jm)

View File

@@ -136,14 +136,19 @@ func newCancelReadCloser(ctx context.Context, rc io.ReadCloser) io.ReadCloser {
rc: rc,
close: sync.OnceValue(rc.Close),
}
context.AfterFunc(ctx, func() { _ = crc.Close() })
crc.stop = context.AfterFunc(ctx, func() { _ = crc.Close() })
return crc
}
type cancelReadCloser struct {
rc io.ReadCloser
close func() error
stop func() bool
}
func (c *cancelReadCloser) Read(p []byte) (int, error) { return c.rc.Read(p) }
func (c *cancelReadCloser) Close() error { return c.close() }
func (c *cancelReadCloser) Close() error {
c.stop() // unregister AfterFunc
return c.close()
}

View File

@@ -45,12 +45,15 @@ func (r stream) Close() error {
// JSONMessages decodes the response stream as a sequence of JSONMessages.
// if stream ends or context is cancelled, the underlying [io.Reader] is closed.
func (r stream) JSONMessages(ctx context.Context) iter.Seq2[jsonstream.Message, error] {
context.AfterFunc(ctx, func() {
stop := context.AfterFunc(ctx, func() {
_ = r.Close()
})
dec := json.NewDecoder(r)
return func(yield func(jsonstream.Message, error) bool) {
defer r.Close()
defer func() {
stop() // unregister AfterFunc
r.Close()
}()
for {
var jm jsonstream.Message
err := dec.Decode(&jm)

View File

@@ -136,14 +136,19 @@ func newCancelReadCloser(ctx context.Context, rc io.ReadCloser) io.ReadCloser {
rc: rc,
close: sync.OnceValue(rc.Close),
}
context.AfterFunc(ctx, func() { _ = crc.Close() })
crc.stop = context.AfterFunc(ctx, func() { _ = crc.Close() })
return crc
}
type cancelReadCloser struct {
rc io.ReadCloser
close func() error
stop func() bool
}
func (c *cancelReadCloser) Read(p []byte) (int, error) { return c.rc.Read(p) }
func (c *cancelReadCloser) Close() error { return c.close() }
func (c *cancelReadCloser) Close() error {
c.stop() // unregister AfterFunc
return c.close()
}