Merge pull request #1439 from ipfs/proto-fix

Proto fix

p2p/net/conn/conn.go

@@ -6,7 +6,6 @@ import (
 	"net"
 	"time"
 
-	msgio "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio"
 	mpool "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio/mpool"
 	ma "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
 	manet "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
@@ -32,7 +31,6 @@ type singleConn struct {
 	local  peer.ID
 	remote peer.ID
 	maconn manet.Conn
-	msgrw  msgio.ReadWriteCloser
 	event  io.Closer
 }
 
@@ -44,7 +42,6 @@ func newSingleConn(ctx context.Context, local, remote peer.ID, maconn manet.Conn
 		local:  local,
 		remote: remote,
 		maconn: maconn,
-		msgrw:  msgio.NewReadWriter(maconn),
 		event:  log.EventBegin(ctx, "connLifetime", ml),
 	}
 
@@ -62,7 +59,7 @@ func (c *singleConn) Close() error {
 	}()
 
 	// close underlying connection
-	return c.msgrw.Close()
+	return c.maconn.Close()
 }
 
 // ID is an identifier unique to this connection.
@@ -123,31 +120,12 @@ func (c *singleConn) RemotePeer() peer.ID {
 
 // Read reads data, net.Conn style
 func (c *singleConn) Read(buf []byte) (int, error) {
-	return c.msgrw.Read(buf)
+	return c.maconn.Read(buf)
 }
 
 // Write writes data, net.Conn style
 func (c *singleConn) Write(buf []byte) (int, error) {
-	return c.msgrw.Write(buf)
-}
-
-func (c *singleConn) NextMsgLen() (int, error) {
-	return c.msgrw.NextMsgLen()
-}
-
-// ReadMsg reads data, net.Conn style
-func (c *singleConn) ReadMsg() ([]byte, error) {
-	return c.msgrw.ReadMsg()
-}
-
-// WriteMsg writes data, net.Conn style
-func (c *singleConn) WriteMsg(buf []byte) error {
-	return c.msgrw.WriteMsg(buf)
-}
-
-// ReleaseMsg releases a buffer
-func (c *singleConn) ReleaseMsg(m []byte) {
-	c.msgrw.ReleaseMsg(m)
+	return c.maconn.Write(buf)
 }
 
 // ID returns the ID of a given Conn.

p2p/net/conn/conn_test.go

@@ -8,17 +8,25 @@ import (
 	"testing"
 	"time"
 
+	msgio "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio"
 	context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
 	travis "github.com/ipfs/go-ipfs/util/testutil/ci/travis"
 )
 
+func msgioWrap(c Conn) msgio.ReadWriter {
+	return msgio.NewReadWriter(c)
+}
+
 func testOneSendRecv(t *testing.T, c1, c2 Conn) {
+	mc1 := msgioWrap(c1)
+	mc2 := msgioWrap(c2)
+
 	log.Debugf("testOneSendRecv from %s to %s", c1.LocalPeer(), c2.LocalPeer())
 	m1 := []byte("hello")
-	if err := c1.WriteMsg(m1); err != nil {
+	if err := mc1.WriteMsg(m1); err != nil {
 		t.Fatal(err)
 	}
-	m2, err := c2.ReadMsg()
+	m2, err := mc2.ReadMsg()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -28,11 +36,14 @@ func testOneSendRecv(t *testing.T, c1, c2 Conn) {
 }
 
 func testNotOneSendRecv(t *testing.T, c1, c2 Conn) {
+	mc1 := msgioWrap(c1)
+	mc2 := msgioWrap(c2)
+
 	m1 := []byte("hello")
-	if err := c1.WriteMsg(m1); err == nil {
+	if err := mc1.WriteMsg(m1); err == nil {
 		t.Fatal("write should have failed", err)
 	}
-	_, err := c2.ReadMsg()
+	_, err := mc2.ReadMsg()
 	if err == nil {
 		t.Fatal("read should have failed", err)
 	}
@@ -72,10 +83,13 @@ func TestCloseLeak(t *testing.T) {
 		ctx, cancel := context.WithCancel(context.Background())
 		c1, c2, _, _ := setupSingleConn(t, ctx)
 
+		mc1 := msgioWrap(c1)
+		mc2 := msgioWrap(c2)
+
 		for i := 0; i < num; i++ {
 			b1 := []byte(fmt.Sprintf("beep%d", i))
-			c1.WriteMsg(b1)
-			b2, err := c2.ReadMsg()
+			mc1.WriteMsg(b1)
+			b2, err := mc2.ReadMsg()
 			if err != nil {
 				panic(err)
 			}
@@ -84,8 +98,8 @@ func TestCloseLeak(t *testing.T) {
 			}
 
 			b2 = []byte(fmt.Sprintf("boop%d", i))
-			c2.WriteMsg(b2)
-			b1, err = c1.ReadMsg()
+			mc2.WriteMsg(b2)
+			b1, err = mc1.ReadMsg()
 			if err != nil {
 				panic(err)
 			}

p2p/net/conn/dial_test.go

@@ -161,10 +161,10 @@ func testDialer(t *testing.T, secure bool) {
 	}
 
 	// fmt.Println("sending")
-	c.WriteMsg([]byte("beep"))
-	c.WriteMsg([]byte("boop"))
-
-	out, err := c.ReadMsg()
+	mc := msgioWrap(c)
+	mc.WriteMsg([]byte("beep"))
+	mc.WriteMsg([]byte("boop"))
+	out, err := mc.ReadMsg()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -175,7 +175,7 @@ func testDialer(t *testing.T, secure bool) {
 		t.Error("unexpected conn output", data)
 	}
 
-	out, err = c.ReadMsg()
+	out, err = mc.ReadMsg()
 	if err != nil {
 		t.Fatal(err)
 	}

p2p/net/conn/interface.go

@@ -10,7 +10,6 @@ import (
 	filter "github.com/ipfs/go-ipfs/p2p/net/filter"
 	peer "github.com/ipfs/go-ipfs/p2p/peer"
 
-	msgio "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio"
 	ma "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
 	manet "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
 )
@@ -46,8 +45,8 @@ type Conn interface {
 	SetReadDeadline(t time.Time) error
 	SetWriteDeadline(t time.Time) error
 
-	msgio.Reader
-	msgio.Writer
+	io.Reader
+	io.Writer
 }
 
 // Dialer is an object that can open connections. We could have a "convenience"

p2p/net/conn/secure_conn.go

@@ -119,20 +119,6 @@ func (c *secureConn) Write(buf []byte) (int, error) {
 	return c.secure.ReadWriter().Write(buf)
 }
 
-func (c *secureConn) NextMsgLen() (int, error) {
-	return c.secure.ReadWriter().NextMsgLen()
-}
-
-// ReadMsg reads data, net.Conn style
-func (c *secureConn) ReadMsg() ([]byte, error) {
-	return c.secure.ReadWriter().ReadMsg()
-}
-
-// WriteMsg writes data, net.Conn style
-func (c *secureConn) WriteMsg(buf []byte) error {
-	return c.secure.ReadWriter().WriteMsg(buf)
-}
-
 // ReleaseMsg releases a buffer
 func (c *secureConn) ReleaseMsg(m []byte) {
 	c.secure.ReadWriter().ReleaseMsg(m)

p2p/net/conn/secure_conn_test.go

@@ -145,13 +145,16 @@ func TestSecureCloseLeak(t *testing.T) {
 	}
 
 	runPair := func(c1, c2 Conn, num int) {
+		mc1 := msgioWrap(c1)
+		mc2 := msgioWrap(c2)
+
 		log.Debugf("runPair %d", num)
 
 		for i := 0; i < num; i++ {
 			log.Debugf("runPair iteration %d", i)
 			b1 := []byte("beep")
-			c1.WriteMsg(b1)
-			b2, err := c2.ReadMsg()
+			mc1.WriteMsg(b1)
+			b2, err := mc2.ReadMsg()
 			if err != nil {
 				panic(err)
 			}
@@ -160,8 +163,8 @@ func TestSecureCloseLeak(t *testing.T) {
 			}
 
 			b2 = []byte("beep")
-			c2.WriteMsg(b2)
-			b1, err = c1.ReadMsg()
+			mc2.WriteMsg(b2)
+			b1, err = mc1.ReadMsg()
 			if err != nil {
 				panic(err)
 			}