Reduce libchan interface in favor using Pipe for nesting

Signed-off-by: Derek McGowan <derek@mcgstyle.net> (github: dmcgowan)

Author: dmcgowan

copy_test.go

@@ -28,17 +28,14 @@ func TestChannelProxy(t *testing.T) {
 	}
 	client := func(t *testing.T, sender Sender) {
 		for i, m := range messages {
-			nestedReceiver, remoteSender, err := sender.CreateNestedReceiver()
-			if err != nil {
-				t.Fatalf("Error creating nested receiver: %s", err)
-			}
+			nestedReceiver, remoteSender := Pipe()
 
 			message := &ProxiedMessage{
 				Message: m,
 				Ret:     remoteSender,
 			}
 
-			err = sender.Send(message)
+			err := sender.Send(message)
 			if err != nil {
 				t.Fatalf("Error sending message: %s", err)
 			}
@@ -167,10 +164,6 @@ func SpawnProxyTest(t *testing.T, client SendTestRoutine, server ReceiveTestRout
 	go func() {
 		defer close(endServer)
 		server(t, receiver2)
-		err := receiver2.Close()
-		if err != nil {
-			t.Errorf("Error closing receiver: %s", err)
-		}
 	}()
 
 	timeout := time.After(RoutineTimeout)

inmem_test.go

@@ -19,13 +19,10 @@ type InMemMessage struct {
 
 func TestInmemRetPipe(t *testing.T) {
 	client := func(t *testing.T, w Sender) {
-		ret, retPipe, err := w.CreateNestedReceiver()
-		if err != nil {
-			t.Fatal(err)
-		}
+		ret, retPipe := Pipe()
 		message := &InMemMessage{Data: "hello", Ret: retPipe}
 
-		err = w.Send(message)
+		err := w.Send(message)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -85,12 +82,9 @@ func TestSimpleSend(t *testing.T) {
 
 func TestSendReply(t *testing.T) {
 	client := func(t *testing.T, w Sender) {
-		ret, retPipe, err := w.CreateNestedReceiver()
-		if err != nil {
-			t.Fatal(err)
-		}
+		ret, retPipe := Pipe()
 		message := &InMemMessage{Data: "this is the request", Ret: retPipe}
-		err = w.Send(message)
+		err := w.Send(message)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -181,14 +175,8 @@ type SimpleMessage struct {
 
 func TestComplexMessage(t *testing.T) {
 	client := func(t *testing.T, w Sender) {
-		send, remoteRecv, err := w.CreateNestedSender()
-		if err != nil {
-			t.Fatalf("Error creating sender: %s", err)
-		}
-		recv, remoteSend, err := w.CreateNestedReceiver()
-		if err != nil {
-			t.Fatalf("Error creating sender: %s", err)
-		}
+		remoteRecv, send := Pipe()
+		recv, remoteSend := Pipe()
 		bs, bsRemote := net.Pipe()
 
 		m1 := &ComplexMessage{
@@ -350,10 +338,6 @@ func SpawnPipeTestRoutines(t *testing.T, s SendTestRoutine, r ReceiveTestRoutine
 	go func() {
 		defer close(end2)
 		r(t, receiver)
-		err := receiver.Close()
-		if err != nil {
-			t.Fatalf("Error closing receiver: %s", err)
-		}
 	}()
 
 	timeout := time.After(RoutineTimeout)

libchan.go

@@ -22,16 +22,6 @@ type Sender interface {
 
 	// Close closes the channel.
 	Close() error
-
-	// CreatePipeReceiver creates a receive-only pipe.  The sender
-	// should be sent across the channel, any call to send directly
-	// will throw an error on any channel on a transport.
-	CreateNestedReceiver() (Receiver, Sender, error)
-
-	// CreateNestedSender creates a send-only pipe.  The receiver
-	// should be sent across the channel, any call to receive directly
-	// will throw an error on any channel on a transport.
-	CreateNestedSender() (Sender, Receiver, error)
 }
 
 // Receiver is a channel which can receive messages of any
@@ -46,9 +36,4 @@ type Receiver interface {
 	// the receive object is incompatible, Receiver will
 	// throw an error.
 	Receive(message interface{}) error
-
-	// Close closes the channel.  Closing does not keep the remote
-	// sender from sending messages.  Normally a receiver can be
-	// automatically closed through receiving an EOF.
-	Close() error
 }

spdy/pipe.go

@@ -17,7 +17,7 @@ type pipeReceiver struct {
 	receiver libchan.Receiver
 }
 
-// Pipe creates a channel pipe using an in memory transport.
+// Pipe creates a top-level channel pipe using an in memory transport.
 func Pipe() (libchan.Sender, libchan.Receiver, error) {
 	c1, c2 := net.Pipe()
 
@@ -72,22 +72,6 @@ func (p *pipeSender) CreateByteStream() (io.ReadWriteCloser, error) {
 	return p.session.createByteStream()
 }
 
-func (p *pipeSender) CreateNestedReceiver() (libchan.Receiver, libchan.Sender, error) {
-	return p.sender.CreateNestedReceiver()
-}
-
-func (p *pipeSender) CreateNestedSender() (libchan.Sender, libchan.Receiver, error) {
-	return p.sender.CreateNestedSender()
-}
-
 func (p *pipeReceiver) Receive(message interface{}) error {
 	return p.receiver.Receive(message)
 }
-
-func (p *pipeReceiver) Close() error {
-	err := p.receiver.Close()
-	if err != nil {
-		return err
-	}
-	return p.session.Close()
-}

spdy/pipe_test.go

@@ -25,10 +25,7 @@ func TestSendFirstPipe(t *testing.T) {
 	client := func(t *testing.T, sender libchan.Sender) {
 		bs, bsRemote := net.Pipe()
 
-		nestedReceiver, remoteSender, err := sender.CreateNestedReceiver()
-		if err != nil {
-			t.Fatalf("Error creating nested receiver: %s", err)
-		}
+		nestedReceiver, remoteSender := libchan.Pipe()
 
 		m1 := &PipeMessage{
 			Message: message1,
@@ -63,11 +60,6 @@ func TestSendFirstPipe(t *testing.T) {
 		if string(readData) != message4 {
 			t.Fatalf("Unexpected message\n\tExpected: %s\n\tActual: %s", message4, string(readData))
 		}
-
-		closeErr := nestedReceiver.Close()
-		if closeErr != nil {
-			t.Fatalf("Error closing nested receiver: %s", closeErr)
-		}
 	}
 	server := func(t *testing.T, receiver libchan.Receiver) {
 		m1 := &PipeMessage{}
@@ -132,10 +124,6 @@ func SpawnPipeTest(t *testing.T, client PipeSenderRoutine, server PipeReceiverRo
 	go func() {
 		defer close(endServer)
 		server(t, receiver)
-		err := receiver.Close()
-		if err != nil {
-			t.Fatalf("Error closing receiver: %s", err)
-		}
 	}()
 
 	timeout := time.After(ClientServerTimeout)

spdy/proxy_test.go

@@ -30,17 +30,14 @@ func TestChannelProxy(t *testing.T) {
 	}
 	client := func(t *testing.T, sender libchan.Sender) {
 		for i, m := range messages {
-			nestedReceiver, remoteSender, err := sender.CreateNestedReceiver()
-			if err != nil {
-				t.Fatalf("Error creating nested receiver: %s", err)
-			}
+			nestedReceiver, remoteSender := libchan.Pipe()
 
 			message := &ProxiedMessage{
 				Message: m,
 				Ret:     remoteSender,
 			}
 
-			err = sender.Send(message)
+			err := sender.Send(message)
 			if err != nil {
 				t.Fatalf("Error sending message: %s", err)
 			}
@@ -173,10 +170,6 @@ func SpawnProxyTest(t *testing.T, client PipeSenderRoutine, server PipeReceiverR
 	go func() {
 		defer close(endServer)
 		server(t, receiver2)
-		err := receiver2.Close()
-		if err != nil {
-			t.Errorf("Error closing receiver: %s", err)
-		}
 	}()
 
 	timeout := time.After(ClientServerTimeout)

spdy/session_test.go

@@ -23,15 +23,8 @@ type SimpleMessage struct {
 
 func TestChannelEncoding(t *testing.T) {
 	client := func(t *testing.T, sender libchan.Sender, s *Transport) {
-		recv, s1, err1 := sender.CreateNestedReceiver()
-		if err1 != nil {
-			t.Fatalf("Error creating receive channel: %s", err1)
-		}
-
-		send, r1, err2 := sender.CreateNestedSender()
-		if err2 != nil {
-			t.Fatalf("Error creating send channel: %s", err2)
-		}
+		recv, s1 := libchan.Pipe()
+		r1, send := libchan.Pipe()
 
 		m1 := &InOutMessage{
 			Message: "WithInOut",
@@ -62,11 +55,6 @@ func TestChannelEncoding(t *testing.T) {
 		if closeErr != nil {
 			t.Fatalf("Error closing s1: %s", closeErr)
 		}
-
-		closeErr = recv.Close()
-		if closeErr != nil {
-			t.Fatalf("Error closing s1: %s", closeErr)
-		}
 	}
 	server := func(t *testing.T, receiver libchan.Receiver, s *Transport) {
 		m1 := &InOutMessage{}
@@ -102,12 +90,7 @@ func TestChannelEncoding(t *testing.T) {
 			t.Fatalf("Error sending SimpleMessage: %s", sendErr)
 		}
 
-		closeErr := m1.Recv.Close()
-		if closeErr != nil {
-			t.Fatalf("Error closing recv connection: %s", closeErr)
-		}
-
-		closeErr = m1.Send.Close()
+		closeErr := m1.Send.Close()
 		if closeErr != nil {
 			t.Fatalf("Error closing send connection: %s", closeErr)
 		}
@@ -122,10 +105,7 @@ type AbstractionMessage struct {
 
 func TestChannelAbstraction(t *testing.T) {
 	client := func(t *testing.T, sender libchan.Sender, s *Transport) {
-		send, recv, channelErr := sender.CreateNestedSender()
-		if channelErr != nil {
-			t.Fatalf("Error creating sub channel: %s", channelErr)
-		}
+		recv, send := libchan.Pipe()
 
 		m1 := &AbstractionMessage{
 			Message: "irrelevant content",
@@ -152,61 +132,6 @@ func TestChannelAbstraction(t *testing.T) {
 		if expected := "irrelevant content"; m1.Message != expected {
 			t.Fatalf("Unexpected message value:\n\tExpected: %s\n\tActual: %s", expected, m1.Message)
 		}
-
-		closeErr := m1.Channel.(libchan.Receiver).Close()
-		if closeErr != nil {
-			t.Fatalf("Error closing channel: %s", closeErr)
-		}
-	}
-	SpawnClientServerTest(t, "localhost:12943", ClientSendWrapper(client), ServerReceiveWrapper(server))
-}
-
-type MessageWithInput struct {
-	Message string
-	Input   libchan.Receiver
-}
-
-func TestBadDirection(t *testing.T) {
-	client := func(t *testing.T, sender libchan.Sender, s *Transport) {
-		send, recv, channelErr := sender.CreateNestedSender()
-		if channelErr != nil {
-			t.Fatalf("Error creating sub channel: %s", channelErr)
-		}
-
-		m1 := &MessageWithInput{
-			Message: "contentless",
-			Input:   recv,
-		}
-
-		sendErr := sender.Send(m1)
-		if sendErr != nil {
-			t.Fatalf("Error sending channel: %s", sendErr)
-		}
-
-		m2 := &SimpleMessage{"Supposedly input"}
-		commErr := recv.Receive(m2)
-		if commErr == nil {
-			t.Fatalf("No error receiving on sender")
-		} else if commErr != ErrWrongDirection {
-			t.Fatalf("Error receiving message: %s", commErr)
-		}
-
-		closeErr := send.Close()
-		if closeErr != nil {
-			t.Fatalf("Error closing sender: %s", closeErr)
-		}
-	}
-	server := func(t *testing.T, receiver libchan.Receiver, s *Transport) {
-		m1 := &MessageWithInput{}
-		recvErr := receiver.Receive(m1)
-		if recvErr != nil {
-			t.Fatalf("Error receiving message: %s", recvErr)
-		}
-
-		closeErr := m1.Input.Close()
-		if closeErr != nil {
-			t.Fatalf("Error closing channel: %s", closeErr)
-		}
 	}
 	SpawnClientServerTest(t, "localhost:12943", ClientSendWrapper(client), ServerReceiveWrapper(server))
 }
@@ -222,7 +147,6 @@ func TestByteStream(t *testing.T) {
 		if bsErr != nil {
 			t.Fatalf("Error creating byte stream: %s", bsErr)
 		}
-		//bs, bsRemote := net.Pipe()
 
 		m1 := &MessageWithByteStream{
 			Message: "with a byte stream",
@@ -410,12 +334,7 @@ func ServerReceiveWrapper(f func(t *testing.T, c libchan.Receiver, s *Transport)
 
 		f(t, receiver, session)
 
-		closeErr := receiver.Close()
-		if closeErr != nil {
-			t.Fatalf("Error closing receiver: %s", closeErr)
-		}
-
-		closeErr = session.Close()
+		closeErr := session.Close()
 		if closeErr != nil {
 			t.Fatalf("Error closing connection: %s", closeErr)
 		}