Update transfer packages

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

services/transfer/service.go

@@ -20,9 +20,15 @@ import (
 	"context"
 
 	transferapi "github.com/containerd/containerd/api/services/transfer/v1"
+	transferTypes "github.com/containerd/containerd/api/types/transfer"
 	"github.com/containerd/containerd/errdefs"
+	"github.com/containerd/containerd/log"
+	"github.com/containerd/containerd/pkg/streaming"
 	"github.com/containerd/containerd/pkg/transfer"
+	"github.com/containerd/containerd/pkg/transfer/plugins"
 	"github.com/containerd/containerd/plugin"
+	ptypes "github.com/containerd/containerd/protobuf/types"
+	"github.com/containerd/typeurl"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -35,13 +41,15 @@ func init() {
 		ID:   "transfer",
 		Requires: []plugin.Type{
 			plugin.TransferPlugin,
+			plugin.StreamingPlugin,
 		},
 		InitFn: newService,
 	})
 }
 
 type service struct {
-	transferers []transfer.Transferer
+	transferers   []transfer.Transferer
+	streamManager streaming.StreamManager
 	transferapi.UnimplementedTransferServer
 }
 
@@ -50,6 +58,7 @@ func newService(ic *plugin.InitContext) (interface{}, error) {
 	if err != nil {
 		return nil, err
 	}
+
 	// TODO: how to determine order?
 	t := make([]transfer.Transferer, 0, len(plugins))
 	for _, p := range plugins {
@@ -59,8 +68,13 @@ func newService(ic *plugin.InitContext) (interface{}, error) {
 		}
 		t = append(t, i.(transfer.Transferer))
 	}
+	sp, err := ic.GetByID(plugin.StreamingPlugin, "manager")
+	if err != nil {
+		return nil, err
+	}
 	return &service{
-		transferers: t,
+		transferers:   t,
+		streamManager: sp.(streaming.StreamManager),
 	}, nil
 }
 
@@ -70,15 +84,75 @@ func (s *service) Register(gs *grpc.Server) error {
 }
 
 func (s *service) Transfer(ctx context.Context, req *transferapi.TransferRequest) (*emptypb.Empty, error) {
-	// TODO: Optionally proxy
+	var transferOpts []transfer.Opt
+	if req.Options != nil {
+		if req.Options.ProgressStream != "" {
+			stream, err := s.streamManager.Get(ctx, req.Options.ProgressStream)
+			if err != nil {
+				return nil, errdefs.ToGRPC(err)
+			}
+			defer stream.Close()
+
+			pf := func(p transfer.Progress) {
+				any, err := typeurl.MarshalAny(&transferTypes.Progress{
+					Event:    p.Event,
+					Name:     p.Name,
+					Parents:  p.Parents,
+					Progress: p.Progress,
+					Total:    p.Total,
+				})
+				if err != nil {
+					log.G(ctx).WithError(err).Warnf("event could not be marshaled: %v/%v", p.Event, p.Name)
+					return
+				}
+				if err := stream.Send(any); err != nil {
+					log.G(ctx).WithError(err).Warnf("event not sent: %v/%v", p.Event, p.Name)
+					return
+				}
+			}
+
+			transferOpts = append(transferOpts, transfer.WithProgress(pf))
+		}
+	}
+	src, err := s.convertAny(ctx, req.Source)
+	if err != nil {
+		return nil, errdefs.ToGRPC(err)
+	}
+	dst, err := s.convertAny(ctx, req.Destination)
+	plugins.ResolveType(req.Source)
+	if err != nil {
+		return nil, errdefs.ToGRPC(err)
+	}
 
-	// TODO: Convert options
 	for _, t := range s.transferers {
-		if err := t.Transfer(ctx, req.Source, req.Destination); err == nil {
-			return nil, nil
+		if err := t.Transfer(ctx, src, dst, transferOpts...); err == nil {
+			return &ptypes.Empty{}, nil
 		} else if !errdefs.IsNotImplemented(err) {
 			return nil, errdefs.ToGRPC(err)
 		}
 	}
 	return nil, status.Errorf(codes.Unimplemented, "method Transfer not implemented for %s to %s", req.Source.GetTypeUrl(), req.Destination.GetTypeUrl())
 }
+
+func (s *service) convertAny(ctx context.Context, a typeurl.Any) (interface{}, error) {
+	obj, err := plugins.ResolveType(a)
+	if err != nil {
+		if errdefs.IsNotFound(err) {
+			return typeurl.UnmarshalAny(a)
+		}
+		return nil, err
+	}
+	switch v := obj.(type) {
+	case streamUnmarshaler:
+		err = v.UnmarshalAny(ctx, s.streamManager, a)
+		return obj, err
+	default:
+		log.G(ctx).Debug("unmarshling to..")
+		err = typeurl.UnmarshalTo(a, obj)
+		return obj, err
+	}
+}
+
+type streamUnmarshaler interface {
+	UnmarshalAny(context.Context, streaming.StreamGetter, typeurl.Any) error
+}