Move utilities and interfaces out of `inmem`

* `utils.Buffer`: a Sender which stores messages in a slice
* `utils.Copy`: beam equivalent of `io.Copy`
* `utlis.NopSender`: a Sender which silently discards messages
* `utlis.NopReceiver`: a Receiver which always returns EOF

beam.go

@@ -1,6 +1,8 @@
 package beam
 
-import ()
+import (
+	"errors"
+)
 
 type Sender interface {
 	Send(msg *Message, mode int) (Receiver, Sender, error)
@@ -20,3 +22,16 @@ const (
 	R = 1 << (32 - 1 - iota)
 	W
 )
+
+type ReceiverFrom interface {
+	ReceiveFrom(Receiver) (int, error)
+}
+
+type SenderTo interface {
+	SendTo(Sender) (int, error)
+}
+
+var (
+	ErrIncompatibleSender   = errors.New("incompatible sender")
+	ErrIncompatibleReceiver = errors.New("incompatible receiver")
+)

inmem/inmem.go

@@ -178,7 +178,7 @@ func (r *PipeReceiver) SendTo(dst beam.Sender) (int, error) {
 	// If the destination is a PipeSender, we can cheat
 	pdst, ok := dst.(*PipeSender)
 	if !ok {
-		return 0, ErrIncompatibleSender
+		return 0, beam.ErrIncompatibleSender
 	}
 	for {
 		pmsg, err := r.p.preceive()
@@ -231,7 +231,7 @@ func (w *PipeSender) ReceiveFrom(src beam.Receiver) (int, error) {
 	// If the destination is a PipeReceiver, we can cheat
 	psrc, ok := src.(*PipeReceiver)
 	if !ok {
-		return 0, ErrIncompatibleReceiver
+		return 0, beam.ErrIncompatibleReceiver
 	}
 	for {
 		pmsg, err := psrc.p.preceive()

utils/buf.go

@@ -1,4 +1,4 @@
-package inmem
+package utils
 
 import (
 	"github.com/docker/beam"

utils/copy.go

@@ -1,34 +1,20 @@
-package inmem
+package utils
 
 import (
-	"errors"
 	"github.com/docker/beam"
 	"sync"
 )
 
-type ReceiverFrom interface {
-	ReceiveFrom(beam.Receiver) (int, error)
-}
-
-type SenderTo interface {
-	SendTo(beam.Sender) (int, error)
-}
-
-var (
-	ErrIncompatibleSender   = errors.New("incompatible sender")
-	ErrIncompatibleReceiver = errors.New("incompatible receiver")
-)
-
 func Copy(dst beam.Sender, src beam.Receiver) (int, error) {
 	var tasks sync.WaitGroup
 	defer tasks.Wait()
-	if senderTo, ok := src.(SenderTo); ok {
-		if n, err := senderTo.SendTo(dst); err != ErrIncompatibleSender {
+	if senderTo, ok := src.(beam.SenderTo); ok {
+		if n, err := senderTo.SendTo(dst); err != beam.ErrIncompatibleSender {
 			return n, err
 		}
 	}
-	if receiverFrom, ok := dst.(ReceiverFrom); ok {
-		if n, err := receiverFrom.ReceiveFrom(src); err != ErrIncompatibleReceiver {
+	if receiverFrom, ok := dst.(beam.ReceiverFrom); ok {
+		if n, err := receiverFrom.ReceiveFrom(src); err != beam.ErrIncompatibleReceiver {
 			return n, err
 		}
 	}

utils/nop.go

@@ -1,4 +1,4 @@
-package inmem
+package utils
 
 import (
 	"github.com/docker/beam"

utils/stack_test.go

@@ -73,7 +73,7 @@ func TestStackLen(t *testing.T) {
 
 func TestStackAdd(t *testing.T) {
 	s := NewStackSender()
-	a := inmem.Buffer{}
+	a := Buffer{}
 	beforeA := s.Add(&a)
 	// Add on an empty StackSender should return an empty StackSender
 	if beforeA.Len() != 0 {
@@ -83,7 +83,7 @@ func TestStackAdd(t *testing.T) {
 		t.Fatalf("%#v", beforeA)
 	}
 	// Add a 2nd element
-	b := inmem.Buffer{}
+	b := Buffer{}
 	beforeB := s.Add(&b)
 	if beforeB.Len() != 1 {
 		t.Fatalf("%#v", beforeA)
@@ -105,7 +105,7 @@ func TestStackAdd(t *testing.T) {
 // Misbehaving backends must be removed
 func TestStackAddBad(t *testing.T) {
 	s := NewStackSender()
-	buf := inmem.Buffer{}
+	buf := Buffer{}
 	s.Add(&buf)
 	r, w := inmem.Pipe()
 	s.Add(w)