PTP API: Make code more object oriented, use less node
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Jun 3, 2017 at 23:28 UTC
104268f1b6e2ff6712ed14186d66d7203710f197
11 files changed
+328
-361
core/commands/ptp.go
+25
-26
@@ -10,13 +10,12 @@ import (
10
11
cmds "github.com/ipfs/go-ipfs/commands"
12
core "github.com/ipfs/go-ipfs/core"
13
- ptpnet "github.com/ipfs/go-ipfs/ptp/net"
13
14
ma "gx/ipfs/QmcyqRMCAXVtYPS4DiBrA7sezL9rRGfW8Ctx7cywL4TXJj/go-multiaddr"
15
)
16
18
-// PTPAppInfoOutput is output type of ls command
19
-type PTPAppInfoOutput struct {
17
+// PTPListenerInfoOutput is output type of ls command
18
+type PTPListenerInfoOutput struct {
19
Protocol string
20
Address string
21
}
@@ -33,7 +32,7 @@ type PTPStreamInfoOutput struct {
32
33
// PTPLsOutput is output type of ls command
34
type PTPLsOutput struct {
36
- Apps []PTPAppInfoOutput
35
+ Listeners []PTPListenerInfoOutput
36
}
37
38
// PTPStreamsOutput is output type of streams command
@@ -87,10 +86,10 @@ var ptpLsCmd = &cmds.Command{
86
87
output := &PTPLsOutput{}
88
90
- for _, app := range n.PTP.Apps.Apps {
91
- output.Apps = append(output.Apps, PTPAppInfoOutput{
92
- Protocol: app.Protocol,
93
- Address: app.Address.String(),
89
+ for _, listener := range n.PTP.Listeners.Listeners {
90
+ output.Listeners = append(output.Listeners, PTPListenerInfoOutput{
91
+ Protocol: listener.Protocol,
92
+ Address: listener.Address.String(),
93
})
94
}
95
@@ -103,12 +102,12 @@ var ptpLsCmd = &cmds.Command{
102
list, _ := res.Output().(*PTPLsOutput)
103
buf := new(bytes.Buffer)
104
w := tabwriter.NewWriter(buf, 1, 2, 1, ' ', 0)
106
- for _, app := range list.Apps {
105
+ for _, listener := range list.Listeners {
106
if headers {
107
fmt.Fprintln(w, "Address\tProtocol")
108
}
109
111
- fmt.Fprintf(w, "%s\t%s\n", app.Address, app.Protocol)
110
+ fmt.Fprintf(w, "%s\t%s\n", listener.Address, listener.Protocol)
111
}
112
w.Flush()
113
@@ -183,9 +182,9 @@ var ptpStreamsCmd = &cmds.Command{
182
183
var ptpListenCmd = &cmds.Command{
184
Helptext: cmds.HelpText{
186
- Tagline: "Create application protocol listener and proxy to network multiaddr.",
185
+ Tagline: "Forward p2p connections to a network multiaddr.",
186
ShortDescription: `
188
-Register a p2p connection handler and proxies the connections to a specified address.
187
+Register a p2p connection handler and forward the connections to a specified address.
188
189
Note that the connections originate from the ipfs daemon process.
190
`,
@@ -212,8 +211,8 @@ Note that the connections originate from the ipfs daemon process.
211
return
212
}
213
215
- proto := "/app/" + req.Arguments()[0]
216
- if ptpnet.CheckProtoExists(n, proto) {
214
+ proto := "/ptp/" + req.Arguments()[0]
215
+ if n.PTP.CheckProtoExists(proto) {
216
res.SetError(errors.New("protocol handler already registered"), cmds.ErrNormal)
217
return
218
}
@@ -224,14 +223,14 @@ Note that the connections originate from the ipfs daemon process.
223
return
224
}
225
227
- _, err = ptpnet.NewListener(n, proto, addr)
226
+ _, err = n.PTP.NewListener(n.Context(), proto, addr)
227
if err != nil {
228
res.SetError(err, cmds.ErrNormal)
229
return
230
}
231
232
// Successful response.
234
- res.SetOutput(&PTPAppInfoOutput{
233
+ res.SetOutput(&PTPListenerInfoOutput{
234
Protocol: proto,
235
Address: addr.String(),
236
})
@@ -253,7 +252,7 @@ transparently connect to a p2p service.
252
Arguments: []cmds.Argument{
253
cmds.StringArg("Peer", true, false, "Remote peer to connect to"),
254
cmds.StringArg("Protocol", true, false, "Protocol identifier."),
256
- cmds.StringArg("BindAddress", false, false, "Address to listen for application/s (default: /ip4/127.0.0.1/tcp/0)."),
255
+ cmds.StringArg("BindAddress", false, false, "Address to listen for connection/s (default: /ip4/127.0.0.1/tcp/0)."),
256
},
257
Run: func(req cmds.Request, res cmds.Response) {
258
n, err := req.InvocContext().GetNode()
@@ -279,7 +278,7 @@ transparently connect to a p2p service.
278
return
279
}
280
282
- proto := "/app/" + req.Arguments()[1]
281
+ proto := "/ptp/" + req.Arguments()[1]
282
283
bindAddr, _ := ma.NewMultiaddr("/ip4/127.0.0.1/tcp/0")
284
if len(req.Arguments()) == 3 {
@@ -290,15 +289,15 @@ transparently connect to a p2p service.
289
}
290
}
291
293
- app, err := ptpnet.Dial(n, addr, peer, proto, bindAddr)
292
+ listenerInfo, err := n.PTP.Dial(n.Context(), addr, peer, proto, bindAddr)
293
if err != nil {
294
res.SetError(err, cmds.ErrNormal)
295
return
296
}
297
299
- output := PTPAppInfoOutput{
300
- Protocol: app.Protocol,
301
- Address: app.Address.String(),
298
+ output := PTPListenerInfoOutput{
299
+ Protocol: listenerInfo.Protocol,
300
+ Address: listenerInfo.Address.String(),
301
}
302
303
res.SetOutput(&output)
@@ -348,7 +347,7 @@ var ptpCloseCmd = &cmds.Command{
347
348
handlerID, err = strconv.ParseUint(req.Arguments()[0], 10, 64)
349
if err != nil {
351
- proto = "/app/" + req.Arguments()[0]
350
+ proto = "/ptp/" + req.Arguments()[0]
351
} else {
352
useHandlerID = true
353
}
@@ -367,11 +366,11 @@ var ptpCloseCmd = &cmds.Command{
366
}
367
368
if closeAll || !useHandlerID {
370
- for _, app := range n.PTP.Apps.Apps {
371
- if !closeAll && app.Protocol != proto {
369
+ for _, listener := range n.PTP.Listeners.Listeners {
370
+ if !closeAll && listener.Protocol != proto {
371
continue
372
}
374
- app.Close()
373
+ listener.Close()
374
if !closeAll {
375
break
376
}
core/commands/root.go
+1
-1
@@ -47,7 +47,7 @@ ADVANCED COMMANDS
47
pin Pin objects to local storage
48
repo Manipulate the IPFS repository
49
stats Various operational stats
50
- ptp Libp2p stream mounting
50
+ ptp Libp2p stream mounting
51
filestore Manage the filestore (experimental)
52
53
NETWORK COMMANDS
core/core.go
+2
-2
@@ -23,7 +23,6 @@ import (
23
24
bstore "github.com/ipfs/go-ipfs/blocks/blockstore"
25
bserv "github.com/ipfs/go-ipfs/blockservice"
26
- ptp "github.com/ipfs/go-ipfs/ptp"
26
exchange "github.com/ipfs/go-ipfs/exchange"
27
bitswap "github.com/ipfs/go-ipfs/exchange/bitswap"
28
bsnet "github.com/ipfs/go-ipfs/exchange/bitswap/network"
@@ -36,6 +35,7 @@ import (
35
ipnsrp "github.com/ipfs/go-ipfs/namesys/republisher"
36
path "github.com/ipfs/go-ipfs/path"
37
pin "github.com/ipfs/go-ipfs/pin"
38
+ ptp "github.com/ipfs/go-ipfs/ptp"
39
repo "github.com/ipfs/go-ipfs/repo"
40
config "github.com/ipfs/go-ipfs/repo/config"
41
nilrouting "github.com/ipfs/go-ipfs/routing/none"
@@ -248,7 +248,7 @@ func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption Routin
248
n.Floodsub = floodsub.NewFloodSub(ctx, peerhost)
249
}
250
251
- n.PTP = ptp.NewPTP()
251
+ n.PTP = ptp.NewPTP(n.Identity, n.PeerHost, n.Peerstore)
252
253
// setup local discovery
254
if do != nil {
ptp/net/dial.go
deleted
-82
@@ -1,82 +0,0 @@
1
-package net
2
-
3
-import (
4
- "errors"
5
-
6
- core "github.com/ipfs/go-ipfs/core"
7
- ptp "github.com/ipfs/go-ipfs/ptp"
8
-
9
- net "gx/ipfs/QmRscs8KxrSmSv4iuevHv8JfuUzHBMoqiaHzxfDRiksd6e/go-libp2p-net"
10
- peerstore "gx/ipfs/QmXZSd1qR5BxZkPyuwfT5jpqQFScZccoZvDneXsKzCNHWX/go-libp2p-peerstore"
11
- ma "gx/ipfs/QmcyqRMCAXVtYPS4DiBrA7sezL9rRGfW8Ctx7cywL4TXJj/go-multiaddr"
12
- peer "gx/ipfs/QmdS9KpbDyPrieswibZhkod1oXqRwZJrUPzxCofAMWpFGq/go-libp2p-peer"
13
- manet "gx/ipfs/Qmf1Gq7N45Rpuw7ev47uWgH6dLPtdnvcMRNPkVBwqjLJg2/go-multiaddr-net"
14
-)
15
-
16
-func Dial(n *core.IpfsNode, addr ma.Multiaddr, peer peer.ID, proto string, bindAddr ma.Multiaddr) (*ptp.ListenerInfo, error) {
17
- lnet, _, err := manet.DialArgs(bindAddr)
18
- if err != nil {
19
- return nil, err
20
- }
21
-
22
- app := ptp.ListenerInfo{
23
- Identity: n.Identity,
24
- Protocol: proto,
25
- }
26
-
27
- n.Peerstore.AddAddr(peer, addr, peerstore.TempAddrTTL)
28
-
29
- remote, err := dial(n, peer, proto)
30
- if err != nil {
31
- return nil, err
32
- }
33
-
34
- switch lnet {
35
- case "tcp", "tcp4", "tcp6":
36
- listener, err := manet.Listen(bindAddr)
37
- if err != nil {
38
- if err2 := remote.Close(); err2 != nil {
39
- return nil, err2
40
- }
41
- return nil, err
42
- }
43
-
44
- app.Address = listener.Multiaddr()
45
- app.Closer = listener
46
- app.Running = true
47
-
48
- go doAccept(n, &app, remote, listener)
49
-
50
- default:
51
- return nil, errors.New("unsupported protocol: " + lnet)
52
- }
53
-
54
- return &app, nil
55
-}
56
-
57
-func doAccept(n *core.IpfsNode, app *ptp.ListenerInfo, remote net.Stream, listener manet.Listener) {
58
- defer listener.Close()
59
-
60
- local, err := listener.Accept()
61
- if err != nil {
62
- return
63
- }
64
-
65
- stream := ptp.StreamInfo{
66
- Protocol: app.Protocol,
67
-
68
- LocalPeer: app.Identity,
69
- LocalAddr: app.Address,
70
-
71
- RemotePeer: remote.Conn().RemotePeer(),
72
- RemoteAddr: remote.Conn().RemoteMultiaddr(),
73
-
74
- Local: local,
75
- Remote: remote,
76
-
77
- Registry: &n.PTP.Streams,
78
- }
79
-
80
- n.PTP.Streams.Register(&stream)
81
- startStreaming(&stream)
82
-}
ptp/net/listen.go
deleted
-67
@@ -1,67 +0,0 @@
1
-package net
2
-
3
-import (
4
- "github.com/ipfs/go-ipfs/core"
5
- "github.com/ipfs/go-ipfs/ptp"
6
-
7
- ma "gx/ipfs/QmcyqRMCAXVtYPS4DiBrA7sezL9rRGfW8Ctx7cywL4TXJj/go-multiaddr"
8
- manet "gx/ipfs/Qmf1Gq7N45Rpuw7ev47uWgH6dLPtdnvcMRNPkVBwqjLJg2/go-multiaddr-net"
9
-)
10
-
11
-// NewListener creates new ptp listener
12
-func NewListener(n *core.IpfsNode, proto string, addr ma.Multiaddr) (*ptp.ListenerInfo, error) {
13
- listener, err := Listen(n, proto)
14
- if err != nil {
15
- return nil, err
16
- }
17
-
18
- listenerInfo := ptp.ListenerInfo{
19
- Identity: n.Identity,
20
- Protocol: proto,
21
- Address: addr,
22
- Closer: listener,
23
- Running: true,
24
- Registry: &n.PTP.Listeners,
25
- }
26
-
27
- go acceptStreams(n, &listenerInfo, listener)
28
-
29
- n.PTP.Listeners.Register(&listenerInfo)
30
-
31
- return &listenerInfo, nil
32
-}
33
-
34
-func acceptStreams(n *core.IpfsNode, listenerInfo *ptp.ListenerInfo, listener Listener) {
35
- for listenerInfo.Running {
36
- remote, err := listener.Accept()
37
- if err != nil {
38
- listener.Close()
39
- break
40
- }
41
-
42
- local, err := manet.Dial(listenerInfo.Address)
43
- if err != nil {
44
- remote.Close()
45
- continue
46
- }
47
-
48
- stream := ptp.StreamInfo{
49
- Protocol: listenerInfo.Protocol,
50
-
51
- LocalPeer: listenerInfo.Identity,
52
- LocalAddr: listenerInfo.Address,
53
-
54
- RemotePeer: remote.Conn().RemotePeer(),
55
- RemoteAddr: remote.Conn().RemoteMultiaddr(),
56
-
57
- Local: local,
58
- Remote: remote,
59
-
60
- Registry: &n.PTP.Streams,
61
- }
62
-
63
- n.PTP.Streams.Register(&stream)
64
- startStreaming(&stream)
65
- }
66
- n.PTP.Listeners.Deregister(listenerInfo.Protocol)
67
-}
ptp/net/net.go
deleted
-93
@@ -1,93 +0,0 @@
1
-package net
2
-
3
-import (
4
- "time"
5
-
6
- context "context"
7
- core "github.com/ipfs/go-ipfs/core"
8
-
9
- net "gx/ipfs/QmRscs8KxrSmSv4iuevHv8JfuUzHBMoqiaHzxfDRiksd6e/go-libp2p-net"
10
- pstore "gx/ipfs/QmXZSd1qR5BxZkPyuwfT5jpqQFScZccoZvDneXsKzCNHWX/go-libp2p-peerstore"
11
- pro "gx/ipfs/QmZNkThpqfVXs9GNbexPrfBbXSLNYeKrE7jwFM2oqHbyqN/go-libp2p-protocol"
12
- peer "gx/ipfs/QmdS9KpbDyPrieswibZhkod1oXqRwZJrUPzxCofAMWpFGq/go-libp2p-peer"
13
-)
14
-
15
-// Listener wraps stream handler into a listener
16
-type Listener interface {
17
- Accept() (net.Stream, error)
18
- Close() error
19
-}
20
-
21
-// IpfsListener holds information on a listener
22
-type IpfsListener struct {
23
- node *core.IpfsNode
24
- conCh chan net.Stream
25
- proto pro.ID
26
- ctx context.Context
27
- cancel func()
28
-}
29
-
30
-// Accept waits for a connection from the listener
31
-func (il *IpfsListener) Accept() (net.Stream, error) {
32
- select {
33
- case c := <-il.conCh:
34
- return c, nil
35
- case <-il.ctx.Done():
36
- return nil, il.ctx.Err()
37
- }
38
-}
39
-
40
-// Close closes the listener and removes stream handler
41
-func (il *IpfsListener) Close() error {
42
- il.cancel()
43
- il.node.PeerHost.RemoveStreamHandler(il.proto)
44
- return nil
45
-}
46
-
47
-// Listen creates new IpfsListener
48
-func Listen(nd *core.IpfsNode, protocol string) (*IpfsListener, error) {
49
- ctx, cancel := context.WithCancel(nd.Context())
50
-
51
- list := &IpfsListener{
52
- node: nd,
53
- proto: pro.ID(protocol),
54
- conCh: make(chan net.Stream),
55
- ctx: ctx,
56
- cancel: cancel,
57
- }
58
-
59
- nd.PeerHost.SetStreamHandler(list.proto, func(s net.Stream) {
60
- select {
61
- case list.conCh <- s:
62
- case <-ctx.Done():
63
- s.Close()
64
- }
65
- })
66
-
67
- return list, nil
68
-}
69
-
70
-// Dial dials to a specified node and protocol
71
-func dial(nd *core.IpfsNode, p peer.ID, protocol string) (net.Stream, error) {
72
- ctx, cancel := context.WithTimeout(nd.Context(), time.Second*30)
73
- defer cancel()
74
- err := nd.PeerHost.Connect(ctx, pstore.PeerInfo{ID: p})
75
- if err != nil {
76
- return nil, err
77
- }
78
- return nd.PeerHost.NewStream(nd.Context(), p, pro.ID(protocol))
79
-}
80
-
81
-// CheckProtoExists checks whether a protocol handler is registered to
82
-// mux handler
83
-func CheckProtoExists(n *core.IpfsNode, proto string) bool {
84
- protos := n.PeerHost.Mux().Protocols()
85
-
86
- for _, p := range protos {
87
- if p != proto {
88
- continue
89
- }
90
- return true
91
- }
92
- return false
93
-}
ptp/net/util.go
deleted
-19
@@ -1,19 +0,0 @@
1
-package net
2
-
3
-import (
4
- "io"
5
-
6
- ptp "github.com/ipfs/go-ipfs/ptp"
7
-)
8
-
9
-func startStreaming(stream *ptp.StreamInfo) {
10
- go func() {
11
- io.Copy(stream.Local, stream.Remote)
12
- stream.Close()
13
- }()
14
-
15
- go func() {
16
- io.Copy(stream.Remote, stream.Local)
17
- stream.Close()
18
- }()
19
-}
ptp/ptp.go
+228
-3
@@ -1,12 +1,237 @@
1
package ptp
2
3
-// PTP structure holds information on currently running streams/apps
3
+import (
4
+ "context"
5
+ "errors"
6
+ "time"
7
+
8
+ net "gx/ipfs/QmRscs8KxrSmSv4iuevHv8JfuUzHBMoqiaHzxfDRiksd6e/go-libp2p-net"
9
+ p2phost "gx/ipfs/QmUywuGNZoUKV8B9iyvup9bPkLiMrhTsyVMkeSXW5VxAfC/go-libp2p-host"
10
+ pstore "gx/ipfs/QmXZSd1qR5BxZkPyuwfT5jpqQFScZccoZvDneXsKzCNHWX/go-libp2p-peerstore"
11
+ pro "gx/ipfs/QmZNkThpqfVXs9GNbexPrfBbXSLNYeKrE7jwFM2oqHbyqN/go-libp2p-protocol"
12
+ ma "gx/ipfs/QmcyqRMCAXVtYPS4DiBrA7sezL9rRGfW8Ctx7cywL4TXJj/go-multiaddr"
13
+ peer "gx/ipfs/QmdS9KpbDyPrieswibZhkod1oXqRwZJrUPzxCofAMWpFGq/go-libp2p-peer"
14
+ manet "gx/ipfs/Qmf1Gq7N45Rpuw7ev47uWgH6dLPtdnvcMRNPkVBwqjLJg2/go-multiaddr-net"
15
+)
16
+
17
+// PTP structure holds information on currently running streams/listeners
18
type PTP struct {
19
Listeners ListenerRegistry
20
Streams StreamRegistry
21
+
22
+ identity peer.ID
23
+ peerHost p2phost.Host
24
+ peerstore pstore.Peerstore
25
}
26
27
// NewPTP creates new PTP struct
10
-func NewPTP() *PTP {
11
- return &PTP{}
28
+func NewPTP(identity peer.ID, peerHost p2phost.Host, peerstore pstore.Peerstore) *PTP {
29
+ return &PTP{
30
+ identity: identity,
31
+ peerHost: peerHost,
32
+ peerstore: peerstore,
33
+ }
34
+}
35
+
36
+func (ptp *PTP) newStreamTo(ctx2 context.Context, p peer.ID, protocol string) (net.Stream, error) {
37
+ ctx, cancel := context.WithTimeout(ctx2, time.Second*30) //TODO: configurable?
38
+ defer cancel()
39
+ err := ptp.peerHost.Connect(ctx, pstore.PeerInfo{ID: p})
40
+ if err != nil {
41
+ return nil, err
42
+ }
43
+ return ptp.peerHost.NewStream(ctx2, p, pro.ID(protocol))
44
+}
45
+
46
+func (ptp *PTP) Dial(ctx context.Context, addr ma.Multiaddr, peer peer.ID, proto string, bindAddr ma.Multiaddr) (*ListenerInfo, error) {
47
+ lnet, _, err := manet.DialArgs(bindAddr)
48
+ if err != nil {
49
+ return nil, err
50
+ }
51
+
52
+ listenerInfo := ListenerInfo{
53
+ Identity: ptp.identity,
54
+ Protocol: proto,
55
+ }
56
+
57
+ remote, err := ptp.newStreamTo(ctx, peer, proto)
58
+ if err != nil {
59
+ return nil, err
60
+ }
61
+
62
+ switch lnet {
63
+ case "tcp", "tcp4", "tcp6":
64
+ listener, err := manet.Listen(bindAddr)
65
+ if err != nil {
66
+ if err2 := remote.Close(); err2 != nil {
67
+ return nil, err2
68
+ }
69
+ return nil, err
70
+ }
71
+
72
+ listenerInfo.Address = listener.Multiaddr()
73
+ listenerInfo.Closer = listener
74
+ listenerInfo.Running = true
75
+
76
+ go ptp.doAccept(&listenerInfo, remote, listener)
77
+
78
+ default:
79
+ return nil, errors.New("unsupported protocol: " + lnet)
80
+ }
81
+
82
+ return &listenerInfo, nil
83
+}
84
+
85
+func (ptp *PTP) doAccept(listenerInfo *ListenerInfo, remote net.Stream, listener manet.Listener) {
86
+ defer listener.Close()
87
+
88
+ local, err := listener.Accept()
89
+ if err != nil {
90
+ return
91
+ }
92
+
93
+ stream := StreamInfo{
94
+ Protocol: listenerInfo.Protocol,
95
+
96
+ LocalPeer: listenerInfo.Identity,
97
+ LocalAddr: listenerInfo.Address,
98
+
99
+ RemotePeer: remote.Conn().RemotePeer(),
100
+ RemoteAddr: remote.Conn().RemoteMultiaddr(),
101
+
102
+ Local: local,
103
+ Remote: remote,
104
+
105
+ Registry: &ptp.Streams,
106
+ }
107
+
108
+ ptp.Streams.Register(&stream)
109
+ stream.startStreaming()
110
+}
111
+
112
+// Listener wraps stream handler into a listener
113
+type Listener interface {
114
+ Accept() (net.Stream, error)
115
+ Close() error
116
+}
117
+
118
+// P2PListener holds information on a listener
119
+type P2PListener struct {
120
+ peerHost p2phost.Host
121
+ conCh chan net.Stream
122
+ proto pro.ID
123
+ ctx context.Context
124
+ cancel func()
125
+}
126
+
127
+// Accept waits for a connection from the listener
128
+func (il *P2PListener) Accept() (net.Stream, error) {
129
+ select {
130
+ case c := <-il.conCh:
131
+ return c, nil
132
+ case <-il.ctx.Done():
133
+ return nil, il.ctx.Err()
134
+ }
135
+}
136
+
137
+// Close closes the listener and removes stream handler
138
+func (il *P2PListener) Close() error {
139
+ il.cancel()
140
+ il.peerHost.RemoveStreamHandler(il.proto)
141
+ return nil
142
+}
143
+
144
+// Listen creates new P2PListener
145
+func (ptp *PTP) registerStreamHandler(ctx2 context.Context, protocol string) (*P2PListener, error) {
146
+ ctx, cancel := context.WithCancel(ctx2)
147
+
148
+ list := &P2PListener{
149
+ peerHost: ptp.peerHost,
150
+ proto: pro.ID(protocol),
151
+ conCh: make(chan net.Stream),
152
+ ctx: ctx,
153
+ cancel: cancel,
154
+ }
155
+
156
+ ptp.peerHost.SetStreamHandler(list.proto, func(s net.Stream) {
157
+ select {
158
+ case list.conCh <- s:
159
+ case <-ctx.Done():
160
+ s.Close()
161
+ }
162
+ })
163
+
164
+ return list, nil
165
+}
166
+
167
+// NewListener creates new ptp listener
168
+func (ptp *PTP) NewListener(ctx context.Context, proto string, addr ma.Multiaddr) (*ListenerInfo, error) {
169
+ listener, err := ptp.registerStreamHandler(ctx, proto)
170
+ if err != nil {
171
+ return nil, err
172
+ }
173
+
174
+ listenerInfo := ListenerInfo{
175
+ Identity: ptp.identity,
176
+ Protocol: proto,
177
+ Address: addr,
178
+ Closer: listener,
179
+ Running: true,
180
+ Registry: &ptp.Listeners,
181
+ }
182
+
183
+ go ptp.acceptStreams(&listenerInfo, listener)
184
+
185
+ ptp.Listeners.Register(&listenerInfo)
186
+
187
+ return &listenerInfo, nil
188
+}
189
+
190
+func (ptp *PTP) acceptStreams(listenerInfo *ListenerInfo, listener Listener) {
191
+ for listenerInfo.Running {
192
+ remote, err := listener.Accept()
193
+ if err != nil {
194
+ listener.Close()
195
+ break
196
+ }
197
+
198
+ local, err := manet.Dial(listenerInfo.Address)
199
+ if err != nil {
200
+ remote.Close()
201
+ continue
202
+ }
203
+
204
+ stream := StreamInfo{
205
+ Protocol: listenerInfo.Protocol,
206
+
207
+ LocalPeer: listenerInfo.Identity,
208
+ LocalAddr: listenerInfo.Address,
209
+
210
+ RemotePeer: remote.Conn().RemotePeer(),
211
+ RemoteAddr: remote.Conn().RemoteMultiaddr(),
212
+
213
+ Local: local,
214
+ Remote: remote,
215
+
216
+ Registry: &ptp.Streams,
217
+ }
218
+
219
+ ptp.Streams.Register(&stream)
220
+ stream.startStreaming()
221
+ }
222
+ ptp.Listeners.Deregister(listenerInfo.Protocol)
223
+}
224
+
225
+// CheckProtoExists checks whether a protocol handler is registered to
226
+// mux handler
227
+func (ptp *PTP) CheckProtoExists(proto string) bool {
228
+ protos := ptp.peerHost.Mux().Protocols()
229
+
230
+ for _, p := range protos {
231
+ if p != proto {
232
+ continue
233
+ }
234
+ return true
235
+ }
236
+ return false
237
}
ptp/registry.go
renamed
+69
-2
@@ -1,11 +1,11 @@
1
package ptp
2
3
import (
4
+ "fmt"
5
"io"
6
7
ma "gx/ipfs/QmcyqRMCAXVtYPS4DiBrA7sezL9rRGfW8Ctx7cywL4TXJj/go-multiaddr"
8
peer "gx/ipfs/QmdS9KpbDyPrieswibZhkod1oXqRwZJrUPzxCofAMWpFGq/go-libp2p-peer"
8
- "fmt"
9
)
10
11
// ListenerInfo holds information on a p2p listener.
@@ -41,7 +41,7 @@ type ListenerRegistry struct {
41
Listeners []*ListenerInfo
42
}
43
44
-// Register registers listenerInfo in this registry
44
+// Register registers listenerInfo2 in this registry
45
func (c *ListenerRegistry) Register(listenerInfo *ListenerInfo) {
46
c.Listeners = append(c.Listeners, listenerInfo)
47
}
@@ -63,3 +63,70 @@ func (c *ListenerRegistry) Deregister(proto string) error {
63
64
return fmt.Errorf("failed to deregister proto %s", proto)
65
}
66
+
67
+// StreamInfo holds information on active incoming and outgoing p2p streams.
68
+type StreamInfo struct {
69
+ HandlerID uint64
70
+
71
+ Protocol string
72
+
73
+ LocalPeer peer.ID
74
+ LocalAddr ma.Multiaddr
75
+
76
+ RemotePeer peer.ID
77
+ RemoteAddr ma.Multiaddr
78
+
79
+ Local io.ReadWriteCloser
80
+ Remote io.ReadWriteCloser
81
+
82
+ Registry *StreamRegistry
83
+}
84
+
85
+// Close closes stream endpoints and deregisters it
86
+func (c *StreamInfo) Close() error {
87
+ c.Local.Close()
88
+ c.Remote.Close()
89
+ c.Registry.Deregister(c.HandlerID)
90
+ return nil
91
+}
92
+
93
+func (s *StreamInfo) startStreaming() {
94
+ go func() {
95
+ io.Copy(s.Local, s.Remote)
96
+ s.Close()
97
+ }()
98
+
99
+ go func() {
100
+ io.Copy(s.Remote, s.Local)
101
+ s.Close()
102
+ }()
103
+}
104
+
105
+// StreamRegistry is a collection of active incoming and outgoing protocol app streams.
106
+type StreamRegistry struct {
107
+ Streams []*StreamInfo
108
+
109
+ nextID uint64
110
+}
111
+
112
+// Register registers a stream to the registry
113
+func (c *StreamRegistry) Register(streamInfo *StreamInfo) {
114
+ streamInfo.HandlerID = c.nextID
115
+ c.Streams = append(c.Streams, streamInfo)
116
+ c.nextID++
117
+}
118
+
119
+// Deregister deregisters stream from the registry
120
+func (c *StreamRegistry) Deregister(handlerID uint64) {
121
+ foundAt := -1
122
+ for i, s := range c.Streams {
123
+ if s.HandlerID == handlerID {
124
+ foundAt = i
125
+ break
126
+ }
127
+ }
128
+
129
+ if foundAt != -1 {
130
+ c.Streams = append(c.Streams[:foundAt], c.Streams[foundAt+1:]...)
131
+ }
132
+}
ptp/streams.go
deleted
-63
@@ -1,63 +0,0 @@
1
-package ptp
2
-
3
-import (
4
- "io"
5
-
6
- ma "gx/ipfs/QmcyqRMCAXVtYPS4DiBrA7sezL9rRGfW8Ctx7cywL4TXJj/go-multiaddr"
7
- peer "gx/ipfs/QmdS9KpbDyPrieswibZhkod1oXqRwZJrUPzxCofAMWpFGq/go-libp2p-peer"
8
-)
9
-
10
-// StreamInfo holds information on active incoming and outgoing p2p streams.
11
-type StreamInfo struct {
12
- HandlerID uint64
13
-
14
- Protocol string
15
-
16
- LocalPeer peer.ID
17
- LocalAddr ma.Multiaddr
18
-
19
- RemotePeer peer.ID
20
- RemoteAddr ma.Multiaddr
21
-
22
- Local io.ReadWriteCloser
23
- Remote io.ReadWriteCloser
24
-
25
- Registry *StreamRegistry
26
-}
27
-
28
-// Close closes stream endpoints and deregisters it
29
-func (c *StreamInfo) Close() error {
30
- c.Local.Close()
31
- c.Remote.Close()
32
- c.Registry.Deregister(c.HandlerID)
33
- return nil
34
-}
35
-
36
-// StreamRegistry is a collection of active incoming and outgoing protocol app streams.
37
-type StreamRegistry struct {
38
- Streams []*StreamInfo
39
-
40
- nextID uint64
41
-}
42
-
43
-// Register registers a stream to the registry
44
-func (c *StreamRegistry) Register(streamInfo *StreamInfo) {
45
- streamInfo.HandlerID = c.nextID
46
- c.Streams = append(c.Streams, streamInfo)
47
- c.nextID++
48
-}
49
-
50
-// Deregister deregisters stream from the registry
51
-func (c *StreamRegistry) Deregister(handlerID uint64) {
52
- foundAt := -1
53
- for i, s := range c.Streams {
54
- if s.HandlerID == handlerID {
55
- foundAt = i
56
- break
57
- }
58
- }
59
-
60
- if foundAt != -1 {
61
- c.Streams = append(c.Streams[:foundAt], c.Streams[foundAt+1:]...)
62
- }
63
-}
test/sharness/t0180-ptp.sh
+3
-3
@@ -66,7 +66,7 @@ test_expect_success 'client to server output looks good' '
66
'
67
68
test_expect_success "'ipfs ptp ls' succeeds" '
69
- echo "/ip4/127.0.0.1/tcp/10101 /app/ptp-test" > expected &&
69
+ echo "/ip4/127.0.0.1/tcp/10101 /ptp/ptp-test" > expected &&
70
ipfsi 0 ptp ls > actual
71
'
72
@@ -94,7 +94,7 @@ test_expect_success "Setup: Idle stream" '
94
'
95
96
test_expect_success "'ipfs ptp streams' succeeds" '
97
- echo "2 /app/ptp-test /ip4/127.0.0.1/tcp/10101 $PEERID_1" > expected
97
+ echo "2 /ptp/ptp-test /ip4/127.0.0.1/tcp/10101 $PEERID_1" > expected
98
ipfsi 0 ptp streams > actual
99
'
100
@@ -127,7 +127,7 @@ test_expect_success "Setup: Idle stream(2)" '
127
'
128
129
test_expect_success "'ipfs ptp streams' succeeds(2)" '
130
- echo "3 /app/ptp-test2 /ip4/127.0.0.1/tcp/10101 $PEERID_1" > expected
130
+ echo "3 /ptp/ptp-test2 /ip4/127.0.0.1/tcp/10101 $PEERID_1" > expected
131
ipfsi 0 ptp streams > actual
132
test_cmp expected actual
133
'