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

Add streaming proxy

Browse Source 
Signed-off-by: Derek McGowan <derek@mcg.dev>

Signed-off-by: Derek McGowan <derek@mcg.dev>
This commit is contained in:
Derek McGowan
2024-05-02 16:57:17 -07:00
parent 193af78800
commit ec04e4f638
2 changed files with 130 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
client/transfer.go
View File

@@ -18,17 +18,12 @@ package client
import (
"context"
"errors"
"io"
streamingapi "github.com/containerd/containerd/api/services/streaming/v1"
transferapi "github.com/containerd/containerd/api/services/transfer/v1"
"github.com/containerd/containerd/v2/core/streaming"
streamproxy "github.com/containerd/containerd/v2/core/streaming/proxy"
"github.com/containerd/containerd/v2/core/transfer"
"github.com/containerd/containerd/v2/core/transfer/proxy"
"github.com/containerd/containerd/v2/pkg/protobuf"
"github.com/containerd/errdefs"
"github.com/containerd/typeurl/v2"
)
func (c *Client) Transfer(ctx context.Context, src interface{}, dest interface{}, opts ...transfer.Opt) error {
@@ -42,68 +37,5 @@ func (c *Client) Transfer(ctx context.Context, src interface{}, dest interface{}
}
func (c *Client) streamCreator() streaming.StreamCreator {
return &streamCreator{
client: streamingapi.NewStreamingClient(c.conn),
}
}
type streamCreator struct {
client streamingapi.StreamingClient
}
func (sc *streamCreator) Create(ctx context.Context, id string) (streaming.Stream, error) {
stream, err := sc.client.Stream(ctx)
if err != nil {
return nil, err
}
a, err := typeurl.MarshalAny(&streamingapi.StreamInit{
ID: id,
})
if err != nil {
return nil, err
}
err = stream.Send(protobuf.FromAny(a))
if err != nil {
if !errors.Is(err, io.EOF) {
err = errdefs.FromGRPC(err)
}
return nil, err
}
// Receive an ack that stream is init and ready
if _, err = stream.Recv(); err != nil {
if !errors.Is(err, io.EOF) {
err = errdefs.FromGRPC(err)
}
return nil, err
}
return &clientStream{
s: stream,
}, nil
}
type clientStream struct {
s streamingapi.Streaming_StreamClient
}
func (cs *clientStream) Send(a typeurl.Any) (err error) {
err = cs.s.Send(protobuf.FromAny(a))
if !errors.Is(err, io.EOF) {
err = errdefs.FromGRPC(err)
}
return
}
func (cs *clientStream) Recv() (a typeurl.Any, err error) {
a, err = cs.s.Recv()
if !errors.Is(err, io.EOF) {
err = errdefs.FromGRPC(err)
}
return
}
func (cs *clientStream) Close() error {
return cs.s.CloseSend()
return streamproxy.NewStreamCreator(c.conn)
}