Uses streams internally for ipc Get/Put Content

This is done because libchan/spdystream does not currently support sending
serialzied objects of size larger than 16MB
See docker-archive-public/docker.libchan#65

Author: BrianBland

storagedriver/ipc/client.go

@@ -1,8 +1,10 @@
 package ipc
 
 import (
+	"bytes"
 	"encoding/json"
 	"io"
+	"io/ioutil"
 	"net"
 	"os"
 	"os/exec"
@@ -116,7 +118,7 @@ func (driver *StorageDriverClient) GetContent(path string) ([]byte, error) {
 		return nil, err
 	}
 
-	var response GetContentResponse
+	var response ReadStreamResponse
 	err = receiver.Receive(&response)
 	if err != nil {
 		return nil, err
@@ -126,22 +128,26 @@ func (driver *StorageDriverClient) GetContent(path string) ([]byte, error) {
 		return nil, response.Error
 	}
 
-	return response.Content, nil
+	defer response.Reader.Close()
+	contents, err := ioutil.ReadAll(response.Reader)
+	if err != nil {
+		return nil, err
+	}
+	return contents, nil
 }
 
 func (driver *StorageDriverClient) PutContent(path string, contents []byte) error {
 	receiver, remoteSender := libchan.Pipe()
 
-	params := map[string]interface{}{"Path": path, "Contents": contents}
+	params := map[string]interface{}{"Path": path, "Reader": WrapReader(bytes.NewReader(contents))}
 	err := driver.sender.Send(&Request{Type: "PutContent", Parameters: params, ResponseChannel: remoteSender})
 	if err != nil {
 		return err
 	}
 
-	var response PutContentResponse
+	var response WriteStreamResponse
 	err = receiver.Receive(&response)
 	if err != nil {
-		panic(err)
 		return err
 	}
 
@@ -177,7 +183,7 @@ func (driver *StorageDriverClient) ReadStream(path string, offset uint64) (io.Re
 func (driver *StorageDriverClient) WriteStream(path string, offset, size uint64, reader io.ReadCloser) error {
 	receiver, remoteSender := libchan.Pipe()
 
-	params := map[string]interface{}{"Path": path, "Offset": offset, "Size": size, "Reader": WrapReadCloser(reader)}
+	params := map[string]interface{}{"Path": path, "Offset": offset, "Size": size, "Reader": WrapReader(reader)}
 	err := driver.sender.Send(&Request{Type: "WriteStream", Parameters: params, ResponseChannel: remoteSender})
 	if err != nil {
 		return err

storagedriver/ipc/ipc.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"reflect"
 
 	"github.com/docker/libchan"
@@ -23,8 +24,14 @@ func (r noWriteReadWriteCloser) Write(p []byte) (n int, err error) {
 	return 0, errors.New("Write unsupported")
 }
 
-func WrapReadCloser(readCloser io.ReadCloser) io.ReadWriteCloser {
-	return noWriteReadWriteCloser{readCloser}
+func WrapReader(reader io.Reader) io.ReadWriteCloser {
+	if readWriteCloser, ok := reader.(io.ReadWriteCloser); ok {
+		return readWriteCloser
+	} else if readCloser, ok := reader.(io.ReadCloser); ok {
+		return noWriteReadWriteCloser{readCloser}
+	} else {
+		return noWriteReadWriteCloser{ioutil.NopCloser(reader)}
+	}
 }
 
 type responseError struct {
@@ -46,15 +53,6 @@ func (err *responseError) Error() string {
 	return fmt.Sprintf("%s: %s", err.Type, err.Message)
 }
 
-type GetContentResponse struct {
-	Content []byte
-	Error   *responseError
-}
-
-type PutContentResponse struct {
-	Error *responseError
-}
-
 type ReadStreamResponse struct {
 	Reader io.ReadWriteCloser
 	Error  *responseError

storagedriver/ipc/server.go

@@ -1,7 +1,9 @@
 package ipc
 
 import (
+	"bytes"
 	"io"
+	"io/ioutil"
 	"net"
 	"os"
 
@@ -44,24 +46,29 @@ func receive(driver storagedriver.StorageDriver, receiver libchan.Receiver) {
 }
 
 func handleRequest(driver storagedriver.StorageDriver, request Request) {
-
 	switch request.Type {
 	case "GetContent":
 		path, _ := request.Parameters["Path"].(string)
 		content, err := driver.GetContent(path)
-		response := GetContentResponse{
-			Content: content,
-			Error:   ResponseError(err),
+		var response ReadStreamResponse
+		if err != nil {
+			response = ReadStreamResponse{Error: ResponseError(err)}
+		} else {
+			response = ReadStreamResponse{Reader: WrapReader(bytes.NewReader(content))}
 		}
 		err = request.ResponseChannel.Send(&response)
 		if err != nil {
 			panic(err)
 		}
 	case "PutContent":
 		path, _ := request.Parameters["Path"].(string)
-		contents, _ := request.Parameters["Contents"].([]byte)
-		err := driver.PutContent(path, contents)
-		response := PutContentResponse{
+		reader, _ := request.Parameters["Reader"].(io.ReadCloser)
+		contents, err := ioutil.ReadAll(reader)
+		defer reader.Close()
+		if err == nil {
+			err = driver.PutContent(path, contents)
+		}
+		response := WriteStreamResponse{
 			Error: ResponseError(err),
 		}
 		err = request.ResponseChannel.Send(&response)
@@ -82,7 +89,7 @@ func handleRequest(driver storagedriver.StorageDriver, request Request) {
 		if err != nil {
 			response = ReadStreamResponse{Error: ResponseError(err)}
 		} else {
-			response = ReadStreamResponse{Reader: WrapReadCloser(reader)}
+			response = ReadStreamResponse{Reader: WrapReader(reader)}
 		}
 		err = request.ResponseChannel.Send(&response)
 		if err != nil {

storagedriver/testsuites/testsuites.go

@@ -128,7 +128,7 @@ func (suite *DriverSuite) TestWriteReadStreams4(c *C) {
 func (suite *DriverSuite) TestContinueStreamAppend(c *C) {
 	filename := randomString(32)
 
-	chunkSize := uint64(32)
+	chunkSize := uint64(10 * 1024 * 1024)
 
 	contentsChunk1 := []byte(randomString(chunkSize))
 	contentsChunk2 := []byte(randomString(chunkSize))