coosld/kubo
1
0
Fork 0
mirror of https://github.com/ipfs/kubo.git synced 2026-02-21 10:27:46 +08:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

p2p: cleanup after listener iface split

Browse Source 
License: MIT
Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
This commit is contained in:
Łukasz Magiera 2018-09-05 23:13:01 +02:00
parent dd48b8237a
commit bba2d05ca9
4 changed files with 9 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
p2p/local_listener.go
View File

@ -4,11 +4,9 @@ import (
"errors"
"sync"
net "gx/ipfs/QmQSbtGXCyNrj34LWL8EgXyNNYDZ8r3SwQcpW5pPxVhLnM/go-libp2p-net"
peer "gx/ipfs/QmQsErDt8Qgw1XrsXf2BpEzDgGWtB1YLsTAARBup5b6B9W/go-libp2p-peer"
ma "gx/ipfs/QmYmsdtJ3HsodkePE3eU3TsCaP2YvPZJ4LoXnNkDE5Tpt7/go-multiaddr"
"gx/ipfs/QmZNkThpqfVXs9GNbexPrfBbXSLNYeKrE7jwFM2oqHbyqN/go-libp2p-protocol"
p2phost "gx/ipfs/QmfH9FKYv3Jp1xiyL8sPchGBUBg6JA6XviwajAo3qgnT3B/go-libp2p-host"
)
// ListenerLocal listens for connections and proxies them to a target
@ -31,40 +29,12 @@ type ListenersLocal struct {
starting map[string]struct{}
}
func newListenerRegistry(id peer.ID, host p2phost.Host) *ListenersLocal {
func newListenerRegistry(id peer.ID) *ListenersLocal {
reg := &ListenersLocal{
Listeners: map[string]ListenerLocal{},
starting: map[string]struct{}{},
}
addr, err := ma.NewMultiaddr(maPrefix + id.Pretty())
if err != nil {
panic(err)
}
host.SetStreamHandlerMatch("/x/", func(p string) bool {
reg.RLock()
defer reg.RUnlock()
for _, l := range reg.Listeners {
if l.ListenAddress().Equal(addr) && string(l.Protocol()) == p {
return true
}
}
return false
}, func(stream net.Stream) {
reg.RLock()
defer reg.RUnlock()
for _, l := range reg.Listeners {
if l.ListenAddress().Equal(addr) && l.Protocol() == stream.Protocol() {
go l.(*remoteListener).handleStream(stream)
return
}
}
})
return reg
}

2
p2p/p2p.go
View File

@ -27,7 +27,7 @@ func NewP2P(identity peer.ID, peerHost p2phost.Host, peerstore pstore.Peerstore)
peerHost: peerHost,
peerstore: peerstore,
ListenersLocal: newListenerRegistry(identity, peerHost),
ListenersLocal: newListenerRegistry(identity),
ListenersP2P: newListenerP2PRegistry(identity, peerHost),
Streams: &StreamRegistry{

11
p2p/p2p_listener.go
View File

@ -12,12 +12,13 @@ import (
)
// Listener listens for connections and proxies them to a target
type P2PListener interface {
type ListenerP2P interface {
Protocol() protocol.ID
ListenAddress() ma.Multiaddr
TargetAddress() ma.Multiaddr
start() error
handleStream(remote net.Stream)
// Close closes the listener. Does not affect child streams
Close() error
@ -27,13 +28,13 @@ type P2PListener interface {
type ListenersP2P struct {
sync.RWMutex
Listeners map[protocol.ID]ListenerLocal
Listeners map[protocol.ID]ListenerP2P
starting map[protocol.ID]struct{}
}
func newListenerP2PRegistry(id peer.ID, host p2phost.Host) *ListenersP2P {
reg := &ListenersP2P{
Listeners: map[protocol.ID]ListenerLocal{},
Listeners: map[protocol.ID]ListenerP2P{},
starting: map[protocol.ID]struct{}{},
}
@ -59,7 +60,7 @@ func newListenerP2PRegistry(id peer.ID, host p2phost.Host) *ListenersP2P {
for _, l := range reg.Listeners {
if l.ListenAddress().Equal(addr) && l.Protocol() == stream.Protocol() {
go l.(*remoteListener).handleStream(stream)
go l.handleStream(stream)
return
}
}
@ -69,7 +70,7 @@ func newListenerP2PRegistry(id peer.ID, host p2phost.Host) *ListenersP2P {
}
// Register registers listenerInfo into this registry and starts it
func (r *ListenersP2P) Register(l ListenerLocal) error {
func (r *ListenersP2P) Register(l ListenerP2P) error {
r.Lock()
k := l.Protocol()

2
p2p/remote.go
View File

@ -23,7 +23,7 @@ type remoteListener struct {
}
// ForwardRemote creates new p2p listener
func (p2p *P2P) ForwardRemote(ctx context.Context, proto protocol.ID, addr ma.Multiaddr) (P2PListener, error) {
func (p2p *P2P) ForwardRemote(ctx context.Context, proto protocol.ID, addr ma.Multiaddr) (ListenerP2P, error) {
listener := &remoteListener{
p2p: p2p,