p2p: simplify listener startup
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Sep 6, 2018 at 11:42 UTC
f5cb640d252f6e432e1843e870cfced45a6e648f
5 files changed
+16
-51
p2p/listener.go
+5
-31
@@ -5,7 +5,6 @@ import (
5
"sync"
6
7
net "gx/ipfs/QmQSbtGXCyNrj34LWL8EgXyNNYDZ8r3SwQcpW5pPxVhLnM/go-libp2p-net"
8
- peer "gx/ipfs/QmQsErDt8Qgw1XrsXf2BpEzDgGWtB1YLsTAARBup5b6B9W/go-libp2p-peer"
8
ma "gx/ipfs/QmYmsdtJ3HsodkePE3eU3TsCaP2YvPZJ4LoXnNkDE5Tpt7/go-multiaddr"
9
"gx/ipfs/QmZNkThpqfVXs9GNbexPrfBbXSLNYeKrE7jwFM2oqHbyqN/go-libp2p-protocol"
10
p2phost "gx/ipfs/QmfH9FKYv3Jp1xiyL8sPchGBUBg6JA6XviwajAo3qgnT3B/go-libp2p-host"
@@ -17,7 +16,6 @@ type Listener interface {
16
ListenAddress() ma.Multiaddr
17
TargetAddress() ma.Multiaddr
18
20
- start() error
19
key() string
20
21
// Close closes the listener. Does not affect child streams
@@ -30,20 +28,17 @@ type Listeners struct {
28
sync.RWMutex
29
30
Listeners map[string]Listener
33
- starting map[string]struct{}
31
}
32
36
-func newListenersLocal(id peer.ID) *Listeners {
33
+func newListenersLocal() *Listeners {
34
return &Listeners{
35
Listeners: map[string]Listener{},
39
- starting: map[string]struct{}{},
36
}
37
}
38
43
-func newListenersP2P(id peer.ID, host p2phost.Host) *Listeners {
39
+func newListenersP2P(host p2phost.Host) *Listeners {
40
reg := &Listeners{
41
Listeners: map[string]Listener{},
46
- starting: map[string]struct{}{},
42
}
43
44
host.SetStreamHandlerMatch("/x/", func(p string) bool {
@@ -67,31 +62,14 @@ func newListenersP2P(id peer.ID, host p2phost.Host) *Listeners {
62
63
// Register registers listenerInfo into this registry and starts it
64
func (r *Listeners) Register(l Listener) error {
70
- r.Lock()
71
- k := l.key()
72
-
73
- if _, ok := r.Listeners[k]; ok {
74
- r.Unlock()
75
- return errors.New("listener already registered")
76
- }
77
-
78
- r.Listeners[k] = l
79
- r.starting[k] = struct{}{}
80
-
81
- r.Unlock()
82
-
83
- err := l.start()
84
-
65
r.Lock()
66
defer r.Unlock()
67
88
- delete(r.starting, k)
89
-
90
- if err != nil {
91
- delete(r.Listeners, k)
92
- return err
68
+ if _, ok := r.Listeners[l.key()]; ok {
69
+ return errors.New("listener already registered")
70
}
71
72
+ r.Listeners[l.key()] = l
73
return nil
74
}
75
@@ -100,10 +78,6 @@ func (r *Listeners) Deregister(k string) (bool, error) {
78
r.Lock()
79
defer r.Unlock()
80
103
- if _, ok := r.starting[k]; ok {
104
- return false, errors.New("listener didn't start yet")
105
- }
106
-
81
_, ok := r.Listeners[k]
82
delete(r.Listeners, k)
83
return ok, nil
p2p/local.go
+7
-13
@@ -17,7 +17,6 @@ type localListener struct {
17
ctx context.Context
18
19
p2p *P2P
20
- id peer.ID
20
21
proto protocol.ID
22
laddr ma.Multiaddr
@@ -32,13 +31,19 @@ func (p2p *P2P) ForwardLocal(ctx context.Context, peer peer.ID, proto protocol.I
31
ctx: ctx,
32
33
p2p: p2p,
35
- id: p2p.identity,
34
35
proto: proto,
36
laddr: bindAddr,
37
peer: peer,
38
}
39
40
+ maListener, err := manet.Listen(listener.laddr)
41
+ if err != nil {
42
+ return nil, err
43
+ }
44
+
45
+ listener.listener = maListener
46
+
47
if err := p2p.ListenersLocal.Register(listener); err != nil {
48
return nil, err
49
}
@@ -91,17 +96,6 @@ func (l *localListener) setupStream(local manet.Conn) {
96
}
97
98
l.p2p.Streams.Register(stream)
94
- stream.startStreaming()
95
-}
96
-
97
-func (l *localListener) start() error {
98
- maListener, err := manet.Listen(l.laddr)
99
- if err != nil {
100
- return err
101
- }
102
-
103
- l.listener = maListener
104
- return nil
99
}
100
101
func (l *localListener) Close() error {
p2p/p2p.go
+2
-2
@@ -27,8 +27,8 @@ func NewP2P(identity peer.ID, peerHost p2phost.Host, peerstore pstore.Peerstore)
27
peerHost: peerHost,
28
peerstore: peerstore,
29
30
- ListenersLocal: newListenersLocal(identity),
31
- ListenersP2P: newListenersP2P(identity, peerHost),
30
+ ListenersLocal: newListenersLocal(),
31
+ ListenersP2P: newListenersP2P(peerHost),
32
33
Streams: &StreamRegistry{
34
Streams: map[uint64]*Stream{},
p2p/remote.go
-5
@@ -38,10 +38,6 @@ func (p2p *P2P) ForwardRemote(ctx context.Context, proto protocol.ID, addr ma.Mu
38
return listener, nil
39
}
40
41
-func (l *remoteListener) start() error {
42
- return nil
43
-}
44
-
41
func (l *remoteListener) handleStream(remote net.Stream) {
42
local, err := manet.Dial(l.addr)
43
if err != nil {
@@ -71,7 +67,6 @@ func (l *remoteListener) handleStream(remote net.Stream) {
67
}
68
69
l.p2p.Streams.Register(stream)
74
- stream.startStreaming()
70
}
71
72
func (l *remoteListener) Protocol() protocol.ID {
p2p/stream.go
+2
@@ -84,6 +84,8 @@ func (r *StreamRegistry) Register(streamInfo *Stream) {
84
streamInfo.id = r.nextID
85
r.Streams[r.nextID] = streamInfo
86
r.nextID++
87
+
88
+ streamInfo.startStreaming()
89
}
90
91
// Deregister deregisters stream from the registry