summaryrefslogtreecommitdiff
path: root/pkg/varlinkapi
diff options
context:
space:
mode:
Diffstat (limited to 'pkg/varlinkapi')
-rw-r--r--pkg/varlinkapi/attach.go5
-rw-r--r--pkg/varlinkapi/images.go67
-rw-r--r--pkg/varlinkapi/transfers.go11
-rw-r--r--pkg/varlinkapi/virtwriter/virtwriter.go106
4 files changed, 124 insertions, 65 deletions
diff --git a/pkg/varlinkapi/attach.go b/pkg/varlinkapi/attach.go
index 2234899a5..8051f07be 100644
--- a/pkg/varlinkapi/attach.go
+++ b/pkg/varlinkapi/attach.go
@@ -60,7 +60,10 @@ func (i *LibpodAPI) Attach(call iopodman.VarlinkCall, name string, detachKeys st
if !start && state != libpod.ContainerStateRunning {
return call.ReplyErrorOccurred("container must be running to attach")
}
- call.Reply(nil)
+
+ // ACK the client upgrade request
+ call.ReplyAttach()
+
reader, writer, _, pw, streams := setupStreams(call)
go func() {
diff --git a/pkg/varlinkapi/images.go b/pkg/varlinkapi/images.go
index fa1a0a109..1abc4f086 100644
--- a/pkg/varlinkapi/images.go
+++ b/pkg/varlinkapi/images.go
@@ -371,7 +371,6 @@ func (i *LibpodAPI) PushImage(call iopodman.VarlinkCall, name, tag string, compr
done = true
default:
if !call.WantsMore() {
- time.Sleep(1 * time.Second)
break
}
br := iopodman.MoreResponse{
@@ -495,6 +494,9 @@ func (i *LibpodAPI) DeleteUnusedImages(call iopodman.VarlinkCall) error {
// Commit ...
func (i *LibpodAPI) Commit(call iopodman.VarlinkCall, name, imageName string, changes []string, author, message string, pause bool, manifestType string) error {
+ var newImage *image.Image
+
+ output := bytes.NewBuffer([]byte{})
ctr, err := i.Runtime.LookupContainer(name)
if err != nil {
return call.ReplyContainerNotFound(name, err.Error())
@@ -515,7 +517,7 @@ func (i *LibpodAPI) Commit(call iopodman.VarlinkCall, name, imageName string, ch
}
coptions := buildah.CommitOptions{
SignaturePolicyPath: rtc.SignaturePolicyPath,
- ReportWriter: nil,
+ ReportWriter: output,
SystemContext: sc,
PreferredManifestType: mimeType,
}
@@ -527,11 +529,61 @@ func (i *LibpodAPI) Commit(call iopodman.VarlinkCall, name, imageName string, ch
Author: author,
}
- newImage, err := ctr.Commit(getContext(), imageName, options)
- if err != nil {
- return call.ReplyErrorOccurred(err.Error())
+ if call.WantsMore() {
+ call.Continues = true
}
- return call.ReplyCommit(newImage.ID())
+
+ c := make(chan error)
+
+ go func() {
+ newImage, err = ctr.Commit(getContext(), imageName, options)
+ if err != nil {
+ c <- err
+ }
+ c <- nil
+ close(c)
+ }()
+
+ var log []string
+ done := false
+ for {
+ line, err := output.ReadString('\n')
+ if err == nil {
+ log = append(log, line)
+ continue
+ } else if err == io.EOF {
+ select {
+ case err := <-c:
+ if err != nil {
+ logrus.Errorf("reading of output during commit failed for %s", name)
+ return call.ReplyErrorOccurred(err.Error())
+ }
+ done = true
+ default:
+ if !call.WantsMore() {
+ break
+ }
+ br := iopodman.MoreResponse{
+ Logs: log,
+ }
+ call.ReplyCommit(br)
+ log = []string{}
+ }
+ } else {
+ return call.ReplyErrorOccurred(err.Error())
+ }
+ if done {
+ break
+ }
+ }
+ call.Continues = false
+
+ br := iopodman.MoreResponse{
+ Logs: log,
+ Id: newImage.ID(),
+ }
+
+ return call.ReplyCommit(br)
}
// ImportImage imports an image from a tarball to the image store
@@ -633,7 +685,6 @@ func (i *LibpodAPI) PullImage(call iopodman.VarlinkCall, name string) error {
done = true
default:
if !call.WantsMore() {
- time.Sleep(1 * time.Second)
break
}
br := iopodman.MoreResponse{
@@ -764,7 +815,6 @@ func (i *LibpodAPI) ImageSave(call iopodman.VarlinkCall, options iopodman.ImageS
done = true
default:
if !call.WantsMore() {
- time.Sleep(1 * time.Second)
break
}
br := iopodman.MoreResponse{
@@ -844,7 +894,6 @@ func (i *LibpodAPI) LoadImage(call iopodman.VarlinkCall, name, inputFile string,
done = true
default:
if !call.WantsMore() {
- time.Sleep(1 * time.Second)
break
}
br := iopodman.MoreResponse{
diff --git a/pkg/varlinkapi/transfers.go b/pkg/varlinkapi/transfers.go
index 96f76bcdc..24a91a86f 100644
--- a/pkg/varlinkapi/transfers.go
+++ b/pkg/varlinkapi/transfers.go
@@ -29,6 +29,12 @@ func (i *LibpodAPI) SendFile(call iopodman.VarlinkCall, ftype string, length int
return call.ReplyErrorOccurred(err.Error())
}
+ // FIXME return parameter
+ if err = call.ReplySendFile("FIXME_file_handle"); err != nil {
+ // If an error occurs while sending the reply, return the error
+ return err
+ }
+
writer := bufio.NewWriter(outputFile)
defer writer.Flush()
@@ -60,9 +66,10 @@ func (i *LibpodAPI) ReceiveFile(call iopodman.VarlinkCall, filepath string, dele
}
// Send the file length down to client
- // Varlink connection upraded
+ // Varlink connection upgraded
if err = call.ReplyReceiveFile(fileInfo.Size()); err != nil {
- return call.ReplyErrorOccurred(err.Error())
+ // If an error occurs while sending the reply, return the error
+ return err
}
reader := bufio.NewReader(fs)
diff --git a/pkg/varlinkapi/virtwriter/virtwriter.go b/pkg/varlinkapi/virtwriter/virtwriter.go
index 3adaf6e17..e747984c7 100644
--- a/pkg/varlinkapi/virtwriter/virtwriter.go
+++ b/pkg/varlinkapi/virtwriter/virtwriter.go
@@ -91,65 +91,65 @@ func (v VirtWriteCloser) Write(input []byte) (int, error) {
// Reader decodes the content that comes over the wire and directs it to the proper destination.
func Reader(r *bufio.Reader, output, errput *os.File, input *io.PipeWriter, resize chan remotecommand.TerminalSize) error {
- var saveb []byte
- var eom int
+ var messageSize int64
+ headerBytes := make([]byte, 8)
+
for {
- readb := make([]byte, 32*1024)
- n, err := r.Read(readb)
- // TODO, later may be worth checking in len of the read is 0
+ n, err := io.ReadFull(r, headerBytes)
if err != nil {
return err
}
- b := append(saveb, readb[0:n]...)
- // no sense in reading less than the header len
- for len(b) > 7 {
- eom = int(binary.BigEndian.Uint32(b[4:8])) + 8
- // The message and header are togther
- if len(b) >= eom {
- out := append([]byte{}, b[8:eom]...)
-
- switch IntToSocketDest(int(b[0])) {
- case ToStdout:
- n, err := output.Write(out)
- if err != nil {
- return err
- }
- if n < len(out) {
- return errors.New("short write error occurred on stdout")
- }
- case ToStderr:
- n, err := errput.Write(out)
- if err != nil {
- return err
- }
- if n < len(out) {
- return errors.New("short write error occurred on stderr")
- }
- case ToStdin:
- n, err := input.Write(out)
- if err != nil {
- return err
- }
- if n < len(out) {
- return errors.New("short write error occurred on stdin")
- }
- case TerminalResize:
- // Resize events come over in bytes, need to be reserialized
- resizeEvent := remotecommand.TerminalSize{}
- if err := json.Unmarshal(out, &resizeEvent); err != nil {
- return err
- }
- resize <- resizeEvent
- case Quit:
- return nil
+ if n < 8 {
+ return errors.New("short read and no full header read")
+ }
+
+ messageSize = int64(binary.BigEndian.Uint32(headerBytes[4:8]))
+
+ switch IntToSocketDest(int(headerBytes[0])) {
+ case ToStdout:
+ _, err := io.CopyN(output, r, messageSize)
+ if err != nil {
+ return err
+ }
+ case ToStderr:
+ _, err := io.CopyN(errput, r, messageSize)
+ if err != nil {
+ return err
+ }
+ case ToStdin:
+ _, err := io.CopyN(input, r, messageSize)
+ if err != nil {
+ return err
+ }
+ case TerminalResize:
+ out := make([]byte, messageSize)
+ if messageSize > 0 {
+ _, err = io.ReadFull(r, out)
+
+ if err != nil {
+ return err
}
- b = b[eom:]
- } else {
- // We do not have the header and full message, need to slurp again
- saveb = b
- break
}
+ // Resize events come over in bytes, need to be reserialized
+ resizeEvent := remotecommand.TerminalSize{}
+ if err := json.Unmarshal(out, &resizeEvent); err != nil {
+ return err
+ }
+ resize <- resizeEvent
+ case Quit:
+ out := make([]byte, messageSize)
+ if messageSize > 0 {
+ _, err = io.ReadFull(r, out)
+
+ if err != nil {
+ return err
+ }
+ }
+ return nil
+
+ default:
+ // Something really went wrong
+ return errors.New("Unknown multiplex destination")
}
}
- return nil
}
generated by cgit v1.2.3-54-g00ecf (git 2.39.0) at 2024-11-01 18:20:29 +0000