@cryptotaxi247 / kubo / commits / a9c971fdb

update go-peerstream to newest version

License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>

Jeromy committed Jul 14, 2015 at 08:59 UTC a9c971fdbb305944756427e01d025c3c983b6103
42 files changed +1352 -181
Godeps/Godeps.json
+5 -1
@@ -180,7 +180,7 @@
180 },
181 {
182 "ImportPath": "github.com/jbenet/go-peerstream",
183 - "Rev": "62fe5ede12f9d9cd9406750160122525b3d6b694"
183 + "Rev": "214e6057e1c0742c0b5f9bb3060e88dea32a4380"
184 },
185 {
186 "ImportPath": "github.com/jbenet/go-random",
@@ -194,6 +194,10 @@
194 "ImportPath": "github.com/jbenet/go-sockaddr/net",
195 "Rev": "da304f94eea1af8ba8d1faf184623e1f9d9777dc"
196 },
197 + {
198 + "ImportPath": "github.com/jbenet/go-stream-muxer",
199 + "Rev": "4a97500beeb081571128d41d539787e137f18404"
200 + },
201 {
202 "ImportPath": "github.com/jbenet/go-temp-err-catcher",
203 "Rev": "aac704a3f4f27190b4ccc05f303a4931fd1241ff"
Godeps/_workspace/src/github.com/jbenet/go-peerstream/.travis.yml
+1 -1
@@ -7,5 +7,5 @@ go:
7 - tip
8
9 script:
10 - - go test ./...
10 + - go test -v ./...
11 # - go test -race -cpu=5 ./...
Godeps/_workspace/src/github.com/jbenet/go-peerstream/Godeps/Godeps.json
+5 -1
@@ -17,13 +17,17 @@
17 "ImportPath": "github.com/inconshreveable/muxado",
18 "Rev": "f693c7e88ba316d1a0ae3e205e22a01aa3ec2848"
19 },
20 + {
21 + "ImportPath": "github.com/jbenet/go-stream-muxer",
22 + "Rev": "e2e261765847234749629e0190fef193a4548303"
23 + },
24 {
25 "ImportPath": "github.com/jbenet/go-temp-err-catcher",
26 "Rev": "aac704a3f4f27190b4ccc05f303a4931fd1241ff"
27 },
28 {
29 "ImportPath": "github.com/whyrusleeping/go-multiplex",
26 - "Rev": "ce5baa716247510379cb7640a14da857afd3b622"
30 + "Rev": "474b9aebeb391746f304ddf7c764a5da12319857"
31 },
32 {
33 "ImportPath": "github.com/whyrusleeping/go-multistream",
Godeps/_workspace/src/github.com/jbenet/go-peerstream/conn.go
+19 -19
@@ -6,7 +6,7 @@ import (
6 "net"
7 "sync"
8
9 - pst "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
9 + smux "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer"
10 )
11
12 // ConnHandler is a function which receives a Conn. It allows
@@ -36,8 +36,8 @@ var ErrNoConnections = errors.New("no connections")
36
37 // Conn is a Swarm-associated connection.
38 type Conn struct {
39 - pstConn pst.Conn
40 - netConn net.Conn // underlying connection
39 + smuxConn smux.Conn
40 + netConn net.Conn // underlying connection
41
42 swarm *Swarm
43 groups groupSet
@@ -46,13 +46,13 @@ type Conn struct {
46 streamLock sync.RWMutex
47 }
48
49 -func newConn(nconn net.Conn, tconn pst.Conn, s *Swarm) *Conn {
49 +func newConn(nconn net.Conn, tconn smux.Conn, s *Swarm) *Conn {
50 return &Conn{
51 - netConn: nconn,
52 - pstConn: tconn,
53 - swarm: s,
54 - groups: groupSet{m: make(map[Group]struct{})},
55 - streams: make(map[*Stream]struct{}),
51 + netConn: nconn,
52 + smuxConn: tconn,
53 + swarm: s,
54 + groups: groupSet{m: make(map[Group]struct{})},
55 + streams: make(map[*Stream]struct{}),
56 }
57 }
58
@@ -77,8 +77,8 @@ func (c *Conn) NetConn() net.Conn {
77
78 // Conn returns the underlying transport Connection we use
79 // Warning: modifying this object is undefined.
80 -func (c *Conn) Conn() pst.Conn {
81 - return c.pstConn
80 +func (c *Conn) Conn() smux.Conn {
81 + return c.smuxConn
82 }
83
84 // Groups returns the Groups this Conn belongs to
@@ -122,7 +122,7 @@ func (c *Conn) Close() error {
122
123 // close underlying connection
124 c.swarm.removeConn(c)
125 - err := c.pstConn.Close()
125 + err := c.smuxConn.Close()
126 c.swarm.notifyAll(func(n Notifiee) {
127 n.Disconnected(c)
128 })
@@ -166,7 +166,7 @@ func (s *Swarm) addConn(netConn net.Conn, isServer bool) (*Conn, error) {
166 s.ConnHandler()(c)
167
168 // go listen for incoming streams on this connection
169 - go c.pstConn.Serve(func(ss pst.Stream) {
169 + go c.smuxConn.Serve(func(ss smux.Stream) {
170 // log.Printf("accepted stream %d from %s\n", ssS.Identifier(), netConn.RemoteAddr())
171 stream := s.setupStream(ss, c)
172 s.StreamHandler()(stream) // call our handler
@@ -225,21 +225,21 @@ func (s *Swarm) setupConn(netConn net.Conn, isServer bool) (*Conn, error) {
225 // all validation has happened.
226 func (s *Swarm) createStream(c *Conn) (*Stream, error) {
227
228 - // Create a new pst.Stream
229 - pstStream, err := c.pstConn.OpenStream()
228 + // Create a new smux.Stream
229 + smuxStream, err := c.smuxConn.OpenStream()
230 if err != nil {
231 return nil, err
232 }
233
234 - return s.setupStream(pstStream, c), nil
234 + return s.setupStream(smuxStream, c), nil
235 }
236
237 // newStream is the internal function that creates a new stream. assumes
238 // all validation has happened.
239 -func (s *Swarm) setupStream(pstStream pst.Stream, c *Conn) *Stream {
239 +func (s *Swarm) setupStream(smuxStream smux.Stream, c *Conn) *Stream {
240
241 // create a new stream
242 - stream := newStream(pstStream, c)
242 + stream := newStream(smuxStream, c)
243
244 // add it to our streams maps
245 s.streamLock.Lock()
@@ -265,7 +265,7 @@ func (s *Swarm) removeStream(stream *Stream) error {
265 s.streamLock.Unlock()
266 stream.conn.streamLock.Unlock()
267
268 - err := stream.pstStream.Close()
268 + err := stream.smuxStream.Close()
269 s.notifyAll(func(n Notifiee) {
270 n.ClosedStream(stream)
271 })
Godeps/_workspace/src/github.com/jbenet/go-peerstream/example/blockhandler/blockhandler.go
+2 -2
@@ -8,7 +8,7 @@ import (
8 "time"
9
10 ps "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream"
11 - pstss "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/spdystream"
11 + spdy "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/spdystream"
12 )
13
14 func die(err error) {
@@ -18,7 +18,7 @@ func die(err error) {
18
19 func main() {
20 // create a new Swarm
21 - swarm := ps.NewSwarm(pstss.Transport)
21 + swarm := ps.NewSwarm(spdy.Transport)
22 defer swarm.Close()
23
24 // tell swarm what to do with a new incoming streams.
Godeps/_workspace/src/github.com/jbenet/go-peerstream/example/example.go
+2 -2
@@ -7,13 +7,13 @@ import (
7 "os"
8
9 ps "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream"
10 - pstss "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/spdystream"
10 + spdy "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/spdystream"
11 )
12
13 func main() {
14
15 log("creating a new swarm with spdystream transport") // create a new Swarm
16 - swarm := ps.NewSwarm(pstss.Transport)
16 + swarm := ps.NewSwarm(spdy.Transport)
17 defer swarm.Close()
18
19 // tell swarm what to do with a new incoming streams.
Godeps/_workspace/src/github.com/jbenet/go-peerstream/muxtest/hack.go new
+15
@@ -0,0 +1,15 @@
1 +package muxtest
2 +
3 +import (
4 + multiplex "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/multiplex"
5 + multistream "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/multistream"
6 + muxado "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/muxado"
7 + spdy "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/spdystream"
8 + yamux "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/yamux"
9 +)
10 +
11 +var _ = multiplex.DefaultTransport
12 +var _ = multistream.NewTransport
13 +var _ = muxado.Transport
14 +var _ = spdy.Transport
15 +var _ = yamux.DefaultTransport
Godeps/_workspace/src/github.com/jbenet/go-peerstream/muxtest/mux_test.go new
+31
@@ -0,0 +1,31 @@
1 +package muxtest
2 +
3 +import (
4 + "testing"
5 +
6 + multiplex "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/multiplex"
7 + multistream "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/multistream"
8 + muxado "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/muxado"
9 + spdy "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/spdystream"
10 + yamux "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/yamux"
11 +)
12 +
13 +func TestYamuxTransport(t *testing.T) {
14 + SubtestAll(t, yamux.DefaultTransport)
15 +}
16 +
17 +func TestSpdyStreamTransport(t *testing.T) {
18 + SubtestAll(t, spdy.Transport)
19 +}
20 +
21 +func TestMultiplexTransport(t *testing.T) {
22 + SubtestAll(t, multiplex.DefaultTransport)
23 +}
24 +
25 +func TestMuxadoTransport(t *testing.T) {
26 + SubtestAll(t, muxado.Transport)
27 +}
28 +
29 +func TestMultistreamTransport(t *testing.T) {
30 + SubtestAll(t, multistream.NewTransport())
31 +}
Godeps/_workspace/src/github.com/jbenet/go-peerstream/muxtest/muxt.go renamed
+30 -43
@@ -1,4 +1,4 @@
1 -package peerstream_transport_test
1 +package muxtest
2
3 import (
4 "bytes"
@@ -14,11 +14,12 @@ import (
14 "testing"
15
16 ps "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream"
17 - pst "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
17 + smux "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer"
18 )
19
20 var randomness []byte
21 var nextPort = 20000
22 +var verbose = false
23
24 func init() {
25 // read 1MB of randomness
@@ -45,7 +46,7 @@ func checkErr(t *testing.T, err error) {
46 }
47
48 func log(s string, v ...interface{}) {
48 - if testing.Verbose() {
49 + if verbose {
50 fmt.Fprintf(os.Stderr, "> "+s+"\n", v...)
51 }
52 }
@@ -55,7 +56,7 @@ type echoSetup struct {
56 conns []*ps.Conn
57 }
58
58 -func singleConn(t *testing.T, tr pst.Transport) echoSetup {
59 +func singleConn(t *testing.T, tr smux.Transport) echoSetup {
60 swarm := ps.NewSwarm(tr)
61 swarm.SetStreamHandler(func(s *ps.Stream) {
62 defer s.Close()
@@ -84,7 +85,7 @@ func singleConn(t *testing.T, tr pst.Transport) echoSetup {
85 }
86 }
87
87 -func makeSwarm(t *testing.T, tr pst.Transport, nListeners int) *ps.Swarm {
88 +func makeSwarm(t *testing.T, tr smux.Transport, nListeners int) *ps.Swarm {
89 swarm := ps.NewSwarm(tr)
90 swarm.SetStreamHandler(func(s *ps.Stream) {
91 defer s.Close()
@@ -104,7 +105,7 @@ func makeSwarm(t *testing.T, tr pst.Transport, nListeners int) *ps.Swarm {
105 return swarm
106 }
107
107 -func makeSwarms(t *testing.T, tr pst.Transport, nSwarms, nListeners int) []*ps.Swarm {
108 +func makeSwarms(t *testing.T, tr smux.Transport, nSwarms, nListeners int) []*ps.Swarm {
109 swarms := make([]*ps.Swarm, nSwarms)
110 for i := 0; i < nSwarms; i++ {
111 swarms[i] = makeSwarm(t, tr, nListeners)
@@ -112,11 +113,11 @@ func makeSwarms(t *testing.T, tr pst.Transport, nSwarms, nListeners int) []*ps.S
113 return swarms
114 }
115
115 -func SubtestConstructSwarm(t *testing.T, tr pst.Transport) {
116 +func SubtestConstructSwarm(t *testing.T, tr smux.Transport) {
117 ps.NewSwarm(tr)
118 }
119
119 -func SubtestSimpleWrite(t *testing.T, tr pst.Transport) {
120 +func SubtestSimpleWrite(t *testing.T, tr smux.Transport) {
121 swarm := ps.NewSwarm(tr)
122 defer swarm.Close()
123
@@ -171,18 +172,18 @@ func SubtestSimpleWrite(t *testing.T, tr pst.Transport) {
172 }
173 }
174
174 -func SubtestSimpleWrite100msgs(t *testing.T, tr pst.Transport) {
175 +func SubtestSimpleWrite100msgs(t *testing.T, tr smux.Transport) {
176
177 msgs := 100
178 msgsize := 1 << 19
179 es := singleConn(t, tr)
180 + defer es.swarm.Close()
181
182 log("creating stream")
183 stream, err := es.conns[0].NewStream()
184 checkErr(t, err)
185
186 bufs := make(chan []byte, msgs)
185 - errs := make(chan error, msgs*100)
187 var wg sync.WaitGroup
188
189 wg.Add(1)
@@ -194,7 +195,7 @@ func SubtestSimpleWrite100msgs(t *testing.T, tr pst.Transport) {
195 bufs <- buf
196 log("writing %d bytes (message %d/%d #%x)", len(buf), i, msgs, buf[:3])
197 if _, err := stream.Write(buf); err != nil {
197 - errs <- fmt.Errorf("stream.Write(buf): %s", err)
198 + t.Error(fmt.Errorf("stream.Write(buf): %s", err))
199 continue
200 }
201 }
@@ -212,26 +213,21 @@ func SubtestSimpleWrite100msgs(t *testing.T, tr pst.Transport) {
213 i++
214
215 if _, err := io.ReadFull(stream, buf2); err != nil {
215 - errs <- fmt.Errorf("readFull(stream, buf2): %s", err)
216 + t.Error(fmt.Errorf("readFull(stream, buf2): %s", err))
217 continue
218 }
219 if !bytes.Equal(buf1, buf2) {
219 - errs <- fmt.Errorf("buffers not equal (%x != %x)", buf1[:3], buf2[:3])
220 + t.Error(fmt.Errorf("buffers not equal (%x != %x)", buf1[:3], buf2[:3]))
221 }
222 }
223 }()
224
225 wg.Wait()
225 - close(errs)
226 - for err := range errs {
227 - t.Error(err)
228 - }
226 }
227
231 -func SubtestStressNSwarmNConnNStreamNMsg(t *testing.T, tr pst.Transport, nSwarm, nConn, nStream, nMsg int) {
228 +func SubtestStressNSwarmNConnNStreamNMsg(t *testing.T, tr smux.Transport, nSwarm, nConn, nStream, nMsg int) {
229
230 msgsize := 1 << 11
234 - errs := make(chan error, nSwarm*nConn*nStream*nMsg*100) // dont block anything.
231
232 rateLimitN := 5000
233 rateLimitChan := make(chan struct{}, rateLimitN) // max of 5k funcs.
@@ -253,7 +249,7 @@ func SubtestStressNSwarmNConnNStreamNMsg(t *testing.T, tr pst.Transport, nSwarm,
249 bufs <- buf
250 log("%p writing %d bytes (message %d/%d #%x)", s, len(buf), i, nMsg, buf[:3])
251 if _, err := s.Write(buf); err != nil {
256 - errs <- fmt.Errorf("s.Write(buf): %s", err)
252 + t.Error(fmt.Errorf("s.Write(buf): %s", err))
253 continue
254 }
255 }
@@ -269,12 +265,12 @@ func SubtestStressNSwarmNConnNStreamNMsg(t *testing.T, tr pst.Transport, nSwarm,
265 log("%p reading %d bytes (message %d/%d #%x)", s, len(buf1), i-1, nMsg, buf1[:3])
266
267 if _, err := io.ReadFull(s, buf2); err != nil {
272 - errs <- fmt.Errorf("io.ReadFull(s, buf2): %s", err)
268 log("%p failed to read %d bytes (message %d/%d #%x)", s, len(buf1), i-1, nMsg, buf1[:3])
269 + t.Error(fmt.Errorf("io.ReadFull(s, buf2): %s", err))
270 continue
271 }
272 if !bytes.Equal(buf1, buf2) {
277 - errs <- fmt.Errorf("buffers not equal (%x != %x)", buf1[:3], buf2[:3])
273 + t.Error(fmt.Errorf("buffers not equal (%x != %x)", buf1[:3], buf2[:3]))
274 }
275 }
276 }
@@ -284,7 +280,7 @@ func SubtestStressNSwarmNConnNStreamNMsg(t *testing.T, tr pst.Transport, nSwarm,
280
281 s, err := c.NewStream()
282 if err != nil {
287 - errs <- fmt.Errorf("Failed to create NewStream: %s", err)
283 + t.Error(fmt.Errorf("Failed to create NewStream: %s", err))
284 return
285 }
286
@@ -308,13 +304,13 @@ func SubtestStressNSwarmNConnNStreamNMsg(t *testing.T, tr pst.Transport, nSwarm,
304
305 nc, err := net.Dial(nla.Network(), nla.String())
306 if err != nil {
311 - errs <- fmt.Errorf("net.Dial(%s, %s): %s", nla.Network(), nla.String(), err)
307 + t.Fatal(fmt.Errorf("net.Dial(%s, %s): %s", nla.Network(), nla.String(), err))
308 return
309 }
310
311 c, err := a.AddConn(nc)
312 if err != nil {
317 - errs <- fmt.Errorf("a.AddConn(%s <--> %s): %s", nc.LocalAddr(), nc.RemoteAddr(), err)
313 + t.Fatal(fmt.Errorf("a.AddConn(%s <--> %s): %s", nc.LocalAddr(), nc.RemoteAddr(), err))
314 return
315 }
316
@@ -363,47 +359,38 @@ func SubtestStressNSwarmNConnNStreamNMsg(t *testing.T, tr pst.Transport, nSwarm,
359 }
360
361 swarms := makeSwarms(t, tr, nSwarm, 3) // 3 listeners per swarm.
366 -
367 - go func() {
368 - connectSwarmsAndRW(swarms)
369 - close(errs) // done
370 - }()
371 -
372 - for err := range errs {
373 - t.Error(err)
374 - }
375 -
362 + connectSwarmsAndRW(swarms)
363 for _, s := range swarms {
364 s.Close()
365 }
366
367 }
368
382 -func SubtestStress1Swarm1Conn1Stream1Msg(t *testing.T, tr pst.Transport) {
369 +func SubtestStress1Swarm1Conn1Stream1Msg(t *testing.T, tr smux.Transport) {
370 SubtestStressNSwarmNConnNStreamNMsg(t, tr, 1, 1, 1, 1)
371 }
372
386 -func SubtestStress1Swarm1Conn1Stream100Msg(t *testing.T, tr pst.Transport) {
373 +func SubtestStress1Swarm1Conn1Stream100Msg(t *testing.T, tr smux.Transport) {
374 SubtestStressNSwarmNConnNStreamNMsg(t, tr, 1, 1, 1, 100)
375 }
376
390 -func SubtestStress1Swarm1Conn100Stream100Msg(t *testing.T, tr pst.Transport) {
377 +func SubtestStress1Swarm1Conn100Stream100Msg(t *testing.T, tr smux.Transport) {
378 SubtestStressNSwarmNConnNStreamNMsg(t, tr, 1, 1, 100, 100)
379 }
380
394 -func SubtestStress1Swarm10Conn50Stream50Msg(t *testing.T, tr pst.Transport) {
381 +func SubtestStress1Swarm10Conn50Stream50Msg(t *testing.T, tr smux.Transport) {
382 SubtestStressNSwarmNConnNStreamNMsg(t, tr, 1, 10, 50, 50)
383 }
384
398 -func SubtestStress5Swarm2Conn20Stream20Msg(t *testing.T, tr pst.Transport) {
385 +func SubtestStress5Swarm2Conn20Stream20Msg(t *testing.T, tr smux.Transport) {
386 SubtestStressNSwarmNConnNStreamNMsg(t, tr, 5, 2, 20, 20)
387 }
388
402 -func SubtestStress10Swarm2Conn100Stream100Msg(t *testing.T, tr pst.Transport) {
389 +func SubtestStress10Swarm2Conn100Stream100Msg(t *testing.T, tr smux.Transport) {
390 SubtestStressNSwarmNConnNStreamNMsg(t, tr, 10, 2, 100, 100)
391 }
392
406 -func SubtestAll(t *testing.T, tr pst.Transport) {
393 +func SubtestAll(t *testing.T, tr smux.Transport) {
394
395 tests := []TransportTest{
396 SubtestConstructSwarm,
@@ -425,7 +412,7 @@ func SubtestAll(t *testing.T, tr pst.Transport) {
412 }
413 }
414
428 -type TransportTest func(t *testing.T, tr pst.Transport)
415 +type TransportTest func(t *testing.T, tr smux.Transport)
416
417 func TestNoOp(t *testing.T) {}
418
Godeps/_workspace/src/github.com/jbenet/go-peerstream/stream.go
+10 -10
@@ -3,7 +3,7 @@ package peerstream
3 import (
4 "fmt"
5
6 - pst "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
6 + smux "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer"
7 )
8
9 // StreamHandler is a function which receives a Stream. It
@@ -17,17 +17,17 @@ type StreamHandler func(s *Stream)
17 // Stream is an io.{Read,Write,Close}r to a remote counterpart.
18 // It wraps a spdystream.Stream, and links it to a Conn and groups
19 type Stream struct {
20 - pstStream pst.Stream
20 + smuxStream smux.Stream
21
22 conn *Conn
23 groups groupSet
24 }
25
26 -func newStream(ss pst.Stream, c *Conn) *Stream {
26 +func newStream(ss smux.Stream, c *Conn) *Stream {
27 s := &Stream{
28 - conn: c,
29 - pstStream: ss,
30 - groups: groupSet{m: make(map[Group]struct{})},
28 + conn: c,
29 + smuxStream: ss,
30 + groups: groupSet{m: make(map[Group]struct{})},
31 }
32 s.groups.AddSet(&c.groups) // inherit groups
33 return s
@@ -40,8 +40,8 @@ func (s *Stream) String() string {
40 }
41
42 // SPDYStream returns the underlying *spdystream.Stream
43 -func (s *Stream) Stream() pst.Stream {
44 - return s.pstStream
43 +func (s *Stream) Stream() smux.Stream {
44 + return s.smuxStream
45 }
46
47 // Conn returns the Conn associated with this Stream
@@ -70,11 +70,11 @@ func (s *Stream) AddGroup(g Group) {
70 }
71
72 func (s *Stream) Read(p []byte) (n int, err error) {
73 - return s.pstStream.Read(p)
73 + return s.smuxStream.Read(p)
74 }
75
76 func (s *Stream) Write(p []byte) (n int, err error) {
77 - return s.pstStream.Write(p)
77 + return s.smuxStream.Write(p)
78 }
79
80 func (s *Stream) Close() error {
Godeps/_workspace/src/github.com/jbenet/go-peerstream/swarm.go
+6 -6
@@ -7,7 +7,7 @@ import (
7 "sync"
8 "time"
9
10 - pst "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
10 + smux "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer"
11 )
12
13 // fd is a (file) descriptor, unix style
@@ -18,7 +18,7 @@ var GarbageCollectTimeout = 5 * time.Second
18
19 type Swarm struct {
20 // the transport we'll use.
21 - transport pst.Transport
21 + transport smux.Transport
22
23 // active streams.
24 streams map[*Stream]struct{}
@@ -46,7 +46,7 @@ type Swarm struct {
46 closed chan struct{}
47 }
48
49 -func NewSwarm(t pst.Transport) *Swarm {
49 +func NewSwarm(t smux.Transport) *Swarm {
50 s := &Swarm{
51 transport: t,
52 streams: make(map[*Stream]struct{}),
@@ -183,7 +183,7 @@ func (s *Swarm) Conns() []*Conn {
183
184 open := make([]*Conn, 0, len(conns))
185 for _, c := range conns {
186 - if c.pstConn.IsClosed() {
186 + if c.smuxConn.IsClosed() {
187 c.Close()
188 } else {
189 open = append(open, c)
@@ -292,7 +292,7 @@ func (s *Swarm) NewStreamWithConn(conn *Conn) (*Stream, error) {
292 return nil, errors.New("connection not associated with swarm")
293 }
294
295 - if conn.pstConn.IsClosed() {
295 + if conn.smuxConn.IsClosed() {
296 go conn.Close()
297 return nil, errors.New("conn is closed")
298 }
@@ -360,7 +360,7 @@ func (s *Swarm) connGarbageCollect() {
360 }
361
362 for _, c := range s.Conns() {
363 - if c.pstConn.IsClosed() {
363 + if c.smuxConn.IsClosed() {
364 go c.Close()
365 }
366 }
Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/multiplex/multiplex_test.go deleted
-11
@@ -1,11 +0,0 @@
1 -package peerstream_multiplex
2 -
3 -import (
4 - "testing"
5 -
6 - psttest "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/test"
7 -)
8 -
9 -func TestMultiplexTransport(t *testing.T) {
10 - psttest.SubtestAll(t, DefaultTransport)
11 -}
Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/multistream/multistream_test.go deleted
-11
@@ -1,11 +0,0 @@
1 -package multistream
2 -
3 -import (
4 - "testing"
5 -
6 - psttest "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/test"
7 -)
8 -
9 -func TestMultiStreamTransport(t *testing.T) {
10 - psttest.SubtestAll(t, NewTransport())
11 -}
Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/muxado/muxado_test.go deleted
-11
@@ -1,11 +0,0 @@
1 -package peerstream_muxado
2 -
3 -import (
4 - "testing"
5 -
6 - psttest "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/test"
7 -)
8 -
9 -func TestMuxadoTransport(t *testing.T) {
10 - psttest.SubtestAll(t, Transport)
11 -}
Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/spdystream/spdystream_test.go deleted
-11
@@ -1,11 +0,0 @@
1 -package peerstream_spdystream
2 -
3 -import (
4 - "testing"
5 -
6 - psttest "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/test"
7 -)
8 -
9 -func TestSpdyStreamTransport(t *testing.T) {
10 - psttest.SubtestAll(t, Transport)
11 -}
Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/yamux/yamux_test.go deleted
-11
@@ -1,11 +0,0 @@
1 -package peerstream_yamux
2 -
3 -import (
4 - "testing"
5 -
6 - psttest "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/test"
7 -)
8 -
9 -func TestYamuxTransport(t *testing.T) {
10 - psttest.SubtestAll(t, DefaultTransport)
11 -}
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/.travis.yml new
+11
@@ -0,0 +1,11 @@
1 +language: go
2 +
3 +go:
4 + - 1.3
5 + - 1.4
6 + - release
7 + - tip
8 +
9 +script:
10 + - go test ./...
11 + # - go test -race -cpu=5 ./...
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/Godeps/Godeps.json new
+29
@@ -0,0 +1,29 @@
1 +{
2 + "ImportPath": "github.com/jbenet/go-stream-muxer",
3 + "GoVersion": "go1.4.2",
4 + "Packages": [
5 + "./..."
6 + ],
7 + "Deps": [
8 + {
9 + "ImportPath": "github.com/docker/spdystream",
10 + "Rev": "b2c3287865f3ad6aa22821ddb7b4692b896ac207"
11 + },
12 + {
13 + "ImportPath": "github.com/hashicorp/yamux",
14 + "Rev": "b2e55852ddaf823a85c67f798080eb7d08acd71d"
15 + },
16 + {
17 + "ImportPath": "github.com/inconshreveable/muxado",
18 + "Rev": "f693c7e88ba316d1a0ae3e205e22a01aa3ec2848"
19 + },
20 + {
21 + "ImportPath": "github.com/whyrusleeping/go-multiplex",
22 + "Rev": "ce5baa716247510379cb7640a14da857afd3b622"
23 + },
24 + {
25 + "ImportPath": "github.com/whyrusleeping/go-multistream",
26 + "Rev": "08e8f9c9f5665ed0c63ffde4fa5ef1d5fb3d516d"
27 + }
28 + ]
29 +}
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/Godeps/Readme new
+5
@@ -0,0 +1,5 @@
1 +This directory tree is generated automatically by godep.
2 +
3 +Please do not edit.
4 +
5 +See https://github.com/tools/godep for more information.
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/LICENSE new
+21
@@ -0,0 +1,21 @@
1 +The MIT License (MIT)
2 +
3 +Copyright (c) 2014 Juan Batiz-Benet
4 +
5 +Permission is hereby granted, free of charge, to any person obtaining a copy
6 +of this software and associated documentation files (the "Software"), to deal
7 +in the Software without restriction, including without limitation the rights
8 +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9 +copies of the Software, and to permit persons to whom the Software is
10 +furnished to do so, subject to the following conditions:
11 +
12 +The above copyright notice and this permission notice shall be included in
13 +all copies or substantial portions of the Software.
14 +
15 +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18 +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20 +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
21 +THE SOFTWARE.
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/Makefile new
+15
@@ -0,0 +1,15 @@
1 +
2 +godep:
3 + go get github.com/tools/godep
4 +
5 +vendor: godep
6 + godep save -r ./...
7 +
8 +build:
9 + go build ./...
10 +
11 +test:
12 + go test ./...
13 +
14 +test_race:
15 + go test -race -cpu 5 ./...
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/README.md new
+98
@@ -0,0 +1,98 @@
1 +# go-stream-muxer - generalized stream multiplexing
2 +
3 +
4 +go-stream-muxer is a common interface for stream muxers, with common tests. It wraps other stream muxers (like [muxado](https://github.com/inconshreveable/muxado), [spdystream](https://github.com/docker/spdystream) and [yamux](https://github.com/hashicorp/yamux)).
5 +
6 +[![](https://img.shields.io/badge/made%20by-Protocol%20Labs-blue.svg?style=flat-square)](http://ipn.io) [![](https://img.shields.io/badge/freenode-%23ipfs-blue.svg?style=flat-square)](http://webchat.freenode.net/?channels=%23ipfs)
7 +
8 +> A test suite and interface you can use to implement a stream muxer.
9 +
10 +### Godoc: https://godoc.org/github.com/jbenet/go-stream-muxer
11 +
12 +## Implementations
13 +
14 +* [yamux](yamux)
15 +* [muxado](muxado)
16 +* [multiplex](multiplex)
17 +* [spdystream](spdystream)
18 +
19 +## Badge
20 +
21 +Include this badge in your readme if you make a new module that uses abstract-stream-muxer API.
22 +
23 +![](img/badge.png)
24 +
25 +## Client example
26 +
27 +```go
28 +import (
29 + "net"
30 + "fmt"
31 + "io"
32 + ymux "github.com/jbenet/go-stream-muxer/yamux"
33 + smux "github.com/jbenet/go-stream-muxer"
34 +)
35 +
36 +func dial() {
37 + nconn, _ := net.Dial("tcp", "localhost:1234")
38 + sconn, _ := ymux.DefaultTransport.NewConn(nconn, false) // false == client
39 +
40 + go sconn.Serve(func(smux.Stream) {}) // no-op
41 +
42 + s1, _ := sconn.OpenStream()
43 + s1.Write([]byte("hello"))
44 +
45 + s2, _ := sconn.OpenStream()
46 + s2.Write([]byte("world"))
47 +
48 + length := 20
49 + buf2 := make([]byte, length)
50 + fmt.Printf("reading %d bytes from stream (echoed)\n", length)
51 +
52 + s1.Read(buf2)
53 +
54 + fmt.Printf("received %s as a response\n", string(buf2))
55 +
56 + s3, _ := sconn.OpenStream()
57 + io.Copy(s3, os.Stdin)
58 +}
59 +```
60 +
61 +## Server example
62 +
63 +```go
64 +import (
65 + "net"
66 + "fmt"
67 + "io"
68 + ymux "github.com/jbenet/go-stream-muxer/yamux"
69 + smux "github.com/jbenet/go-stream-muxer"
70 +)
71 +
72 +func listen() {
73 + tr := ymux.DefaultTransport
74 + l, _ := net.Listen("tcp", "localhost:1234")
75 +
76 + go func() {
77 + for {
78 + c, _ := l.Accept()
79 +
80 + fmt.Println("accepted connection")
81 + sc, _ := tr.NewConn(c, true)
82 +
83 + go sc.Serve(func(s smux.Stream) {
84 + fmt.Println("serving connection")
85 + echoStream(s)
86 + })
87 + }
88 + }()
89 +}
90 +
91 +func echoStream(s smux.Stream) {
92 + defer s.Close()
93 +
94 + fmt.Println("accepted stream")
95 + io.Copy(s, s) // echo everything
96 + fmt.Println("closing stream")
97 +}
98 +```
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/img/badge.png
Binary files /dev/null and b/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/img/badge.png differ
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/multiplex/multiplex.go renamed
+14 -7
@@ -1,18 +1,20 @@
1 package peerstream_multiplex
2
3 import (
4 + "errors"
5 "net"
6
6 - pst "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
7 - mp "github.com/whyrusleeping/go-multiplex"
7 + smux "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer"
8 + mp "github.com/whyrusleeping/go-multiplex" // Conn is a connection to a remote peer.
9 )
10
11 +var ErrUseServe = errors.New("not implemented, use Serve")
12 +
13 type conn struct {
14 *mp.Multiplex
15 }
16
14 -func ( // Conn is a connection to a remote peer.
15 -c *conn) Close() error {
17 +func (c *conn) Close() error {
18 return c.Multiplex.Close()
19 }
20
@@ -21,13 +23,18 @@ func (c *conn) IsClosed() bool {
23 }
24
25 // OpenStream creates a new stream.
24 -func (c *conn) OpenStream() (pst.Stream, error) {
26 +func (c *conn) OpenStream() (smux.Stream, error) {
27 return c.Multiplex.NewStream(), nil
28 }
29
30 +// AcceptStream accepts a stream opened by the other side.
31 +func (c *conn) AcceptStream() (smux.Stream, error) {
32 + return nil, ErrUseServe
33 +}
34 +
35 // Serve starts listening for incoming requests and handles them
36 // using given StreamHandler
30 -func (c *conn) Serve(handler pst.StreamHandler) {
37 +func (c *conn) Serve(handler smux.StreamHandler) {
38 c.Multiplex.Serve(func(s *mp.Stream) {
39 handler(s)
40 })
@@ -40,6 +47,6 @@ type Transport struct{}
47 // DefaultTransport has default settings for multiplex
48 var DefaultTransport = &Transport{}
49
43 -func (t *Transport) NewConn(nc net.Conn, isServer bool) (pst.Conn, error) {
50 +func (t *Transport) NewConn(nc net.Conn, isServer bool) (smux.Conn, error) {
51 return &conn{mp.NewMultiplex(nc, isServer)}, nil
52 }
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/multiplex/multiplex_test.go new
+11
@@ -0,0 +1,11 @@
1 +package peerstream_multiplex
2 +
3 +import (
4 + "testing"
5 +
6 + test "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/test"
7 +)
8 +
9 +func TestMultiplexTransport(t *testing.T) {
10 + test.SubtestAll(t, DefaultTransport)
11 +}
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/multistream/multistream.go renamed
+9 -8
@@ -5,26 +5,27 @@ package multistream
5 import (
6 "net"
7
8 - pst "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
9 - multiplex "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/multiplex"
10 - spdy "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/spdystream"
11 - yamux "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/yamux"
8 mss "github.com/whyrusleeping/go-multistream"
9 +
10 + smux "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer"
11 + multiplex "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/multiplex"
12 + spdy "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/spdystream"
13 + yamux "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/yamux"
14 )
15
16 type transport struct {
17 mux *mss.MultistreamMuxer
18
18 - tpts map[string]pst.Transport
19 + tpts map[string]smux.Transport
20 }
21
21 -func NewTransport() pst.Transport {
22 +func NewTransport() smux.Transport {
23 mux := mss.NewMultistreamMuxer()
24 mux.AddHandler("/multiplex", nil)
25 mux.AddHandler("/spdystream", nil)
26 mux.AddHandler("/yamux", nil)
27
27 - tpts := map[string]pst.Transport{
28 + tpts := map[string]smux.Transport{
29 "/multiplex": multiplex.DefaultTransport,
30 "/spdystream": spdy.Transport,
31 "/yamux": yamux.DefaultTransport,
@@ -36,7 +37,7 @@ func NewTransport() pst.Transport {
37 }
38 }
39
39 -func (t *transport) NewConn(nc net.Conn, isServer bool) (pst.Conn, error) {
40 +func (t *transport) NewConn(nc net.Conn, isServer bool) (smux.Conn, error) {
41 var proto string
42 if isServer {
43 selected, _, err := t.mux.Negotiate(nc)
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/multistream/multistream_test.go new
+11
@@ -0,0 +1,11 @@
1 +package multistream
2 +
3 +import (
4 + "testing"
5 +
6 + test "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/test"
7 +)
8 +
9 +func TestMultiStreamTransport(t *testing.T) {
10 + test.SubtestAll(t, NewTransport())
11 +}
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/muxado/muxado.go renamed
+16 -7
@@ -4,10 +4,10 @@ import (
4 "net"
5
6 muxado "github.com/inconshreveable/muxado"
7 - pst "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
7 + smux "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer"
8 )
9
10 -// stream implements pst.Stream using a ss.Stream
10 +// stream implements smux.Stream using a ss.Stream
11 type stream struct {
12 ms muxado.Stream
13 }
@@ -53,7 +53,7 @@ func (c *conn) IsClosed() bool {
53 }
54
55 // OpenStream creates a new stream.
56 -func (c *conn) OpenStream() (pst.Stream, error) {
56 +func (c *conn) OpenStream() (smux.Stream, error) {
57 s, err := c.ms.Open()
58 if err != nil {
59 return nil, err
@@ -62,15 +62,24 @@ func (c *conn) OpenStream() (pst.Stream, error) {
62 return &stream{ms: s}, nil
63 }
64
65 +// AcceptStream accepts a stream opened by the other side.
66 +func (c *conn) AcceptStream() (smux.Stream, error) {
67 + s, err := c.ms.Accept()
68 + if err != nil {
69 + return nil, err
70 + }
71 + return &stream{ms: s}, nil
72 +}
73 +
74 // Serve starts listening for incoming requests and handles them
75 // using given StreamHandler
67 -func (c *conn) Serve(handler pst.StreamHandler) {
76 +func (c *conn) Serve(handler smux.StreamHandler) {
77 for { // accept loop
69 - s, err := c.ms.Accept()
78 + s, err := c.AcceptStream()
79 if err != nil {
80 return // err always means closed.
81 }
73 - go handler(&stream{ms: s})
82 + go handler(s)
83 }
84 }
85
@@ -80,7 +89,7 @@ type transport struct{}
89 // spdystream-backed connections.
90 var Transport = transport{}
91
83 -func (t transport) NewConn(nc net.Conn, isServer bool) (pst.Conn, error) {
92 +func (t transport) NewConn(nc net.Conn, isServer bool) (smux.Conn, error) {
93 var s muxado.Session
94 if isServer {
95 s = muxado.Server(nc)
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/muxado/muxado_test.go new
+11
@@ -0,0 +1,11 @@
1 +package peerstream_muxado
2 +
3 +import (
4 + "testing"
5 +
6 + test "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/test"
7 +)
8 +
9 +func TestMuxadoTransport(t *testing.T) {
10 + test.SubtestAll(t, Transport)
11 +}
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/muxer.go renamed
+10 -4
@@ -1,4 +1,4 @@
1 -package peerstream_transport
1 +package streammux
2
3 import (
4 "io"
@@ -16,6 +16,9 @@ type Stream interface {
16 // (usually those opened by the remote side)
17 type StreamHandler func(Stream)
18
19 +// NoOpHandler do nothing. close streams as soon as they are opened.
20 +var NoOpHandler = func(s Stream) { s.Close() }
21 +
22 // Conn is a stream-multiplexing connection to a remote peer.
23 type Conn interface {
24 io.Closer
@@ -27,12 +30,15 @@ type Conn interface {
30 // OpenStream creates a new stream.
31 OpenStream() (Stream, error)
32
30 - // Serve starts listening for incoming requests and handles them
31 - // using given StreamHandler
33 + // AcceptStream accepts a stream opened by the other side.
34 + AcceptStream() (Stream, error)
35 +
36 + // Serve starts a loop, accepting incoming requests and calling
37 + // `StreamHandler with them. (Use _instead of_ accept. not both.)
38 Serve(StreamHandler)
39 }
40
35 -// Transport constructs go-peerstream compatible connections.
41 +// Transport constructs go-stream-muxer compatible connections.
42 type Transport interface {
43
44 // NewConn constructs a new connection
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/spdystream/spdystream.go renamed
+13 -5
@@ -1,14 +1,17 @@
1 package peerstream_spdystream
2
3 import (
4 + "errors"
5 "net"
6 "net/http"
7
8 ss "github.com/docker/spdystream"
8 - pst "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
9 + smux "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer"
10 )
11
11 -// stream implements pst.Stream using a ss.Stream
12 +var ErrUseServe = errors.New("not implemented, use Serve")
13 +
14 +// stream implements smux.Stream using a ss.Stream
15 type stream ss.Stream
16
17 func (s *stream) spdyStream() *ss.Stream {
@@ -63,7 +66,7 @@ func (c *conn) IsClosed() bool {
66 }
67
68 // OpenStream creates a new stream.
66 -func (c *conn) OpenStream() (pst.Stream, error) {
69 +func (c *conn) OpenStream() (smux.Stream, error) {
70 s, err := c.spdyConn().CreateStream(http.Header{
71 ":method": []string{"GET"}, // this is here for HTTP/SPDY interop
72 ":path": []string{"/"}, // this is here for HTTP/SPDY interop
@@ -78,9 +81,14 @@ func (c *conn) OpenStream() (pst.Stream, error) {
81 return (*stream)(s), nil
82 }
83
84 +// AcceptStream accepts a stream opened by the other side.
85 +func (c *conn) AcceptStream() (smux.Stream, error) {
86 + return nil, ErrUseServe
87 +}
88 +
89 // Serve starts listening for incoming requests and handles them
90 // using given StreamHandler
83 -func (c *conn) Serve(handler pst.StreamHandler) {
91 +func (c *conn) Serve(handler smux.StreamHandler) {
92 c.spdyConn().Serve(func(s *ss.Stream) {
93
94 // Flow control and backpressure of Opening streams is broken.
@@ -109,7 +117,7 @@ type transport struct{}
117 // spdystream-backed connections.
118 var Transport = transport{}
119
112 -func (t transport) NewConn(nc net.Conn, isServer bool) (pst.Conn, error) {
120 +func (t transport) NewConn(nc net.Conn, isServer bool) (smux.Conn, error) {
121 sc, err := ss.NewConnection(nc, isServer)
122 return &conn{sc: sc, closed: make(chan struct{})}, err
123 }
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/spdystream/spdystream_test.go new
+11
@@ -0,0 +1,11 @@
1 +package peerstream_spdystream
2 +
3 +import (
4 + "testing"
5 +
6 + test "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/test"
7 +)
8 +
9 +func TestSpdyStreamTransport(t *testing.T) {
10 + test.SubtestAll(t, Transport)
11 +}
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/test/ttest.go new
+378
@@ -0,0 +1,378 @@
1 +package sm_test
2 +
3 +import (
4 + "bytes"
5 + crand "crypto/rand"
6 + "fmt"
7 + "io"
8 + mrand "math/rand"
9 + "net"
10 + "os"
11 + "reflect"
12 + "runtime"
13 + "runtime/debug"
14 + "sync"
15 + "testing"
16 +
17 + smux "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer"
18 +)
19 +
20 +var randomness []byte
21 +
22 +func init() {
23 + // read 1MB of randomness
24 + randomness = make([]byte, 1<<20)
25 + if _, err := crand.Read(randomness); err != nil {
26 + panic(err)
27 + }
28 +}
29 +
30 +type Options struct {
31 + tr smux.Transport
32 + connNum int
33 + streamNum int
34 + msgNum int
35 + msgMin int
36 + msgMax int
37 +}
38 +
39 +func randBuf(size int) []byte {
40 + n := len(randomness) - size
41 + if size < 1 {
42 + panic(fmt.Errorf("requested too large buffer (%d). max is %d", size, len(randomness)))
43 + }
44 +
45 + start := mrand.Intn(n)
46 + return randomness[start : start+size]
47 +}
48 +
49 +func checkErr(t *testing.T, err error) {
50 + if err != nil {
51 + debug.PrintStack()
52 + t.Fatal(err)
53 + }
54 +}
55 +
56 +func log(s string, v ...interface{}) {
57 + if testing.Verbose() {
58 + fmt.Fprintf(os.Stderr, "> "+s+"\n", v...)
59 + }
60 +}
61 +
62 +func echoStream(s smux.Stream) {
63 + defer s.Close()
64 + log("accepted stream")
65 + io.Copy(&LogWriter{s}, s) // echo everything
66 + log("closing stream")
67 +}
68 +
69 +type LogWriter struct {
70 + W io.Writer
71 +}
72 +
73 +func (lw *LogWriter) Write(buf []byte) (int, error) {
74 + if testing.Verbose() {
75 + log("logwriter: writing %d bytes", len(buf))
76 + }
77 + return lw.W.Write(buf)
78 +}
79 +
80 +func GoServe(t *testing.T, tr smux.Transport, l net.Listener) (done func()) {
81 + closed := make(chan struct{}, 1)
82 +
83 + go func() {
84 + for {
85 + c1, err := l.Accept()
86 + if err != nil {
87 + select {
88 + case <-closed:
89 + return // closed naturally.
90 + default:
91 + checkErr(t, err)
92 + }
93 + }
94 +
95 + log("accepted connection")
96 + sc1, err := tr.NewConn(c1, true)
97 + checkErr(t, err)
98 + go sc1.Serve(echoStream)
99 + }
100 + }()
101 +
102 + return func() {
103 + closed <- struct{}{}
104 + }
105 +}
106 +
107 +func SubtestSimpleWrite(t *testing.T, tr smux.Transport) {
108 + l, err := net.Listen("tcp", "localhost:0")
109 + checkErr(t, err)
110 + log("listening at %s", l.Addr().String())
111 + done := GoServe(t, tr, l)
112 + defer done()
113 +
114 + log("dialing to %s", l.Addr().String())
115 + nc1, err := net.Dial("tcp", l.Addr().String())
116 + checkErr(t, err)
117 + defer nc1.Close()
118 +
119 + log("wrapping conn")
120 + c1, err := tr.NewConn(nc1, false)
121 + checkErr(t, err)
122 + defer c1.Close()
123 +
124 + // serve the outgoing conn, because some muxers assume
125 + // that we _always_ call serve. (this is an error?)
126 + go c1.Serve(smux.NoOpHandler)
127 +
128 + log("creating stream")
129 + s1, err := c1.OpenStream()
130 + checkErr(t, err)
131 + defer s1.Close()
132 +
133 + buf1 := randBuf(4096)
134 + log("writing %d bytes to stream", len(buf1))
135 + _, err = s1.Write(buf1)
136 + checkErr(t, err)
137 +
138 + buf2 := make([]byte, len(buf1))
139 + log("reading %d bytes from stream (echoed)", len(buf2))
140 + _, err = s1.Read(buf2)
141 + checkErr(t, err)
142 +
143 + if string(buf2) != string(buf1) {
144 + t.Error("buf1 and buf2 not equal: %s != %s", string(buf1), string(buf2))
145 + }
146 + log("done")
147 +}
148 +
149 +func SubtestStress(t *testing.T, opt Options) {
150 +
151 + msgsize := 1 << 11
152 + errs := make(chan error, 0) // dont block anything.
153 +
154 + rateLimitN := 5000 // max of 5k funcs, because -race has 8k max.
155 + rateLimitChan := make(chan struct{}, rateLimitN)
156 + for i := 0; i < rateLimitN; i++ {
157 + rateLimitChan <- struct{}{}
158 + }
159 +
160 + rateLimit := func(f func()) {
161 + <-rateLimitChan
162 + f()
163 + rateLimitChan <- struct{}{}
164 + }
165 +
166 + writeStream := func(s smux.Stream, bufs chan<- []byte) {
167 + log("writeStream %p, %d msgNum", s, opt.msgNum)
168 +
169 + for i := 0; i < opt.msgNum; i++ {
170 + buf := randBuf(msgsize)
171 + bufs <- buf
172 + log("%p writing %d bytes (message %d/%d #%x)", s, len(buf), i, opt.msgNum, buf[:3])
173 + if _, err := s.Write(buf); err != nil {
174 + errs <- fmt.Errorf("s.Write(buf): %s", err)
175 + continue
176 + }
177 + }
178 + }
179 +
180 + readStream := func(s smux.Stream, bufs <-chan []byte) {
181 + log("readStream %p, %d msgNum", s, opt.msgNum)
182 +
183 + buf2 := make([]byte, msgsize)
184 + i := 0
185 + for buf1 := range bufs {
186 + i++
187 + log("%p reading %d bytes (message %d/%d #%x)", s, len(buf1), i-1, opt.msgNum, buf1[:3])
188 +
189 + if _, err := io.ReadFull(s, buf2); err != nil {
190 + errs <- fmt.Errorf("io.ReadFull(s, buf2): %s", err)
191 + log("%p failed to read %d bytes (message %d/%d #%x)", s, len(buf1), i-1, opt.msgNum, buf1[:3])
192 + continue
193 + }
194 + if !bytes.Equal(buf1, buf2) {
195 + errs <- fmt.Errorf("buffers not equal (%x != %x)", buf1[:3], buf2[:3])
196 + }
197 + }
198 + }
199 +
200 + openStreamAndRW := func(c smux.Conn) {
201 + log("openStreamAndRW %p, %d opt.msgNum", c, opt.msgNum)
202 +
203 + s, err := c.OpenStream()
204 + if err != nil {
205 + errs <- fmt.Errorf("Failed to create NewStream: %s", err)
206 + return
207 + }
208 +
209 + bufs := make(chan []byte, opt.msgNum)
210 + go func() {
211 + writeStream(s, bufs)
212 + close(bufs)
213 + }()
214 +
215 + readStream(s, bufs)
216 + s.Close()
217 + }
218 +
219 + openConnAndRW := func() {
220 + log("openConnAndRW")
221 +
222 + l, err := net.Listen("tcp", "localhost:0")
223 + checkErr(t, err)
224 + done := GoServe(t, opt.tr, l)
225 + defer done()
226 +
227 + nla := l.Addr()
228 + nc, err := net.Dial(nla.Network(), nla.String())
229 + checkErr(t, err)
230 + if err != nil {
231 + t.Fatal(fmt.Errorf("net.Dial(%s, %s): %s", nla.Network(), nla.String(), err))
232 + return
233 + }
234 +
235 + c, err := opt.tr.NewConn(nc, false)
236 + if err != nil {
237 + t.Fatal(fmt.Errorf("a.AddConn(%s <--> %s): %s", nc.LocalAddr(), nc.RemoteAddr(), err))
238 + return
239 + }
240 +
241 + // serve the outgoing conn, because some muxers assume
242 + // that we _always_ call serve. (this is an error?)
243 + go c.Serve(func(s smux.Stream) {
244 + log("serving connection")
245 + echoStream(s)
246 + s.Close()
247 + })
248 +
249 + var wg sync.WaitGroup
250 + for i := 0; i < opt.streamNum; i++ {
251 + wg.Add(1)
252 + go rateLimit(func() {
253 + defer wg.Done()
254 + openStreamAndRW(c)
255 + })
256 + }
257 + wg.Wait()
258 + c.Close()
259 + }
260 +
261 + openConnsAndRW := func() {
262 + log("openConnsAndRW, %d conns", opt.connNum)
263 +
264 + var wg sync.WaitGroup
265 + for i := 0; i < opt.connNum; i++ {
266 + wg.Add(1)
267 + go rateLimit(func() {
268 + defer wg.Done()
269 + openConnAndRW()
270 + })
271 + }
272 + wg.Wait()
273 + }
274 +
275 + go func() {
276 + openConnsAndRW()
277 + close(errs) // done
278 + }()
279 +
280 + for err := range errs {
281 + t.Error(err)
282 + }
283 +
284 +}
285 +
286 +func SubtestStress1Conn1Stream1Msg(t *testing.T, tr smux.Transport) {
287 + SubtestStress(t, Options{
288 + tr: tr,
289 + connNum: 1,
290 + streamNum: 1,
291 + msgNum: 1,
292 + msgMax: 100,
293 + msgMin: 100,
294 + })
295 +}
296 +
297 +func SubtestStress1Conn1Stream100Msg(t *testing.T, tr smux.Transport) {
298 + SubtestStress(t, Options{
299 + tr: tr,
300 + connNum: 1,
301 + streamNum: 1,
302 + msgNum: 100,
303 + msgMax: 100,
304 + msgMin: 100,
305 + })
306 +}
307 +
308 +func SubtestStress1Conn100Stream100Msg(t *testing.T, tr smux.Transport) {
309 + SubtestStress(t, Options{
310 + tr: tr,
311 + connNum: 1,
312 + streamNum: 100,
313 + msgNum: 100,
314 + msgMax: 100,
315 + msgMin: 100,
316 + })
317 +}
318 +
319 +func SubtestStress50Conn10Stream50Msg(t *testing.T, tr smux.Transport) {
320 + SubtestStress(t, Options{
321 + tr: tr,
322 + connNum: 50,
323 + streamNum: 10,
324 + msgNum: 50,
325 + msgMax: 100,
326 + msgMin: 100,
327 + })
328 +}
329 +
330 +func SubtestStress1Conn1000Stream10Msg(t *testing.T, tr smux.Transport) {
331 + SubtestStress(t, Options{
332 + tr: tr,
333 + connNum: 1,
334 + streamNum: 1000,
335 + msgNum: 10,
336 + msgMax: 100,
337 + msgMin: 100,
338 + })
339 +}
340 +
341 +func SubtestStress1Conn100Stream100Msg10MB(t *testing.T, tr smux.Transport) {
342 + SubtestStress(t, Options{
343 + tr: tr,
344 + connNum: 1,
345 + streamNum: 100,
346 + msgNum: 100,
347 + msgMax: 10000,
348 + msgMin: 1000,
349 + })
350 +}
351 +
352 +func SubtestAll(t *testing.T, tr smux.Transport) {
353 +
354 + tests := []TransportTest{
355 + SubtestSimpleWrite,
356 + SubtestStress1Conn1Stream1Msg,
357 + SubtestStress1Conn1Stream100Msg,
358 + SubtestStress1Conn100Stream100Msg,
359 + SubtestStress50Conn10Stream50Msg,
360 + SubtestStress1Conn1000Stream10Msg,
361 + SubtestStress1Conn100Stream100Msg10MB,
362 + }
363 +
364 + for _, f := range tests {
365 + if testing.Verbose() {
366 + fmt.Fprintf(os.Stderr, "==== RUN %s\n", GetFunctionName(f))
367 + }
368 + f(t, tr)
369 + }
370 +}
371 +
372 +type TransportTest func(t *testing.T, tr smux.Transport)
373 +
374 +func TestNoOp(t *testing.T) {}
375 +
376 +func GetFunctionName(i interface{}) string {
377 + return runtime.FuncForPC(reflect.ValueOf(i).Pointer()).Name()
378 +}
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/yamux/yamux.go renamed
+14 -8
@@ -1,4 +1,4 @@
1 -package peerstream_yamux
1 +package sm_yamux
2
3 import (
4 "io/ioutil"
@@ -6,10 +6,10 @@ import (
6 "time"
7
8 yamux "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/hashicorp/yamux"
9 - pst "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
9 + smux "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer"
10 )
11
12 -// stream implements pst.Stream using a ss.Stream
12 +// stream implements smux.Stream using a ss.Stream
13 type stream yamux.Stream
14
15 func (s *stream) yamuxStream() *yamux.Stream {
@@ -44,7 +44,7 @@ func (c *conn) IsClosed() bool {
44 }
45
46 // OpenStream creates a new stream.
47 -func (c *conn) OpenStream() (pst.Stream, error) {
47 +func (c *conn) OpenStream() (smux.Stream, error) {
48 s, err := c.yamuxSession().OpenStream()
49 if err != nil {
50 return nil, err
@@ -53,15 +53,21 @@ func (c *conn) OpenStream() (pst.Stream, error) {
53 return (*stream)(s), nil
54 }
55
56 +// AcceptStream accepts a stream opened by the other side.
57 +func (c *conn) AcceptStream() (smux.Stream, error) {
58 + s, err := c.yamuxSession().AcceptStream()
59 + return (*stream)(s), err
60 +}
61 +
62 // Serve starts listening for incoming requests and handles them
63 // using given StreamHandler
58 -func (c *conn) Serve(handler pst.StreamHandler) {
64 +func (c *conn) Serve(handler smux.StreamHandler) {
65 for { // accept loop
60 - s, err := c.yamuxSession().AcceptStream()
66 + s, err := c.AcceptStream()
67 if err != nil {
68 return // err always means closed.
69 }
64 - go handler((*stream)(s))
70 + go handler(s)
71 }
72 }
73
@@ -78,7 +84,7 @@ var DefaultTransport = (*Transport)(&yamux.Config{
84 LogOutput: ioutil.Discard,
85 })
86
81 -func (t *Transport) NewConn(nc net.Conn, isServer bool) (pst.Conn, error) {
87 +func (t *Transport) NewConn(nc net.Conn, isServer bool) (smux.Conn, error) {
88 var s *yamux.Session
89 var err error
90 if isServer {
Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/yamux/yamux_test.go new
+11
@@ -0,0 +1,11 @@
1 +package sm_yamux
2 +
3 +import (
4 + "testing"
5 +
6 + test "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/test"
7 +)
8 +
9 +func TestYamuxTransport(t *testing.T) {
10 + test.SubtestAll(t, DefaultTransport)
11 +}
Godeps/_workspace/src/github.com/whyrusleeping/go-multiplex/.gitignore new
+1
@@ -0,0 +1 @@
1 +*.swp
Godeps/_workspace/src/github.com/whyrusleeping/go-multiplex/.gxlastpubver new
+1
@@ -0,0 +1 @@
1 +QmYmW76Y7NxvWwW3GPUmpbHd4mm1iW14f5UimeJ9UoofEs
\ No newline at end of file
Godeps/_workspace/src/github.com/whyrusleeping/go-multiplex/README.md new
+18
@@ -0,0 +1,18 @@
1 +# go-multiplex
2 +
3 +A super simple stream muxing library compatible with [multiplex](http://github.com/maxogden/multiplex)
4 +
5 +## Usage
6 +
7 +```go
8 +mplex := multiplex.NewMultiplex(mysocket)
9 +
10 +s := mplex.NewStream()
11 +s.Write([]byte("Hello World!")
12 +s.Close()
13 +
14 +mplex.Serve(func(s *multiplex.Stream) {
15 + // echo back everything received
16 + io.Copy(s, s)
17 +})
18 +```
Godeps/_workspace/src/github.com/whyrusleeping/go-multiplex/multiplex.go new
+393
@@ -0,0 +1,393 @@
1 +package multiplex
2 +
3 +import (
4 + "bufio"
5 + "encoding/binary"
6 + "errors"
7 + "fmt"
8 + "io"
9 + "io/ioutil"
10 + "sync"
11 +)
12 +
13 +const (
14 + NewStream = iota
15 + Receiver
16 + Initiator
17 + Unknown
18 + Close
19 +)
20 +
21 +var _ = ioutil.ReadAll
22 +var _ = bufio.NewReadWriter
23 +var _ = binary.MaxVarintLen16
24 +
25 +type msg struct {
26 + header uint64
27 + data []byte
28 + err chan<- error
29 +}
30 +
31 +type Stream struct {
32 + id uint64
33 + name string
34 + header uint64
35 + closed chan struct{}
36 + data_in chan []byte
37 + data_out chan<- msg
38 + extra []byte
39 +}
40 +
41 +func newStream(id uint64, name string, initiator bool, send chan<- msg) *Stream {
42 + var hfn uint64
43 + if initiator {
44 + hfn = 2
45 + } else {
46 + hfn = 1
47 + }
48 + return &Stream{
49 + id: id,
50 + name: name,
51 + header: (id << 3) | hfn,
52 + data_in: make(chan []byte, 8),
53 + data_out: send,
54 + closed: make(chan struct{}),
55 + }
56 +}
57 +
58 +func (s *Stream) Name() string {
59 + return s.name
60 +}
61 +
62 +func (s *Stream) receive(b []byte) {
63 + select {
64 + case s.data_in <- b:
65 + case <-s.closed:
66 + }
67 +}
68 +
69 +func (m *Multiplex) Accept() (*Stream, error) {
70 + select {
71 + case s, ok := <-m.nstreams:
72 + if !ok {
73 + return nil, errors.New("multiplex closed")
74 + }
75 + return s, nil
76 + case err := <-m.errs:
77 + return nil, err
78 + case <-m.closed:
79 + return nil, errors.New("multiplex closed")
80 + }
81 +}
82 +
83 +func (s *Stream) Read(b []byte) (int, error) {
84 + if s.extra == nil {
85 + select {
86 + case <-s.closed:
87 + return 0, io.EOF
88 + case read, ok := <-s.data_in:
89 + if !ok {
90 + return 0, io.EOF
91 + }
92 + s.extra = read
93 + }
94 + }
95 + n := copy(b, s.extra)
96 + if n < len(s.extra) {
97 + s.extra = s.extra[n:]
98 + } else {
99 + s.extra = nil
100 + }
101 + return n, nil
102 +}
103 +
104 +func (s *Stream) Write(b []byte) (int, error) {
105 + errs := make(chan error, 1)
106 + select {
107 + case s.data_out <- msg{header: s.header, data: b, err: errs}:
108 + select {
109 + case err := <-errs:
110 + return len(b), err
111 + case <-s.closed:
112 + return 0, errors.New("stream closed")
113 + }
114 +
115 + case <-s.closed:
116 + return 0, errors.New("stream closed")
117 + }
118 +}
119 +
120 +func (s *Stream) Close() error {
121 + select {
122 + case <-s.closed:
123 + return nil
124 + default:
125 + close(s.closed)
126 + select {
127 + case s.data_out <- msg{
128 + header: (s.id << 3) | Close,
129 + err: make(chan error, 1), //throw away error, whatever
130 + }:
131 + default:
132 + }
133 + close(s.data_in)
134 + return nil
135 + }
136 +}
137 +
138 +type Multiplex struct {
139 + con io.ReadWriteCloser
140 + buf *bufio.Reader
141 + nextID uint64
142 + outchan chan msg
143 + closed chan struct{}
144 + initiator bool
145 +
146 + nstreams chan *Stream
147 + errs chan error
148 +
149 + channels map[uint64]*Stream
150 + ch_lock sync.Mutex
151 +}
152 +
153 +func NewMultiplex(con io.ReadWriteCloser, initiator bool) *Multiplex {
154 + mp := &Multiplex{
155 + con: con,
156 + initiator: initiator,
157 + buf: bufio.NewReader(con),
158 + channels: make(map[uint64]*Stream),
159 + outchan: make(chan msg),
160 + closed: make(chan struct{}),
161 + nstreams: make(chan *Stream, 16),
162 + errs: make(chan error),
163 + }
164 +
165 + go mp.handleOutgoing()
166 + go mp.handleIncoming()
167 +
168 + return mp
169 +}
170 +
171 +func (mp *Multiplex) Close() error {
172 + if mp.IsClosed() {
173 + return nil
174 + }
175 + close(mp.closed)
176 + mp.ch_lock.Lock()
177 + defer mp.ch_lock.Unlock()
178 + for _, s := range mp.channels {
179 + err := s.Close()
180 + if err != nil {
181 + return err
182 + }
183 + }
184 + return nil
185 +}
186 +
187 +func (mp *Multiplex) IsClosed() bool {
188 + select {
189 + case <-mp.closed:
190 + return true
191 + default:
192 + return false
193 + }
194 +}
195 +
196 +func (mp *Multiplex) handleOutgoing() {
197 + for {
198 + select {
199 + case msg, ok := <-mp.outchan:
200 + if !ok {
201 + return
202 + }
203 +
204 + buf := EncodeVarint(msg.header)
205 + _, err := mp.con.Write(buf)
206 + if err != nil {
207 + msg.err <- err
208 + continue
209 + }
210 +
211 + buf = EncodeVarint(uint64(len(msg.data)))
212 + _, err = mp.con.Write(buf)
213 + if err != nil {
214 + msg.err <- err
215 + continue
216 + }
217 +
218 + _, err = mp.con.Write(msg.data)
219 + if err != nil {
220 + msg.err <- err
221 + continue
222 + }
223 +
224 + msg.err <- nil
225 + case <-mp.closed:
226 + return
227 + }
228 + }
229 +}
230 +
231 +func (mp *Multiplex) nextChanID() (out uint64) {
232 + if mp.initiator {
233 + out = mp.nextID + 1
234 + } else {
235 + out = mp.nextID
236 + }
237 + mp.nextID += 2
238 + return
239 +}
240 +
241 +func (mp *Multiplex) NewStream() *Stream {
242 + return mp.NewNamedStream("")
243 +}
244 +
245 +func (mp *Multiplex) NewNamedStream(name string) *Stream {
246 + mp.ch_lock.Lock()
247 + sid := mp.nextChanID()
248 + header := (sid << 3) | NewStream
249 +
250 + if name == "" {
251 + name = fmt.Sprint(sid)
252 + }
253 + s := newStream(sid, name, true, mp.outchan)
254 + mp.channels[sid] = s
255 + mp.ch_lock.Unlock()
256 +
257 + mp.outchan <- msg{
258 + header: header,
259 + data: []byte(name),
260 + err: make(chan error, 1), //throw away error
261 + }
262 +
263 + return s
264 +}
265 +
266 +func (mp *Multiplex) sendErr(err error) {
267 + select {
268 + case mp.errs <- err:
269 + case <-mp.closed:
270 + }
271 +}
272 +
273 +func (mp *Multiplex) handleIncoming() {
274 + defer mp.shutdown()
275 + for {
276 + ch, tag, err := mp.readNextHeader()
277 + if err != nil {
278 + mp.sendErr(err)
279 + return
280 + }
281 +
282 + b, err := mp.readNext()
283 + if err != nil {
284 + mp.sendErr(err)
285 + return
286 + }
287 +
288 + mp.ch_lock.Lock()
289 + msch, ok := mp.channels[ch]
290 + if !ok {
291 + var name string
292 + if tag == NewStream {
293 + name = string(b)
294 + }
295 + msch = newStream(ch, name, false, mp.outchan)
296 + mp.channels[ch] = msch
297 + select {
298 + case mp.nstreams <- msch:
299 + case <-mp.closed:
300 + return
301 + }
302 + if tag == NewStream {
303 + mp.ch_lock.Unlock()
304 + continue
305 + }
306 + }
307 + mp.ch_lock.Unlock()
308 +
309 + if tag == Close {
310 + msch.Close()
311 + mp.ch_lock.Lock()
312 + delete(mp.channels, ch)
313 + mp.ch_lock.Unlock()
314 + continue
315 + }
316 +
317 + msch.receive(b)
318 + }
319 +}
320 +
321 +func (mp *Multiplex) shutdown() {
322 + mp.ch_lock.Lock()
323 + defer mp.ch_lock.Unlock()
324 + for _, s := range mp.channels {
325 + s.Close()
326 + }
327 +}
328 +
329 +func (mp *Multiplex) readNextHeader() (uint64, uint64, error) {
330 + h, _, err := DecodeVarint(mp.buf)
331 + if err != nil {
332 + return 0, 0, err
333 + }
334 +
335 + // get channel ID
336 + ch := h >> 3
337 +
338 + rem := h & 7
339 +
340 + return ch, rem, nil
341 +}
342 +
343 +func (mp *Multiplex) readNext() ([]byte, error) {
344 + // get length
345 + l, _, err := DecodeVarint(mp.buf)
346 + if err != nil {
347 + return nil, err
348 + }
349 +
350 + buf := make([]byte, l)
351 + n, err := io.ReadFull(mp.buf, buf)
352 + if err != nil {
353 + return nil, err
354 + }
355 +
356 + if n != int(l) {
357 + panic("NOT THE SAME")
358 + }
359 +
360 + return buf, nil
361 +}
362 +
363 +func EncodeVarint(x uint64) []byte {
364 + var buf [10]byte
365 + var n int
366 + for n = 0; x > 127; n++ {
367 + buf[n] = 0x80 | uint8(x&0x7F)
368 + x >>= 7
369 + }
370 + buf[n] = uint8(x)
371 + n++
372 + return buf[0:n]
373 +}
374 +
375 +func DecodeVarint(r *bufio.Reader) (x uint64, n int, err error) {
376 + // x, n already 0
377 + for shift := uint(0); shift < 64; shift += 7 {
378 + val, err := r.ReadByte()
379 + if err != nil {
380 + return 0, 0, err
381 + }
382 +
383 + b := uint64(val)
384 + n++
385 + x |= (b & 0x7F) << shift
386 + if (b & 0x80) == 0 {
387 + return x, n, nil
388 + }
389 + }
390 +
391 + // The number is too large to represent in a 64-bit value.
392 + return 0, 0, errors.New("Too large of a number!")
393 +}
Godeps/_workspace/src/github.com/whyrusleeping/go-multiplex/multiplex_test.go new
+118
@@ -0,0 +1,118 @@
1 +package multiplex
2 +
3 +import (
4 + "fmt"
5 + "io"
6 + "net"
7 + "testing"
8 +
9 + rand "github.com/dustin/randbo"
10 +)
11 +
12 +func TestBasicStreams(t *testing.T) {
13 + a, b := net.Pipe()
14 +
15 + mpa := NewMultiplex(a, false)
16 + mpb := NewMultiplex(b, true)
17 +
18 + mes := []byte("Hello world")
19 + go func() {
20 + s, err := mpb.Accept()
21 + if err != nil {
22 + t.Fatal(err)
23 + }
24 +
25 + _, err = s.Write(mes)
26 + if err != nil {
27 + t.Fatal(err)
28 + }
29 +
30 + err = s.Close()
31 + if err != nil {
32 + t.Fatal(err)
33 + }
34 + }()
35 +
36 + s := mpa.NewStream()
37 +
38 + buf := make([]byte, len(mes))
39 + n, err := s.Read(buf)
40 + if err != nil {
41 + t.Fatal(err)
42 + }
43 +
44 + if n != len(mes) {
45 + t.Fatal("read wrong amount")
46 + }
47 +
48 + if string(buf) != string(mes) {
49 + t.Fatal("got bad data")
50 + }
51 +
52 + s.Close()
53 +
54 + mpa.Close()
55 + mpb.Close()
56 +}
57 +
58 +func TestEcho(t *testing.T) {
59 + a, b := net.Pipe()
60 +
61 + mpa := NewMultiplex(a, false)
62 + mpb := NewMultiplex(b, true)
63 +
64 + mes := make([]byte, 40960)
65 + rand.New().Read(mes)
66 + go func() {
67 + s, err := mpb.Accept()
68 + if err != nil {
69 + t.Fatal(err)
70 + }
71 +
72 + defer s.Close()
73 + io.Copy(s, s)
74 + }()
75 +
76 + s := mpa.NewStream()
77 +
78 + _, err := s.Write(mes)
79 + if err != nil {
80 + t.Fatal(err)
81 + }
82 +
83 + buf := make([]byte, len(mes))
84 + n, err := io.ReadFull(s, buf)
85 + if err != nil {
86 + t.Fatal(err)
87 + }
88 +
89 + if n != len(mes) {
90 + t.Fatal("read wrong amount")
91 + }
92 +
93 + if err := arrComp(buf, mes); err != nil {
94 + t.Fatal(err)
95 + }
96 + s.Close()
97 +
98 + mpa.Close()
99 + mpb.Close()
100 +}
101 +
102 +func arrComp(a, b []byte) error {
103 + msg := ""
104 + if len(a) != len(b) {
105 + msg += fmt.Sprintf("arrays differ in length: %d %d\n", len(a), len(b))
106 + }
107 +
108 + for i := 0; i < len(a) && i < len(b); i++ {
109 + if a[i] != b[i] {
110 + msg += fmt.Sprintf("content differs at index %d [%d != %d]", i, a[i], b[i])
111 + return fmt.Errorf(msg)
112 + }
113 + }
114 + if len(msg) > 0 {
115 + return fmt.Errorf(msg)
116 + }
117 + return nil
118 +}
Godeps/_workspace/src/github.com/whyrusleeping/go-multiplex/package.json new
+5
@@ -0,0 +1,5 @@
1 +{
2 + "name": "go-multiplex",
3 + "version": "1.0.0",
4 + "language": "go"
5 +}
\ No newline at end of file
p2p/net/swarm/swarm.go
+2 -2
@@ -16,8 +16,8 @@ import (
16
17 ma "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
18 ps "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream"
19 - pst "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
20 - psy "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/yamux"
19 + pst "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer"
20 + psy "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-stream-muxer/yamux"
21 "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
22 goprocessctx "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/context"
23 prom "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/prometheus/client_golang/prometheus"