Merge pull request #9 from shykes/pr_out_beam_messages_can_attach_arbitrary_file_descriptors

beam.go

@@ -2,6 +2,7 @@ package beam
 
 import (
 	"errors"
+	"os"
 )
 
 type Sender interface {
@@ -16,6 +17,7 @@ type Receiver interface {
 type Message struct {
 	Name string
 	Args []string
+	Att  *os.File
 }
 
 const (

inmem/inmem_test.go

@@ -1,8 +1,11 @@
 package inmem
 
 import (
+	"fmt"
 	"github.com/docker/beam"
 	"github.com/dotcloud/docker/pkg/testutils"
+	"io/ioutil"
+	"os"
 	"testing"
 )
 
@@ -141,3 +144,42 @@ func TestSendNested(t *testing.T) {
 		}
 	})
 }
+
+func TestSendFile(t *testing.T) {
+	r, w := Pipe()
+	defer r.Close()
+	defer w.Close()
+	tmp, err := ioutil.TempFile("", "beam-test-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(tmp.Name())
+	fmt.Fprintf(tmp, "hello world\n")
+	tmp.Sync()
+	tmp.Seek(0, 0)
+	testutils.Timeout(t, func() {
+		go func() {
+			_, _, err := w.Send(&beam.Message{"file", []string{"path=" + tmp.Name()}, tmp}, 0)
+			if err != nil {
+				t.Fatal(err)
+			}
+		}()
+		msg, _, _, err := r.Receive(0)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if msg.Name != "file" {
+			t.Fatalf("%#v", msg)
+		}
+		if msg.Args[0] != "path="+tmp.Name() {
+			t.Fatalf("%#v", msg)
+		}
+		txt, err := ioutil.ReadAll(msg.Att)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if string(txt) != "hello world\n" {
+			t.Fatalf("%s\n", txt)
+		}
+	})
+}

unix/conn.go

@@ -51,6 +51,9 @@ func (c *Conn) Close() error {
 }
 
 func (c *Conn) Send(msg *beam.Message, mode int) (beam.Receiver, beam.Sender, error) {
+	if msg.Att != nil {
+		return nil, nil, fmt.Errorf("file attachment not yet implemented in unix transport")
+	}
 	parts := []string{msg.Name}
 	parts = append(parts, msg.Args...)
 	b := []byte(data.EncodeList(parts))
@@ -94,7 +97,7 @@ func (c *Conn) Receive(mode int) (*beam.Message, beam.Receiver, beam.Sender, err
 	if len(parts) == 0 {
 		return nil, nil, nil, fmt.Errorf("malformed message")
 	}
-	msg := &beam.Message{parts[0], parts[1:]}
+	msg := &beam.Message{Name: parts[0], Args: parts[1:]}
 
 	// Setup nested streams
 	var (

utils/hub_test.go

@@ -9,7 +9,7 @@ import (
 func TestHubSendEmpty(t *testing.T) {
 	hub := NewHub()
 	// Send to empty hub should silently drop
-	if r, w, err := hub.Send(&beam.Message{"hello", nil}, beam.R|beam.W); err != nil {
+	if r, w, err := hub.Send(&beam.Message{Name: "hello", Args: nil}, beam.R|beam.W); err != nil {
 		t.Fatal(err)
 	} else if r != nil {
 		t.Fatalf("%#v", r)
@@ -29,12 +29,12 @@ func TestHubSendOneHandler(t *testing.T) {
 	hub := NewHub()
 	defer hub.Close()
 	testutils.Timeout(t, func() {
-		in, _, err := hub.Send(&beam.Message{"register", nil}, beam.R)
+		in, _, err := hub.Send(&beam.Message{Name: "register", Args: nil}, beam.R)
 		if err != nil {
 			t.Fatal(err)
 		}
 		go func() {
-			if _, _, err := hub.Send(&beam.Message{"hello", nil}, 0); err != nil {
+			if _, _, err := hub.Send(&beam.Message{Name: "hello", Args: nil}, 0); err != nil {
 				t.Fatal(err)
 			}
 		}()

utils/stack_test.go

@@ -28,7 +28,7 @@ func TestStackWithPipe(t *testing.T) {
 				t.Fatalf("%#v", msg)
 			}
 		}()
-		_, _, err := s.Send(&beam.Message{"hello", []string{"wonderful", "world"}}, 0)
+		_, _, err := s.Send(&beam.Message{Name: "hello", Args: []string{"wonderful", "world"}}, 0)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -57,7 +57,7 @@ func TestStackWithPair(t *testing.T) {
 				t.Fatalf("%#v", msg)
 			}
 		}()
-		_, _, err := s.Send(&beam.Message{"hello", []string{"wonderful", "world"}}, 0)
+		_, _, err := s.Send(&beam.Message{Name: "hello", Args: []string{"wonderful", "world"}}, 0)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -91,9 +91,9 @@ func TestStackAdd(t *testing.T) {
 	if s.Len() != 2 {
 		t.Fatalf("%#v", beforeA)
 	}
-	s.Send(&beam.Message{"for b", nil}, 0)
-	beforeB.Send(&beam.Message{"for a", nil}, 0)
-	beforeA.Send(&beam.Message{"for nobody", nil}, 0)
+	s.Send(&beam.Message{Name: "for b", Args: nil}, 0)
+	beforeB.Send(&beam.Message{Name: "for a", Args: nil}, 0)
+	beforeA.Send(&beam.Message{Name: "for nobody", Args: nil}, 0)
 	if len(a) != 1 {
 		t.Fatalf("%#v", a)
 	}
@@ -113,7 +113,7 @@ func TestStackAddBad(t *testing.T) {
 		t.Fatalf("%#v", s)
 	}
 	r.Close()
-	if _, _, err := s.Send(&beam.Message{"for the buffer", nil}, 0); err != nil {
+	if _, _, err := s.Send(&beam.Message{Name: "for the buffer", Args: nil}, 0); err != nil {
 		t.Fatal(err)
 	}
 	if s.Len() != 1 {