p2p/net: notify on listens
Network now signals when it successfully listens on some address or when an address shuts down. This will be used to establish and close nat port mappings. It could also be used to notify peers of address changes.
Juan Batiz-Benet committed
Jan 30, 2015 at 20:17 UTC
98f2b0779f9a2c42b10415c969e1d14106c36bfd
8 files changed
+64
-12
exchange/bitswap/network/ipfs_impl.go
+3
@@ -2,6 +2,7 @@ package network
2
3
import (
4
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
5
+ ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
6
7
bsmsg "github.com/jbenet/go-ipfs/exchange/bitswap/message"
8
host "github.com/jbenet/go-ipfs/p2p/host"
@@ -171,3 +172,5 @@ func (nn *netNotifiee) Disconnected(n inet.Network, v inet.Conn) {
172
173
func (nn *netNotifiee) OpenedStream(n inet.Network, v inet.Stream) {}
174
func (nn *netNotifiee) ClosedStream(n inet.Network, v inet.Stream) {}
175
+func (nn *netNotifiee) Listen(n inet.Network, a ma.Multiaddr) {}
176
+func (nn *netNotifiee) ListenClose(n inet.Network, a ma.Multiaddr) {}
p2p/net/interface.go
+6
-4
@@ -142,10 +142,12 @@ const (
142
// Notifiee is an interface for an object wishing to receive
143
// notifications from a Network.
144
type Notifiee interface {
145
- Connected(Network, Conn) // called when a connection opened
146
- Disconnected(Network, Conn) // called when a connection closed
147
- OpenedStream(Network, Stream) // called when a stream opened
148
- ClosedStream(Network, Stream) // called when a stream closed
145
+ Listen(Network, ma.Multiaddr) // called when network starts listening on an addr
146
+ ListenClose(Network, ma.Multiaddr) // called when network starts listening on an addr
147
+ Connected(Network, Conn) // called when a connection opened
148
+ Disconnected(Network, Conn) // called when a connection closed
149
+ OpenedStream(Network, Stream) // called when a stream opened
150
+ ClosedStream(Network, Stream) // called when a stream closed
151
152
// TODO
153
// PeerConnected(Network, peer.ID) // called when a peer connected
p2p/net/mock/mock_notif_test.go
+13
-2
@@ -4,9 +4,10 @@ import (
4
"testing"
5
"time"
6
7
- inet "github.com/jbenet/go-ipfs/p2p/net"
8
-
7
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
8
+ ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
9
+
10
+ inet "github.com/jbenet/go-ipfs/p2p/net"
11
)
12
13
func TestNotifications(t *testing.T) {
@@ -169,6 +170,8 @@ func TestNotifications(t *testing.T) {
170
}
171
172
type netNotifiee struct {
173
+ listen chan ma.Multiaddr
174
+ listenClose chan ma.Multiaddr
175
connected chan inet.Conn
176
disconnected chan inet.Conn
177
openedStream chan inet.Stream
@@ -177,6 +180,8 @@ type netNotifiee struct {
180
181
func newNetNotifiee() *netNotifiee {
182
return &netNotifiee{
183
+ listen: make(chan ma.Multiaddr),
184
+ listenClose: make(chan ma.Multiaddr),
185
connected: make(chan inet.Conn),
186
disconnected: make(chan inet.Conn),
187
openedStream: make(chan inet.Stream),
@@ -184,6 +189,12 @@ func newNetNotifiee() *netNotifiee {
189
}
190
}
191
192
+func (nn *netNotifiee) Listen(n inet.Network, a ma.Multiaddr) {
193
+ nn.listen <- a
194
+}
195
+func (nn *netNotifiee) ListenClose(n inet.Network, a ma.Multiaddr) {
196
+ nn.listenClose <- a
197
+}
198
func (nn *netNotifiee) Connected(n inet.Network, v inet.Conn) {
199
nn.connected <- v
200
}
p2p/net/swarm/swarm.go
+9
@@ -202,6 +202,15 @@ func (s *Swarm) LocalPeer() peer.ID {
202
return s.local
203
}
204
205
+// notifyAll sends a signal to all Notifiees
206
+func (s *Swarm) notifyAll(notify func(inet.Notifiee)) {
207
+ s.notifmu.RLock()
208
+ for f := range s.notifs {
209
+ go notify(f)
210
+ }
211
+ s.notifmu.RUnlock()
212
+}
213
+
214
// Notify signs up Notifiee to receive signals when events happen
215
func (s *Swarm) Notify(f inet.Notifiee) {
216
// wrap with our notifiee, to translate function calls
p2p/net/swarm/swarm_dial.go
+1
-1
@@ -382,7 +382,7 @@ func (s *Swarm) dialAddrs(ctx context.Context, d *conn.Dialer, p peer.ID, remote
382
for i := 0; i < len(remoteAddrs); i++ {
383
select {
384
case err = <-errs:
385
- log.Info(err)
385
+ log.Debug(err)
386
case connC := <-conns:
387
// take the first + return asap
388
close(foundConn)
p2p/net/swarm/swarm_listen.go
+15
-3
@@ -3,6 +3,7 @@ package swarm
3
import (
4
"fmt"
5
6
+ inet "github.com/jbenet/go-ipfs/p2p/net"
7
conn "github.com/jbenet/go-ipfs/p2p/net/conn"
8
addrutil "github.com/jbenet/go-ipfs/p2p/net/swarm/addr"
9
lgbl "github.com/jbenet/go-ipfs/util/eventlog/loggables"
@@ -60,7 +61,7 @@ func (s *Swarm) setupListener(maddr ma.Multiaddr) error {
61
// may be fine for sk to be nil, just log a warning.
62
log.Warning("Listener not given PrivateKey, so WILL NOT SECURE conns.")
63
}
63
- log.Infof("Swarm Listening at %s", maddr)
64
+ log.Debugf("Swarm Listening at %s", maddr)
65
list, err := conn.Listen(s.cg.Context(), maddr, s.local, sk)
66
if err != nil {
67
return err
@@ -72,20 +73,31 @@ func (s *Swarm) setupListener(maddr ma.Multiaddr) error {
73
if err != nil {
74
return err
75
}
75
- log.Infof("Swarm Listeners at %s", s.ListenAddresses())
76
+ log.Debugf("Swarm Listeners at %s", s.ListenAddresses())
77
+
78
+ // signal to our notifiees on successful conn.
79
+ s.notifyAll(func(n inet.Notifiee) {
80
+ n.Listen((*Network)(s), maddr)
81
+ })
82
83
// go consume peerstream's listen accept errors. note, these ARE errors.
84
// they may be killing the listener, and if we get _any_ we should be
85
// fixing this in our conn.Listener (to ignore them or handle them
86
// differently.)
87
go func(ctx context.Context, sl *ps.Listener) {
88
+
89
+ // signal to our notifiees closing
90
+ defer s.notifyAll(func(n inet.Notifiee) {
91
+ n.ListenClose((*Network)(s), maddr)
92
+ })
93
+
94
for {
95
select {
96
case err, more := <-sl.AcceptErrors():
97
if !more {
98
return
99
}
88
- log.Info(err)
100
+ log.Debugf("swarm listener accept error: %s", err)
101
case <-ctx.Done():
102
return
103
}
p2p/net/swarm/swarm_notif_test.go
+13
-2
@@ -4,9 +4,10 @@ import (
4
"testing"
5
"time"
6
7
- inet "github.com/jbenet/go-ipfs/p2p/net"
8
-
7
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
8
+ ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
9
+
10
+ inet "github.com/jbenet/go-ipfs/p2p/net"
11
)
12
13
func TestNotifications(t *testing.T) {
@@ -157,6 +158,8 @@ func TestNotifications(t *testing.T) {
158
}
159
160
type netNotifiee struct {
161
+ listen chan ma.Multiaddr
162
+ listenClose chan ma.Multiaddr
163
connected chan inet.Conn
164
disconnected chan inet.Conn
165
openedStream chan inet.Stream
@@ -165,6 +168,8 @@ type netNotifiee struct {
168
169
func newNetNotifiee() *netNotifiee {
170
return &netNotifiee{
171
+ listen: make(chan ma.Multiaddr),
172
+ listenClose: make(chan ma.Multiaddr),
173
connected: make(chan inet.Conn),
174
disconnected: make(chan inet.Conn),
175
openedStream: make(chan inet.Stream),
@@ -172,6 +177,12 @@ func newNetNotifiee() *netNotifiee {
177
}
178
}
179
180
+func (nn *netNotifiee) Listen(n inet.Network, a ma.Multiaddr) {
181
+ nn.listen <- a
182
+}
183
+func (nn *netNotifiee) ListenClose(n inet.Network, a ma.Multiaddr) {
184
+ nn.listenClose <- a
185
+}
186
func (nn *netNotifiee) Connected(n inet.Network, v inet.Conn) {
187
nn.connected <- v
188
}
routing/dht/notif.go
+4
@@ -1,6 +1,8 @@
1
package dht
2
3
import (
4
+ ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
5
+
6
inet "github.com/jbenet/go-ipfs/p2p/net"
7
)
8
@@ -31,3 +33,5 @@ func (nn *netNotifiee) Disconnected(n inet.Network, v inet.Conn) {
33
34
func (nn *netNotifiee) OpenedStream(n inet.Network, v inet.Stream) {}
35
func (nn *netNotifiee) ClosedStream(n inet.Network, v inet.Stream) {}
36
+func (nn *netNotifiee) Listen(n inet.Network, a ma.Multiaddr) {}
37
+func (nn *netNotifiee) ListenClose(n inet.Network, a ma.Multiaddr) {}