p2p: deduplicate some listeners logic
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Sep 6, 2018 at 11:07 UTC
228a71aef271e5a2ce6701d8e8a8a3a60ef3d36b
6 files changed
+39
-111
core/commands/p2p.go
+3
-3
@@ -324,7 +324,7 @@ var p2pCloseCmd = &cmds.Command{
324
return
325
}
326
327
- match := func(listener p2p.ListenerLocal) bool {
327
+ match := func(listener p2p.Listener) bool {
328
if closeAll {
329
return true
330
}
@@ -340,7 +340,7 @@ var p2pCloseCmd = &cmds.Command{
340
return true
341
}
342
343
- todo := make([]p2p.ListenerLocal, 0)
343
+ todo := make([]p2p.Listener, 0)
344
n.P2P.ListenersLocal.Lock()
345
for _, l := range n.P2P.ListenersLocal.Listeners {
346
if !match(l) {
@@ -389,7 +389,7 @@ var p2pCloseCmd = &cmds.Command{
389
}
390
391
///////
392
-// Listener
392
+// Stream
393
//
394
395
// p2pStreamCmd is the 'ipfs p2p stream' command
p2p/listener.go
renamed
+20
-14
@@ -12,30 +12,36 @@ import (
12
)
13
14
// Listener listens for connections and proxies them to a target
15
-type ListenerP2P interface {
15
+type Listener interface {
16
Protocol() protocol.ID
17
ListenAddress() ma.Multiaddr
18
TargetAddress() ma.Multiaddr
19
20
start() error
21
- handleStream(remote net.Stream)
21
+ key() string
22
23
// Close closes the listener. Does not affect child streams
24
Close() error
25
}
26
27
-// ListenerRegistry is a collection of local application proto listeners.
28
-type ListenersP2P struct {
27
+type Listeners struct {
28
sync.RWMutex
29
31
- Listeners map[protocol.ID]ListenerP2P
32
- starting map[protocol.ID]struct{}
30
+ Listeners map[string]Listener
31
+ starting map[string]struct{}
32
}
33
35
-func newListenerP2PRegistry(id peer.ID, host p2phost.Host) *ListenersP2P {
36
- reg := &ListenersP2P{
37
- Listeners: map[protocol.ID]ListenerP2P{},
38
- starting: map[protocol.ID]struct{}{},
34
+func newListenersLocal(id peer.ID) *Listeners {
35
+ return &Listeners{
36
+ Listeners: map[string]Listener{},
37
+ starting: map[string]struct{}{},
38
+ }
39
+}
40
+
41
+func newListenersP2P(id peer.ID, host p2phost.Host) *Listeners {
42
+ reg := &Listeners{
43
+ Listeners: map[string]Listener{},
44
+ starting: map[string]struct{}{},
45
}
46
47
addr, err := ma.NewMultiaddr(maPrefix + id.Pretty())
@@ -60,7 +66,7 @@ func newListenerP2PRegistry(id peer.ID, host p2phost.Host) *ListenersP2P {
66
67
for _, l := range reg.Listeners {
68
if l.ListenAddress().Equal(addr) && l.Protocol() == stream.Protocol() {
63
- go l.handleStream(stream)
69
+ go l.(*remoteListener).handleStream(stream)
70
return
71
}
72
}
@@ -70,10 +76,10 @@ func newListenerP2PRegistry(id peer.ID, host p2phost.Host) *ListenersP2P {
76
}
77
78
// Register registers listenerInfo into this registry and starts it
73
-func (r *ListenersP2P) Register(l ListenerP2P) error {
79
+func (r *Listeners) Register(l Listener) error {
80
r.Lock()
81
+ k := l.key()
82
76
- k := l.Protocol()
83
if _, ok := r.Listeners[k]; ok {
84
r.Unlock()
85
return errors.New("listener already registered")
@@ -100,7 +106,7 @@ func (r *ListenersP2P) Register(l ListenerP2P) error {
106
}
107
108
// Deregister removes p2p listener from this registry
103
-func (r *ListenersP2P) Deregister(k protocol.ID) (bool, error) {
109
+func (r *Listeners) Deregister(k string) (bool, error) {
110
r.Lock()
111
defer r.Unlock()
112
p2p/local.go
+5
-1
@@ -27,7 +27,7 @@ type localListener struct {
27
}
28
29
// ForwardLocal creates new P2P stream to a remote listener
30
-func (p2p *P2P) ForwardLocal(ctx context.Context, peer peer.ID, proto protocol.ID, bindAddr ma.Multiaddr) (ListenerLocal, error) {
30
+func (p2p *P2P) ForwardLocal(ctx context.Context, peer peer.ID, proto protocol.ID, bindAddr ma.Multiaddr) (Listener, error) {
31
listener := &localListener{
32
ctx: ctx,
33
@@ -130,3 +130,7 @@ func (l *localListener) TargetAddress() ma.Multiaddr {
130
}
131
return addr
132
}
133
+
134
+func (l *localListener) key() string {
135
+ return l.ListenAddress().String()
136
+}
p2p/local_listener.go
deleted
-83
@@ -1,83 +0,0 @@
1
-package p2p
2
-
3
-import (
4
- "errors"
5
- "sync"
6
-
7
- peer "gx/ipfs/QmQsErDt8Qgw1XrsXf2BpEzDgGWtB1YLsTAARBup5b6B9W/go-libp2p-peer"
8
- ma "gx/ipfs/QmYmsdtJ3HsodkePE3eU3TsCaP2YvPZJ4LoXnNkDE5Tpt7/go-multiaddr"
9
- "gx/ipfs/QmZNkThpqfVXs9GNbexPrfBbXSLNYeKrE7jwFM2oqHbyqN/go-libp2p-protocol"
10
-)
11
-
12
-// ListenerLocal listens for connections and proxies them to a target
13
-type ListenerLocal interface {
14
- Protocol() protocol.ID
15
- ListenAddress() ma.Multiaddr
16
- TargetAddress() ma.Multiaddr
17
-
18
- start() error
19
-
20
- // Close closes the listener. Does not affect child streams
21
- Close() error
22
-}
23
-
24
-// ListenersLocal is a collection of local application proto listeners.
25
-type ListenersLocal struct {
26
- sync.RWMutex
27
-
28
- Listeners map[string]ListenerLocal
29
- starting map[string]struct{}
30
-}
31
-
32
-func newListenerRegistry(id peer.ID) *ListenersLocal {
33
- reg := &ListenersLocal{
34
- Listeners: map[string]ListenerLocal{},
35
- starting: map[string]struct{}{},
36
- }
37
-
38
- return reg
39
-}
40
-
41
-// Register registers listenerInfo into this registry and starts it
42
-func (r *ListenersLocal) Register(l ListenerLocal) error {
43
- r.Lock()
44
- k := l.ListenAddress().String()
45
-
46
- if _, ok := r.Listeners[k]; ok {
47
- r.Unlock()
48
- return errors.New("listener already registered")
49
- }
50
-
51
- r.Listeners[k] = l
52
- r.starting[k] = struct{}{}
53
-
54
- r.Unlock()
55
-
56
- err := l.start()
57
-
58
- r.Lock()
59
- defer r.Unlock()
60
-
61
- delete(r.starting, k)
62
-
63
- if err != nil {
64
- delete(r.Listeners, k)
65
- return err
66
- }
67
-
68
- return nil
69
-}
70
-
71
-// Deregister removes p2p listener from this registry
72
-func (r *ListenersLocal) Deregister(k string) (bool, error) {
73
- r.Lock()
74
- defer r.Unlock()
75
-
76
- if _, ok := r.starting[k]; ok {
77
- return false, errors.New("listener didn't start yet")
78
- }
79
-
80
- _, ok := r.Listeners[k]
81
- delete(r.Listeners, k)
82
- return ok, nil
83
-}
p2p/p2p.go
+5
-5
@@ -9,10 +9,10 @@ import (
9
10
var log = logging.Logger("p2p-mount")
11
12
-// P2P structure holds information on currently running streams/listeners
12
+// P2P structure holds information on currently running streams/Listeners
13
type P2P struct {
14
- ListenersLocal *ListenersLocal
15
- ListenersP2P *ListenersP2P
14
+ ListenersLocal *Listeners
15
+ ListenersP2P *Listeners
16
Streams *StreamRegistry
17
18
identity peer.ID
@@ -27,8 +27,8 @@ func NewP2P(identity peer.ID, peerHost p2phost.Host, peerstore pstore.Peerstore)
27
peerHost: peerHost,
28
peerstore: peerstore,
29
30
- ListenersLocal: newListenerRegistry(identity),
31
- ListenersP2P: newListenerP2PRegistry(identity, peerHost),
30
+ ListenersLocal: newListenersLocal(identity),
31
+ ListenersP2P: newListenersP2P(identity, peerHost),
32
33
Streams: &StreamRegistry{
34
Streams: map[uint64]*Stream{},
p2p/remote.go
+6
-5
@@ -23,7 +23,7 @@ type remoteListener struct {
23
}
24
25
// ForwardRemote creates new p2p listener
26
-func (p2p *P2P) ForwardRemote(ctx context.Context, proto protocol.ID, addr ma.Multiaddr) (ListenerP2P, error) {
26
+func (p2p *P2P) ForwardRemote(ctx context.Context, proto protocol.ID, addr ma.Multiaddr) (Listener, error) {
27
listener := &remoteListener{
28
p2p: p2p,
29
@@ -91,12 +91,13 @@ func (l *remoteListener) TargetAddress() ma.Multiaddr {
91
}
92
93
func (l *remoteListener) Close() error {
94
- ok, err := l.p2p.ListenersP2P.Deregister(l.proto)
94
+ _, err := l.p2p.ListenersP2P.Deregister(string(l.proto))
95
if err != nil {
96
return err
97
}
98
- if ok {
99
- l.p2p.peerHost.RemoveStreamHandler(l.proto)
100
- }
98
return nil
99
}
100
+
101
+func (l *remoteListener) key() string {
102
+ return string(l.proto)
103
+}