swap transport over to spdystream

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

@@ -9,7 +9,6 @@ import (
  ic "github.com/ipfs/go-ipfs/p2p/crypto"
  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"
 )
@@ -45,8 +44,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"