fixed issue with ChannelReadWriter for unbuffered channels.

Was a concurrency issue int he Write method that retained the
slice p that was passed in, making concurrency issues much more likely
with unbuffered channels.

pkg/comms/agentserver_test.go

@@ -4,8 +4,8 @@ import (
 	"context"
 	"git.wamblee.org/converge/pkg/testsupport"
 	"github.com/stretchr/testify/suite"
+	"io"
 	"log"
-	"net"
 	"net/http"
 	"testing"
 	"time"
@@ -18,8 +18,8 @@ type AgentServerTestSuite struct {
 	cancelFunc  context.CancelFunc
 	pprofServer *http.Server
 
-	agentConnection  net.Conn
-	serverConnection net.Conn
+	agentConnection  io.ReadWriteCloser
+	serverConnection io.ReadWriteCloser
 }
 
 func (s *AgentServerTestSuite) SetupSuite() {
@@ -34,10 +34,14 @@ func (s *AgentServerTestSuite) SetupTest() {
 	ctx, cancelFunc := testsupport.CreateTestContext(context.Background(), 10*time.Second)
 	s.ctx = ctx
 	s.cancelFunc = cancelFunc
-	serverConnection, agentConnection := net.Pipe()
-	deadline := time.Now().Add(10 * time.Second)
-	serverConnection.SetDeadline(deadline)
-	agentConnection.SetDeadline(deadline)
+
+	// Could have also used net.Pipe but net.Pipe uses synchronous communication
+	// by default and the bitpipe implementation can become asynchronous when
+	// a channels ize > 0 is passed in. Also the test utility respects the context
+	// so also deals with cancellation much better than net.Pipe.
+	bitpipe := testsupport.NewInmemoryConnection(s.ctx, "inmemory", 10)
+	agentConnection := bitpipe.Front()
+	serverConnection := bitpipe.Back()
 	s.serverConnection = serverConnection
 	s.agentConnection = agentConnection
 }

pkg/testsupport/channelreadwriter.go

@@ -49,7 +49,7 @@ func (rw *ChannelReadWriter) Read(p []byte) (n int, err error) {
 		return 0, io.ErrClosedPipe
 	case data, ok := <-rw.receiver:
 		if !ok {
-			return 0, errors.New("ladida") //io.EOF
+			return 0, io.EOF
 		}
 		nread = copy(p, data)
 		rw.readBuf = data[nread:]
@@ -57,7 +57,9 @@ func (rw *ChannelReadWriter) Read(p []byte) (n int, err error) {
 	}
 }
 
-func (rw *ChannelReadWriter) Write(p []byte) (n int, err error) {
+func (rw *ChannelReadWriter) Write(pIn []byte) (n int, err error) {
+	p := make([]byte, len(pIn), len(pIn))
+	copy(p, pIn)
 	rw.senderMutex.Lock()
 	defer rw.senderMutex.Unlock()
 

pkg/testsupport/inmemoryconnection.go

@@ -9,12 +9,12 @@ type InmemoryConnection struct {
 	addr        string
 }
 
-func NewInmemoryConnection(ctx context.Context, addr string) *InmemoryConnection {
+func NewInmemoryConnection(ctx context.Context, addr string, channelSize int) *InmemoryConnection {
 	pipe := InmemoryConnection{
 		ctx: ctx,
 		// TODO: somehow does not work with unbuffered channel and yamux
-		frontToBack: make(chan []byte, 0),
-		backToFront: make(chan []byte, 0),
+		frontToBack: make(chan []byte, channelSize),
+		backToFront: make(chan []byte, channelSize),
 		addr:        addr,
 	}
 	return &pipe

pkg/testsupport/inmemoryconnection_test.go

@@ -25,7 +25,7 @@ func (s *InMemoryTestSuite) createConnection() {
 	ctx, cancelFunc := CreateTestContext(context.Background(), 10*time.Second)
 	s.ctx = ctx
 	s.cancelFunc = cancelFunc
-	s.pipe = NewInmemoryConnection(ctx, "inmemory")
+	s.pipe = NewInmemoryConnection(ctx, "inmemory", 10)
 }
 
 func (s *InMemoryTestSuite) SetupSuite() {