updated peerstream
Juan Batiz-Benet committed
Jan 24, 2015 at 05:36 UTC
b63cbf5331495e9ab9683d98b7efde34bc7d955a
6 files changed
+65
-5
Godeps/Godeps.json
+1
-1
@@ -165,7 +165,7 @@
165
},
166
{
167
"ImportPath": "github.com/jbenet/go-peerstream",
168
- "Rev": "530b09b2300da11cc19f479289be5d014c146581"
168
+ "Rev": "bbe2a6461aa80ee25fd87eccf35bd54bac7f788d"
169
},
170
{
171
"ImportPath": "github.com/jbenet/go-random",
Godeps/_workspace/src/github.com/jbenet/go-peerstream/.gitignore
new
+1
@@ -0,0 +1 @@
1
+example/closer
Godeps/_workspace/src/github.com/jbenet/go-peerstream/.travis.yml
+1
-2
@@ -1,11 +1,10 @@
1
language: go
2
3
go:
4
- - 1.2
4
- 1.3
5
- 1.4
6
- release
7
- tip
8
9
script:
11
- - go test -race -cpu=5 -v ./...
10
+ - go test -race -cpu=5 ./...
Godeps/_workspace/src/github.com/jbenet/go-peerstream/conn.go
+17
-2
@@ -122,7 +122,11 @@ func (c *Conn) Close() error {
122
123
// close underlying connection
124
c.swarm.removeConn(c)
125
- return c.pstConn.Close()
125
+ err := c.pstConn.Close()
126
+ c.swarm.notifyAll(func(n Notifiee) {
127
+ n.Disconnected(c)
128
+ })
129
+ return err
130
}
131
132
// ConnsWithGroup narrows down a set of connections to those in a given group.
@@ -198,6 +202,9 @@ func (s *Swarm) addConn(netConn net.Conn, isServer bool) (*Conn, error) {
202
s.StreamHandler()(stream) // call our handler
203
})
204
205
+ s.notifyAll(func(n Notifiee) {
206
+ n.Connected(c)
207
+ })
208
return c, nil
209
}
210
@@ -228,6 +235,10 @@ func (s *Swarm) setupStream(pstStream pst.Stream, c *Conn) *Stream {
235
c.streams[stream] = struct{}{}
236
s.streamLock.Unlock()
237
c.streamLock.Unlock()
238
+
239
+ s.notifyAll(func(n Notifiee) {
240
+ n.OpenedStream(stream)
241
+ })
242
return stream
243
}
244
@@ -241,7 +252,11 @@ func (s *Swarm) removeStream(stream *Stream) error {
252
s.streamLock.Unlock()
253
stream.conn.streamLock.Unlock()
254
244
- return stream.pstStream.Close()
255
+ err := stream.pstStream.Close()
256
+ s.notifyAll(func(n Notifiee) {
257
+ n.ClosedStream(stream)
258
+ })
259
+ return err
260
}
261
262
func (s *Swarm) removeConn(conn *Conn) {
Godeps/_workspace/src/github.com/jbenet/go-peerstream/swarm.go
+39
@@ -39,6 +39,10 @@ type Swarm struct {
39
streamHandler StreamHandler // receives Streams initiated remotely
40
selectConn SelectConn // default SelectConn function
41
42
+ // notification listeners
43
+ notifiees map[Notifiee]struct{}
44
+ notifieeLock sync.RWMutex
45
+
46
closed chan struct{}
47
}
48
@@ -48,6 +52,7 @@ func NewSwarm(t pst.Transport) *Swarm {
52
streams: make(map[*Stream]struct{}),
53
conns: make(map[*Conn]struct{}),
54
listeners: make(map[*Listener]struct{}),
55
+ notifiees: make(map[Notifiee]struct{}),
56
selectConn: SelectRandomConn,
57
streamHandler: NoOpStreamHandler,
58
connHandler: NoOpConnHandler,
@@ -361,3 +366,37 @@ func (s *Swarm) connGarbageCollect() {
366
}
367
}
368
}
369
+
370
+// Notify signs up Notifiee to receive signals when events happen
371
+func (s *Swarm) Notify(n Notifiee) {
372
+ s.notifieeLock.Lock()
373
+ s.notifiees[n] = struct{}{}
374
+ s.notifieeLock.Unlock()
375
+}
376
+
377
+// StopNotify unregisters Notifiee fromr receiving signals
378
+func (s *Swarm) StopNotify(n Notifiee) {
379
+ s.notifieeLock.Lock()
380
+ delete(s.notifiees, n)
381
+ s.notifieeLock.Unlock()
382
+}
383
+
384
+// notifyAll runs the notification function on all Notifiees
385
+func (s *Swarm) notifyAll(notification func(n Notifiee)) {
386
+ s.notifieeLock.RLock()
387
+ for n := range s.notifiees {
388
+ // make sure we dont block
389
+ // and they dont block each other.
390
+ go notification(n)
391
+ }
392
+ s.notifieeLock.RUnlock()
393
+}
394
+
395
+// Notifiee is an interface for an object wishing to receive
396
+// notifications from a Swarm
397
+type Notifiee interface {
398
+ Connected(*Conn) // called when a connection opened
399
+ Disconnected(*Conn) // called when a connection closed
400
+ OpenedStream(*Stream) // called when a stream opened
401
+ ClosedStream(*Stream) // called when a stream closed
402
+}
Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/test/ttest.go
+6
@@ -350,6 +350,8 @@ func SubtestStressNSwarmNConnNStreamNMsg(t *testing.T, tr pst.Transport, nSwarm,
350
for _, a := range swarms {
351
for _, b := range swarms {
352
wg.Add(1)
353
+ a := a // race
354
+ b := b // race
355
go rateLimit(func() {
356
defer wg.Done()
357
openConnsAndRW(a, b)
@@ -370,6 +372,10 @@ func SubtestStressNSwarmNConnNStreamNMsg(t *testing.T, tr pst.Transport, nSwarm,
372
t.Error(err)
373
}
374
375
+ for _, s := range swarms {
376
+ s.Close()
377
+ }
378
+
379
}
380
381
func SubtestStress1Swarm1Conn1Stream1Msg(t *testing.T, tr pst.Transport) {