@cryptotaxi247 / kubo / commits / 2f17a67c5

Experimental corenet application support.

License: MIT Signed-off-by: Tom Swindell <t.swindell@rubyx.co.uk>

Tom Swindell committed Dec 19, 2016 at 13:03 UTC 2f17a67c5a0b3dfad48f797f2a10695d11e30980
5 files changed +590 -1
core/commands/corenet.go new
+500
@@ -0,0 +1,500 @@
1 +package commands
2 +
3 +import (
4 + "errors"
5 + "io"
6 + "strconv"
7 +
8 + cmds "github.com/ipfs/go-ipfs/commands"
9 + corenet "github.com/ipfs/go-ipfs/core/corenet"
10 +
11 + manet "gx/ipfs/QmT6Cp31887FpAc25z25YHgpFJohZedrYLWPPspRtj1Brp/go-multiaddr-net"
12 + ma "gx/ipfs/QmUAQaWbKxGCUTuoQVvvicbQNZ9APF5pDGWyAZSe93AtKH/go-multiaddr"
13 + net "gx/ipfs/QmQx1dHDDYENugYgqA22BaBrRfuv1coSsuPiM7rYh1wwGH/go-libp2p-net"
14 + peerstore "gx/ipfs/QmeXj9VAjmYQZxpmVz7VzccbJrpmr8qkCDSjfVNsPTWTYU/go-libp2p-peerstore"
15 + peer "gx/ipfs/QmfMmLGoKzCHDN7cGgk64PJr4iipzidDRME8HABSJqvmhC/go-libp2p-peer"
16 +)
17 +
18 +// Command output types.
19 +type AppInfoOutput struct {
20 + Identity string
21 + Protocol string
22 + Address string
23 +}
24 +
25 +type StreamInfoOutput struct {
26 + HandlerId string
27 + Protocol string
28 + LocalPeer string
29 + LocalAddress string
30 + RemotePeer string
31 + RemoteAddress string
32 +}
33 +
34 +type ListCommandOutput struct {
35 + Apps []AppInfoOutput
36 + Streams []StreamInfoOutput
37 +}
38 +
39 +// cnAppInfo holds information on a local application protocol listener service.
40 +type cnAppInfo struct {
41 + // Application protocol identifier.
42 + protocol string
43 +
44 + // Node identity
45 + identity peer.ID
46 +
47 + // Local protocol stream address.
48 + address ma.Multiaddr
49 +
50 + // Local protocol stream listener.
51 + closer io.Closer
52 +
53 + // Flag indicating whether we're still accepting incoming connections, or
54 + // whether this application listener has been shutdown.
55 + running bool
56 +}
57 +
58 +func (c *cnAppInfo) Close() error {
59 + apps.Deregister(c.protocol)
60 + c.closer.Close()
61 + return nil
62 +}
63 +
64 +// cnAppRegistry is a collection of local application protocol listeners.
65 +type cnAppRegistry struct {
66 + apps []*cnAppInfo
67 +}
68 +
69 +func (c *cnAppRegistry) Register(appInfo *cnAppInfo) {
70 + c.apps = append(c.apps, appInfo)
71 +}
72 +
73 +func (c *cnAppRegistry) Deregister(proto string) {
74 + foundAt := -1
75 + for i, a := range c.apps {
76 + if a.protocol == proto {
77 + foundAt = i
78 + break
79 + }
80 + }
81 +
82 + if foundAt != -1 {
83 + c.apps = append(c.apps[:foundAt], c.apps[foundAt+1:]...)
84 + }
85 +}
86 +
87 +// cnStreamInfo holds information on active incoming and outgoing protocol app streams.
88 +type cnStreamInfo struct {
89 + handlerId uint64
90 +
91 + protocol string
92 +
93 + localPeer peer.ID
94 + localAddr ma.Multiaddr
95 +
96 + remotePeer peer.ID
97 + remoteAddr ma.Multiaddr
98 +
99 + local io.ReadWriteCloser
100 + remote io.ReadWriteCloser
101 +}
102 +
103 +func (c *cnStreamInfo) Close() error {
104 + c.local.Close()
105 + c.remote.Close()
106 + streams.Deregister(c.handlerId)
107 + return nil
108 +}
109 +
110 +// cnStreamRegistry is a collection of active incoming and outgoing protocol app streams.
111 +type cnStreamRegistry struct {
112 + streams []*cnStreamInfo
113 +
114 + nextId uint64
115 +}
116 +
117 +func (c *cnStreamRegistry) Register(streamInfo *cnStreamInfo) {
118 + streamInfo.handlerId = c.nextId
119 + c.streams = append(c.streams, streamInfo)
120 + c.nextId += 1
121 +}
122 +
123 +func (c *cnStreamRegistry) Deregister(handlerId uint64) {
124 + foundAt := -1
125 + for i, s := range c.streams {
126 + if s.handlerId == handlerId {
127 + foundAt = i
128 + break
129 + }
130 + }
131 +
132 + if foundAt != -1 {
133 + c.streams = append(c.streams[:foundAt], c.streams[foundAt+1:]...)
134 + }
135 +}
136 +
137 +//TODO: Ideally I'd like to see these combined into a module in core.
138 +var apps cnAppRegistry
139 +var streams cnStreamRegistry
140 +
141 +var CorenetCmd = &cmds.Command{
142 + Helptext: cmds.HelpText{
143 + Tagline: "Application network streams.",
144 + },
145 +
146 + Subcommands: map[string]*cmds.Command{
147 + "list": listCmd,
148 + "dial": dialCmd,
149 + "listen": listenCmd,
150 + "close": closeCmd,
151 + },
152 +}
153 +
154 +var listCmd = &cmds.Command{
155 + Helptext: cmds.HelpText{
156 + Tagline: "List active application protocol connections.",
157 + },
158 + Options: []cmds.Option{
159 + cmds.BoolOption("apps", "a", "Display only local application protocol listeners.").Default(false),
160 + cmds.BoolOption("streams", "s", "Display active application protocol streams.").Default(false),
161 + },
162 + Run: func(req cmds.Request, res cmds.Response) {
163 + n, err := req.InvocContext().GetNode()
164 + if err != nil {
165 + res.SetError(err, cmds.ErrNormal)
166 + return
167 + }
168 +
169 + if !n.OnlineMode() {
170 + res.SetError(errNotOnline, cmds.ErrClient)
171 + return
172 + }
173 +
174 + var output ListCommandOutput
175 +
176 + for _, a := range apps.apps {
177 + output.Apps = append(output.Apps, AppInfoOutput{
178 + Identity: a.identity.Pretty(),
179 + Protocol: a.protocol,
180 + Address: a.address.String(),
181 + })
182 + }
183 +
184 + for _, s := range streams.streams {
185 + output.Streams = append(output.Streams, StreamInfoOutput{
186 + HandlerId: strconv.FormatUint(s.handlerId, 10),
187 +
188 + Protocol: s.protocol,
189 +
190 + LocalPeer: s.localPeer.Pretty(),
191 + LocalAddress: s.localAddr.String(),
192 +
193 + RemotePeer: s.remotePeer.Pretty(),
194 + RemoteAddress: s.remoteAddr.String(),
195 + })
196 + }
197 +
198 + res.SetOutput(&output)
199 + },
200 +}
201 +
202 +var listenCmd = &cmds.Command{
203 + Helptext: cmds.HelpText{
204 + Tagline: "Create application protocol listener and proxy to network multiaddr.",
205 + },
206 + Arguments: []cmds.Argument{
207 + cmds.StringArg("Protocol", true, false, "Protocol identifier."),
208 + cmds.StringArg("Address", true, false, "Request handling application address."),
209 + },
210 + Run: func(req cmds.Request, res cmds.Response) {
211 + n, err := req.InvocContext().GetNode()
212 + if err != nil {
213 + res.SetError(err, cmds.ErrNormal)
214 + return
215 + }
216 +
217 + if !n.OnlineMode() {
218 + res.SetError(errNotOnline, cmds.ErrClient)
219 + return
220 + }
221 +
222 + proto := "/app/" + req.Arguments()[0]
223 + if checkProtoExists(n.PeerHost.Mux().Protocols(), proto) {
224 + res.SetError(errors.New("Protocol handler already registered."), cmds.ErrNormal)
225 + return
226 + }
227 +
228 + addr, err := ma.NewMultiaddr(req.Arguments()[1])
229 + if err != nil {
230 + res.SetError(err, cmds.ErrNormal)
231 + return
232 + }
233 +
234 + listener, err := corenet.Listen(n, proto)
235 + if err != nil {
236 + res.SetError(err, cmds.ErrNormal)
237 + return
238 + }
239 +
240 + app := cnAppInfo{
241 + identity: n.Identity,
242 + protocol: proto,
243 + address: addr,
244 + closer: listener,
245 + running: true,
246 + }
247 +
248 + go acceptStreams(&app, listener)
249 +
250 + apps.Register(&app)
251 +
252 + // Successful response.
253 + res.SetOutput(&AppInfoOutput{
254 + Identity: app.identity.Pretty(),
255 + Protocol: proto,
256 + Address: addr.String(),
257 + })
258 + },
259 +}
260 +
261 +func checkProtoExists(protos []string, proto string) bool {
262 + for _, p := range protos {
263 + if p != proto {
264 + continue
265 + }
266 + return true
267 + }
268 + return false
269 +}
270 +
271 +func acceptStreams(app *cnAppInfo, listener corenet.Listener) {
272 + for app.running {
273 + remote, err := listener.Accept()
274 + if err != nil {
275 + listener.Close()
276 + break
277 + }
278 +
279 + local, err := manet.Dial(app.address)
280 + if err != nil {
281 + remote.Close()
282 + continue
283 + }
284 +
285 + stream := cnStreamInfo{
286 + protocol: app.protocol,
287 +
288 + localPeer: app.identity,
289 + localAddr: app.address,
290 +
291 + remotePeer: remote.Conn().RemotePeer(),
292 + remoteAddr: remote.Conn().RemoteMultiaddr(),
293 +
294 + local: local,
295 + remote: remote,
296 + }
297 +
298 + streams.Register(&stream)
299 + startStreaming(&stream)
300 + }
301 + apps.Deregister(app.protocol)
302 +}
303 +
304 +func startStreaming(stream *cnStreamInfo) {
305 + go func() {
306 + io.Copy(stream.local, stream.remote)
307 + stream.Close()
308 + }()
309 +
310 + go func() {
311 + io.Copy(stream.remote, stream.local)
312 + stream.Close()
313 + }()
314 +}
315 +
316 +var dialCmd = &cmds.Command{
317 + Helptext: cmds.HelpText{
318 + Tagline: "Dial to an application service.",
319 + },
320 + Arguments: []cmds.Argument{
321 + cmds.StringArg("Peer", true, false, "Remote peer to connect to"),
322 + cmds.StringArg("Protocol", true, false, "Protocol identifier."),
323 + cmds.StringArg("BindAddress", false, false, "Address to listen for application/s (default: /ip4/127.0.0.1/tcp/0)."),
324 + },
325 + Run: func(req cmds.Request, res cmds.Response) {
326 + n, err := req.InvocContext().GetNode()
327 + if err != nil {
328 + res.SetError(err, cmds.ErrNormal)
329 + return
330 + }
331 +
332 + if !n.OnlineMode() {
333 + res.SetError(errNotOnline, cmds.ErrClient)
334 + return
335 + }
336 +
337 + addr, peer, err := ParsePeerParam(req.Arguments()[0])
338 + if err != nil {
339 + res.SetError(err, cmds.ErrNormal)
340 + return
341 + }
342 +
343 + proto := "/app/" + req.Arguments()[1]
344 +
345 + bindAddr, _ := ma.NewMultiaddr("/ip4/127.0.0.1/tcp/0")
346 + if len(req.Arguments()) == 3 {
347 + bindAddr, err = ma.NewMultiaddr(req.Arguments()[2])
348 + if err != nil {
349 + res.SetError(err, cmds.ErrNormal)
350 + return
351 + }
352 + }
353 +
354 + lnet, _, err := manet.DialArgs(bindAddr)
355 + if err != nil {
356 + res.SetError(err, cmds.ErrNormal)
357 + return
358 + }
359 +
360 + app := cnAppInfo{
361 + identity: n.Identity,
362 + protocol: proto,
363 + }
364 +
365 + n.Peerstore.AddAddr(peer, addr, peerstore.TempAddrTTL)
366 +
367 + remote, err := corenet.Dial(n, peer, proto)
368 + if err != nil {
369 + res.SetError(err, cmds.ErrNormal)
370 + return
371 + }
372 +
373 + switch lnet {
374 + case "tcp", "tcp4", "tcp6":
375 + listener, err := manet.Listen(bindAddr)
376 + if err != nil {
377 + res.SetError(err, cmds.ErrNormal)
378 + if err := remote.Close(); err != nil {
379 + res.SetError(err, cmds.ErrNormal)
380 + }
381 + return
382 + }
383 +
384 + app.address = listener.Multiaddr()
385 + app.closer = listener
386 + app.running = true
387 +
388 + go doAccept(&app, remote, listener)
389 +
390 + default:
391 + res.SetError(errors.New("Unsupported protocol: "+lnet), cmds.ErrNormal)
392 + return
393 + }
394 +
395 + output := AppInfoOutput{
396 + Identity: app.identity.Pretty(),
397 + Protocol: app.protocol,
398 + Address: app.address.String(),
399 + }
400 +
401 + res.SetOutput(&output)
402 + },
403 +}
404 +
405 +func doAccept(app *cnAppInfo, remote net.Stream, listener manet.Listener) {
406 + local, err := listener.Accept()
407 + if err != nil {
408 + return
409 + }
410 + defer listener.Close()
411 +
412 + stream := cnStreamInfo{
413 + protocol: app.protocol,
414 +
415 + localPeer: app.identity,
416 + localAddr: app.address,
417 +
418 + remotePeer: remote.Conn().RemotePeer(),
419 + remoteAddr: remote.Conn().RemoteMultiaddr(),
420 +
421 + local: local,
422 + remote: remote,
423 + }
424 +
425 + streams.Register(&stream)
426 + startStreaming(&stream)
427 +}
428 +
429 +var closeCmd = &cmds.Command{
430 + Helptext: cmds.HelpText{
431 + Tagline: "Closes an active stream listener or client.",
432 + },
433 + Arguments: []cmds.Argument{
434 + cmds.StringArg("HandlerId", false, false, "Application listener or client HandlerId"),
435 + cmds.StringArg("Protocol", false, false, "Application listener or client HandlerId"),
436 + },
437 + Options: []cmds.Option{
438 + cmds.BoolOption("all", "a", "Close all streams and listeners.").Default(false),
439 + },
440 + Run: func(req cmds.Request, res cmds.Response) {
441 + n, err := req.InvocContext().GetNode()
442 + if err != nil {
443 + res.SetError(err, cmds.ErrNormal)
444 + return
445 + }
446 +
447 + if !n.OnlineMode() {
448 + res.SetError(errNotOnline, cmds.ErrClient)
449 + return
450 + }
451 +
452 + closeAll, _, _ := req.Option("all").Bool()
453 +
454 + var proto string
455 + var handlerId uint64
456 +
457 + useHandlerId := false
458 +
459 + if !closeAll && len(req.Arguments()) == 0 {
460 + res.SetError(errors.New("You must supply a handlerId or stream protocol."), cmds.ErrNormal)
461 + return
462 +
463 + } else if !closeAll {
464 + handlerId, err = strconv.ParseUint(req.Arguments()[0], 10, 64)
465 + if err != nil {
466 + proto = "/app/" + req.Arguments()[0]
467 +
468 + } else {
469 + useHandlerId = true
470 + }
471 + }
472 +
473 + if closeAll || useHandlerId {
474 + for _, s := range streams.streams {
475 + if !closeAll && handlerId != s.handlerId {
476 + continue
477 + }
478 + s.Close()
479 + if !closeAll {
480 + break
481 + }
482 + }
483 + }
484 +
485 + if closeAll || !useHandlerId {
486 + for _, a := range apps.apps {
487 + if !closeAll && a.protocol != proto {
488 + continue
489 + }
490 + a.Close()
491 + if !closeAll {
492 + break
493 + }
494 + }
495 + }
496 +
497 + if len(req.Arguments()) != 1 {
498 + }
499 + },
500 +}
core/commands/experimental.go new
+15
@@ -0,0 +1,15 @@
1 +package commands
2 +
3 +import (
4 + cmds "github.com/ipfs/go-ipfs/commands"
5 +)
6 +
7 +var ExpCmd = &cmds.Command{
8 + Helptext: cmds.HelpText{
9 + Tagline: "Experimental commands",
10 + ShortDescription: `'ipfs exp' groups experimental features that are subject to change or removal at any time.`,
11 + },
12 + Subcommands: map[string]*cmds.Command{
13 + "corenet": CorenetCmd,
14 + },
15 +}
core/commands/root.go
+1
@@ -47,6 +47,7 @@ ADVANCED COMMANDS
47 pin Pin objects to local storage
48 repo Manipulate the IPFS repository
49 stats Various operational stats
50 + exp Experimental commands
51 filestore Manage the filestore (experimental)
52
53 NETWORK COMMANDS
core/corenet/net.go
+8 -1
@@ -11,7 +11,13 @@ import (
11 peer "gx/ipfs/QmdS9KpbDyPrieswibZhkod1oXqRwZJrUPzxCofAMWpFGq/go-libp2p-peer"
12 )
13
14 +type Listener interface {
15 + Accept() (net.Stream, error)
16 + Close() error
17 +}
18 +
19 type ipfsListener struct {
20 + node *core.IpfsNode
21 conCh chan net.Stream
22 proto pro.ID
23 ctx context.Context
@@ -29,7 +35,7 @@ func (il *ipfsListener) Accept() (net.Stream, error) {
35
36 func (il *ipfsListener) Close() error {
37 il.cancel()
32 - // TODO: unregister handler from peerhost
38 + il.node.PeerHost.RemoveStreamHandler(il.proto)
39 return nil
40 }
41
@@ -37,6 +43,7 @@ func Listen(nd *core.IpfsNode, protocol string) (*ipfsListener, error) {
43 ctx, cancel := context.WithCancel(nd.Context())
44
45 list := &ipfsListener{
46 + node: nd,
47 proto: pro.ID(protocol),
48 conCh: make(chan net.Stream),
49 ctx: ctx,
test/sharness/t0180-corenet.sh new
+66
@@ -0,0 +1,66 @@
1 +#!/bin/sh
2 +
3 +test_description="Test experimental corenet commands"
4 +
5 +. lib/test-lib.sh
6 +
7 +# start iptb + wait for peering
8 +test_expect_success 'init iptb' '
9 + iptb init -n 2 --bootstrap=none --port=0
10 +'
11 +
12 +test_expect_success 'generate test data' '
13 + echo "ABCDEF" > corenet0.bin &&
14 + echo "012345" > corenet1.bin
15 +'
16 +
17 +startup_cluster 2
18 +
19 +test_expect_success 'peer ids' '
20 + PEERID_0=$(iptb get id 0) &&
21 + PEERID_1=$(iptb get id 1)
22 +'
23 +
24 +# netcat (nc) is needed for the following tests
25 +test_expect_success "nc is available" '
26 + type nc >/dev/null
27 +'
28 +
29 +test_expect_success 'start ipfs listener' '
30 + ipfsi 0 exp corenet listen corenet-test /ip4/127.0.0.1/tcp/10001 2>&1 > listener-stdouterr.log
31 +'
32 +
33 +test_expect_success 'Test server to client communications' '
34 + dd if=corenet0.bin | nc -l 127.0.0.1 10001 &
35 + NC_SERVER_PID=$!
36 +
37 + ipfsi 1 exp corenet dial $PEERID_0 corenet-test /ip4/127.0.0.1/tcp/10002 2>&1 > dialer-stdouterr.log &&
38 + nc -v 127.0.0.1 10002 | dd of=client.out &&
39 +
40 + wait $NC_SERVER_PID
41 +'
42 +
43 +test_expect_success 'Test server to client communications' '
44 + nc -l 127.0.0.1 10001 | dd of=server.out &
45 + NC_SERVER_PID=$!
46 +
47 + ipfsi 1 exp corenet dial $PEERID_0 corenet-test /ip4/127.0.0.1/tcp/10002 2>&1 > dialer-stdouterr.log &&
48 + dd of=corenet1.bin | nc -v 127.0.0.1 10002 &&
49 +
50 + wait $NC_SERVER_PID
51 +'
52 +
53 +test_expect_success 'server to client output looks good' '
54 + test_cmp client.out corenet0.bin
55 +'
56 +
57 +test_expect_success 'client to server output looks good' '
58 + test_cmp server.out corenet1.bin
59 +'
60 +
61 +test_expect_success 'stop iptb' '
62 + iptb stop
63 +'
64 +
65 +test_done
66 +