github/containerd
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

Unwrap proto errors in streaming client

Browse Source 
Allows clients to properly detect context cancellation

Signed-off-by: Derek McGowan <derek@mcg.dev>
This commit is contained in:
Derek McGowan 2022-12-02 11:24:25 -08:00
parent 51195ad099
commit 8a25fa584f
No known key found for this signature in database
GPG Key ID: F58C5D0A4405ACDB
3 changed files with 29 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
pkg/transfer/streaming/stream.go
View File

@ -157,14 +157,14 @@ func ReceiveStream(ctx context.Context, stream streaming.Stream) io.Reader {
// check window update error after recv, stream may be complete // check window update error after recv, stream may be complete
if werr = stream.Send(any); werr == nil { if werr = stream.Send(any); werr == nil {
window += windowSize window += windowSize
} else if werr == io.EOF { } else if errors.Is(werr, io.EOF) {
// TODO: Why does send return EOF here // TODO: Why does send return EOF here
werr = nil werr = nil
} }
} }
any, err := stream.Recv() any, err := stream.Recv()
if err != nil { if err != nil {
if err == io.EOF { if errors.Is(err, io.EOF) {
err = nil err = nil
} else { } else {
err = fmt.Errorf("received failed: %w", err) err = fmt.Errorf("received failed: %w", err)

8
services/streaming/service.go
View File

@ -102,8 +102,12 @@ type serviceStream struct {
cc chan struct{} cc chan struct{}
} }
func (ss *serviceStream) Send(a typeurl.Any) error { func (ss *serviceStream) Send(a typeurl.Any) (err error) {
return errdefs.FromGRPC(ss.s.Send(protobuf.FromAny(a))) err = errdefs.FromGRPC(ss.s.Send(protobuf.FromAny(a)))
if !errors.Is(err, io.EOF) {
err = errdefs.FromGRPC(err)
}
return
} }
func (ss *serviceStream) Recv() (a typeurl.Any, err error) { func (ss *serviceStream) Recv() (a typeurl.Any, err error) {

25
transfer.go
View File

@ -18,9 +18,12 @@ package containerd
import ( import (
"context" "context"
"errors"
"io"
streamingapi "github.com/containerd/containerd/api/services/streaming/v1" streamingapi "github.com/containerd/containerd/api/services/streaming/v1"
transferapi "github.com/containerd/containerd/api/services/transfer/v1" transferapi "github.com/containerd/containerd/api/services/transfer/v1"
"github.com/containerd/containerd/errdefs"
"github.com/containerd/containerd/pkg/streaming" "github.com/containerd/containerd/pkg/streaming"
"github.com/containerd/containerd/pkg/transfer" "github.com/containerd/containerd/pkg/transfer"
"github.com/containerd/containerd/pkg/transfer/proxy" "github.com/containerd/containerd/pkg/transfer/proxy"
@ -56,11 +59,17 @@ func (sc *streamCreator) Create(ctx context.Context, id string) (streaming.Strea
} }
err = stream.Send(protobuf.FromAny(a)) err = stream.Send(protobuf.FromAny(a))
if err != nil { if err != nil {
if !errors.Is(err, io.EOF) {
err = errdefs.FromGRPC(err)
}
return nil, err return nil, err
} }
// Receive an ack that stream is init and ready // Receive an ack that stream is init and ready
if _, err = stream.Recv(); err != nil { if _, err = stream.Recv(); err != nil {
if !errors.Is(err, io.EOF) {
err = errdefs.FromGRPC(err)
}
return nil, err return nil, err
} }
@ -73,12 +82,20 @@ type clientStream struct {
s streamingapi.Streaming_StreamClient s streamingapi.Streaming_StreamClient
} }
func (cs *clientStream) Send(a typeurl.Any) error { func (cs *clientStream) Send(a typeurl.Any) (err error) {
return cs.s.Send(protobuf.FromAny(a)) err = cs.s.Send(protobuf.FromAny(a))
if !errors.Is(err, io.EOF) {
err = errdefs.FromGRPC(err)
}
return
} }
func (cs *clientStream) Recv() (typeurl.Any, error) { func (cs *clientStream) Recv() (a typeurl.Any, err error) {
return cs.s.Recv() a, err = cs.s.Recv()
if !errors.Is(err, io.EOF) {
err = errdefs.FromGRPC(err)
}
return
} }
func (cs *clientStream) Close() error { func (cs *clientStream) Close() error {