p2p: refactor local/remote
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
May 23, 2018 at 16:10 UTC
2487c99a7f8c36c4a783c3632df92be3bdc5eafd
4 files changed
+47
-36
core/commands/p2p.go
+16
-14
@@ -21,8 +21,9 @@ import (
21
22
// P2PListenerInfoOutput is output type of ls command
23
type P2PListenerInfoOutput struct {
24
- Protocol string
25
- Address string
24
+ Protocol string
25
+ ListenAddress string
26
+ TargetAddress string
27
}
28
29
// P2PStreamInfoOutput is output type of streams command
@@ -130,7 +131,7 @@ func forwardRemote(ctx context.Context, p *p2p.P2P, proto string, target string)
131
}
132
133
// TODO: return some info
133
- _, err = p.NewListener(ctx, proto, addr)
134
+ _, err = p.ForwardRemote(ctx, proto, addr)
135
return err
136
}
137
@@ -151,7 +152,7 @@ func forwardLocal(ctx context.Context, p *p2p.P2P, ps pstore.Peerstore, proto st
152
}
153
154
// TODO: return some info
154
- _, err = p.Dial(ctx, peer, proto, bindAddr)
155
+ _, err = p.ForwardLocal(ctx, peer, proto, bindAddr)
156
return err
157
}
158
@@ -206,8 +207,9 @@ var p2pListenerLsCmd = &cmds.Command{
207
208
for _, listener := range n.P2P.Listeners.Listeners {
209
output.Listeners = append(output.Listeners, P2PListenerInfoOutput{
209
- Protocol: listener.Protocol(),
210
- Address: listener.Address(),
210
+ Protocol: listener.Protocol(),
211
+ ListenAddress: listener.ListenAddress(),
212
+ TargetAddress: listener.TargetAddress(),
213
})
214
}
215
@@ -227,10 +229,10 @@ var p2pListenerLsCmd = &cmds.Command{
229
w := tabwriter.NewWriter(buf, 1, 2, 1, ' ', 0)
230
for _, listener := range list.Listeners {
231
if headers {
230
- fmt.Fprintln(w, "Address\tProtocol")
232
+ fmt.Fprintln(w, "Protocol\tListen Address\tTarget Address")
233
}
234
233
- fmt.Fprintf(w, "%s\t%s\n", listener.Address, listener.Protocol)
235
+ fmt.Fprintf(w, "%s\t%s\t%s\n", listener.Protocol, listener.ListenAddress, listener.TargetAddress)
236
}
237
w.Flush()
238
@@ -330,7 +332,7 @@ Note that the connections originate from the ipfs daemon process.
332
return
333
}
334
333
- _, err = n.P2P.NewListener(n.Context(), proto, addr)
335
+ _, err = n.P2P.ForwardRemote(n.Context(), proto, addr)
336
if err != nil {
337
res.SetError(err, cmdkit.ErrNormal)
338
return
@@ -338,8 +340,8 @@ Note that the connections originate from the ipfs daemon process.
340
341
// Successful response.
342
res.SetOutput(&P2PListenerInfoOutput{
341
- Protocol: proto,
342
- Address: addr.String(),
343
+ Protocol: proto,
344
+ TargetAddress: addr.String(),
345
})
346
},
347
}
@@ -389,15 +391,15 @@ can transparently connect to a p2p service.
391
}
392
}
393
392
- listenerInfo, err := n.P2P.Dial(n.Context(), peer, proto, bindAddr)
394
+ listenerInfo, err := n.P2P.ForwardLocal(n.Context(), peer, proto, bindAddr)
395
if err != nil {
396
res.SetError(err, cmdkit.ErrNormal)
397
return
398
}
399
400
output := P2PListenerInfoOutput{
399
- Protocol: listenerInfo.Protocol(),
400
- Address: listenerInfo.Address(),
401
+ Protocol: listenerInfo.Protocol(),
402
+ ListenAddress: listenerInfo.ListenAddress(),
403
}
404
405
res.SetOutput(&output)
p2p/listener.go
+2
-1
@@ -2,7 +2,8 @@ package p2p
2
3
type Listener interface {
4
Protocol() string
5
- Address() string
5
+ ListenAddress() string
6
+ TargetAddress() string
7
8
// Close closes the listener. Does not affect child streams
9
Close() error
p2p/local.go
renamed
+15
-11
@@ -12,8 +12,8 @@ import (
12
peer "gx/ipfs/QmcJukH2sAFjY3HdBKq35WDzWoL3UUu2gt9wdfqZTUyM74/go-libp2p-peer"
13
)
14
15
-// outboundListener accepts libp2p streams and proxies them to a manet host
16
-type outboundListener struct {
15
+// localListener manet streams and proxies them to libp2p services
16
+type localListener struct {
17
ctx context.Context
18
19
p2p *P2P
@@ -25,14 +25,14 @@ type outboundListener struct {
25
listener manet.Listener
26
}
27
28
-// Dial creates new P2P stream to a remote listener
29
-func (p2p *P2P) Dial(ctx context.Context, peer peer.ID, proto string, bindAddr ma.Multiaddr) (Listener, error) {
28
+// ForwardLocal creates new P2P stream to a remote listener
29
+func (p2p *P2P) ForwardLocal(ctx context.Context, peer peer.ID, proto string, bindAddr ma.Multiaddr) (Listener, error) {
30
maListener, err := manet.Listen(bindAddr)
31
if err != nil {
32
return nil, err
33
}
34
35
- listener := &outboundListener{
35
+ listener := &localListener{
36
ctx: ctx,
37
38
p2p: p2p,
@@ -50,7 +50,7 @@ func (p2p *P2P) Dial(ctx context.Context, peer peer.ID, proto string, bindAddr m
50
return listener, nil
51
}
52
53
-func (l *outboundListener) dial() (net.Stream, error) {
53
+func (l *localListener) dial() (net.Stream, error) {
54
ctx, cancel := context.WithTimeout(l.ctx, time.Second*30) //TODO: configurable?
55
defer cancel()
56
@@ -62,7 +62,7 @@ func (l *outboundListener) dial() (net.Stream, error) {
62
return l.p2p.peerHost.NewStream(l.ctx, l.peer, protocol.ID(l.proto))
63
}
64
65
-func (l *outboundListener) acceptConns() {
65
+func (l *localListener) acceptConns() {
66
for {
67
local, err := l.listener.Accept()
68
if err != nil {
@@ -95,16 +95,20 @@ func (l *outboundListener) acceptConns() {
95
}
96
}
97
98
-func (l *outboundListener) Close() error {
98
+func (l *localListener) Close() error {
99
l.listener.Close()
100
l.p2p.Listeners.Deregister(l.proto)
101
return nil
102
}
103
104
-func (l *outboundListener) Protocol() string {
104
+func (l *localListener) Protocol() string {
105
return l.proto
106
}
107
108
-func (l *outboundListener) Address() string {
109
- return "/ipfs/" + l.peer.String()
108
+func (l *localListener) ListenAddress() string {
109
+ return l.listener.Multiaddr().String()
110
+}
111
+
112
+func (l *localListener) TargetAddress() string {
113
+ return "/ipfs/" + l.peer.Pretty()
114
}
p2p/remote.go
renamed
+14
-10
@@ -9,8 +9,8 @@ import (
9
manet "gx/ipfs/QmcGXGdw9BWDysPJQHxJinjGHha3eEg4vzFETre4woNwcX/go-multiaddr-net"
10
)
11
12
-// inboundListener accepts libp2p streams and proxies them to a manet host
13
-type inboundListener struct {
12
+// remoteListener accepts libp2p streams and proxies them to a manet host
13
+type remoteListener struct {
14
p2p *P2P
15
16
// Application proto identifier.
@@ -20,15 +20,17 @@ type inboundListener struct {
20
addr ma.Multiaddr
21
}
22
23
-// NewListener creates new p2p listener
24
-func (p2p *P2P) NewListener(ctx context.Context, proto string, addr ma.Multiaddr) (Listener, error) {
25
- listenerInfo := &inboundListener{
23
+// ForwardRemote creates new p2p listener
24
+func (p2p *P2P) ForwardRemote(ctx context.Context, proto string, addr ma.Multiaddr) (Listener, error) {
25
+ listenerInfo := &remoteListener{
26
p2p: p2p,
27
28
proto: proto,
29
addr: addr,
30
}
31
32
+ p2p.Listeners.Register(listenerInfo)
33
+
34
p2p.peerHost.SetStreamHandler(protocol.ID(proto), func(remote net.Stream) {
35
local, err := manet.Dial(addr)
36
if err != nil {
@@ -55,20 +57,22 @@ func (p2p *P2P) NewListener(ctx context.Context, proto string, addr ma.Multiaddr
57
stream.startStreaming()
58
})
59
58
- p2p.Listeners.Register(listenerInfo)
59
-
60
return listenerInfo, nil
61
}
62
63
-func (l *inboundListener) Protocol() string {
63
+func (l *remoteListener) Protocol() string {
64
return l.proto
65
}
66
67
-func (l *inboundListener) Address() string {
67
+func (l *remoteListener) ListenAddress() string {
68
+ return "/ipfs"
69
+}
70
+
71
+func (l *remoteListener) TargetAddress() string {
72
return l.addr.String()
73
}
74
71
-func (l *inboundListener) Close() error {
75
+func (l *remoteListener) Close() error {
76
l.p2p.peerHost.RemoveStreamHandler(protocol.ID(l.proto))
77
l.p2p.Listeners.Deregister(l.proto)
78
return nil