164 lines
3.4 KiB
Go
164 lines
3.4 KiB
Go
package content
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
|
|
contentapi "github.com/docker/containerd/api/services/content"
|
|
"github.com/docker/containerd/content"
|
|
|
|
digest "github.com/opencontainers/go-digest"
|
|
"github.com/pkg/errors"
|
|
)
|
|
|
|
func NewIngesterFromClient(client contentapi.ContentClient) content.Ingester {
|
|
return &remoteIngester{
|
|
client: client,
|
|
}
|
|
}
|
|
|
|
type remoteIngester struct {
|
|
client contentapi.ContentClient
|
|
}
|
|
|
|
func (ri *remoteIngester) Writer(ctx context.Context, ref string, size int64, expected digest.Digest) (content.Writer, error) {
|
|
wrclient, offset, err := ri.negotiate(ctx, ref, size, expected)
|
|
if err != nil {
|
|
return nil, rewriteGRPCError(err)
|
|
}
|
|
|
|
return &remoteWriter{
|
|
client: wrclient,
|
|
offset: offset,
|
|
}, nil
|
|
}
|
|
|
|
func (ri *remoteIngester) negotiate(ctx context.Context, ref string, size int64, expected digest.Digest) (contentapi.Content_WriteClient, int64, error) {
|
|
wrclient, err := ri.client.Write(ctx)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
|
|
if err := wrclient.Send(&contentapi.WriteRequest{
|
|
Action: contentapi.WriteActionStat,
|
|
Ref: ref,
|
|
Total: size,
|
|
Expected: expected,
|
|
}); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
|
|
resp, err := wrclient.Recv()
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
|
|
return wrclient, resp.Offset, nil
|
|
}
|
|
|
|
type remoteWriter struct {
|
|
ref string
|
|
client contentapi.Content_WriteClient
|
|
offset int64
|
|
digest digest.Digest
|
|
}
|
|
|
|
func newRemoteWriter(client contentapi.Content_WriteClient, ref string, offset int64) (*remoteWriter, error) {
|
|
return &remoteWriter{
|
|
ref: ref,
|
|
client: client,
|
|
offset: offset,
|
|
}, nil
|
|
}
|
|
|
|
// send performs a synchronous req-resp cycle on the client.
|
|
func (rw *remoteWriter) send(req *contentapi.WriteRequest) (*contentapi.WriteResponse, error) {
|
|
if err := rw.client.Send(req); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
resp, err := rw.client.Recv()
|
|
|
|
if err == nil {
|
|
// try to keep these in sync
|
|
if resp.Digest != "" {
|
|
rw.digest = resp.Digest
|
|
}
|
|
}
|
|
|
|
return resp, err
|
|
}
|
|
|
|
func (rw *remoteWriter) Status() (content.Status, error) {
|
|
resp, err := rw.send(&contentapi.WriteRequest{
|
|
Action: contentapi.WriteActionStat,
|
|
})
|
|
if err != nil {
|
|
return content.Status{}, err
|
|
}
|
|
|
|
return content.Status{
|
|
Ref: rw.ref,
|
|
Offset: resp.Offset,
|
|
StartedAt: resp.StartedAt,
|
|
UpdatedAt: resp.UpdatedAt,
|
|
}, nil
|
|
}
|
|
|
|
func (rw *remoteWriter) Digest() digest.Digest {
|
|
return rw.digest
|
|
}
|
|
|
|
func (rw *remoteWriter) Write(p []byte) (n int, err error) {
|
|
offset := rw.offset
|
|
|
|
resp, err := rw.send(&contentapi.WriteRequest{
|
|
Action: contentapi.WriteActionWrite,
|
|
Offset: offset,
|
|
Data: p,
|
|
})
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
n = int(resp.Offset - offset)
|
|
if n < len(p) {
|
|
err = io.ErrShortWrite
|
|
}
|
|
|
|
rw.offset += int64(n)
|
|
return
|
|
}
|
|
|
|
func (rw *remoteWriter) Commit(size int64, expected digest.Digest) error {
|
|
resp, err := rw.send(&contentapi.WriteRequest{
|
|
Action: contentapi.WriteActionCommit,
|
|
Total: size,
|
|
Offset: rw.offset,
|
|
Expected: expected,
|
|
})
|
|
if err != nil {
|
|
return rewriteGRPCError(err)
|
|
}
|
|
|
|
if size != 0 && resp.Offset != size {
|
|
return errors.Errorf("unexpected size: %v != %v", resp.Offset, size)
|
|
}
|
|
|
|
if expected != "" && resp.Digest != expected {
|
|
return errors.Errorf("unexpected digest: %v != %v", resp.Digest, expected)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (rw *remoteWriter) Truncate(size int64) error {
|
|
// This truncation won't actually be validated until a write is issued.
|
|
rw.offset = size
|
|
return nil
|
|
}
|
|
|
|
func (rw *remoteWriter) Close() error {
|
|
return rw.client.CloseSend()
|
|
}
|