Corenet API: Store state in node
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
May 30, 2017 at 17:13 UTC
8948519590571222a3161a3dccdfceaa577c0669
6 files changed
+203
-164
core/commands/corenet.go
+70
-163
@@ -10,20 +10,22 @@ import (
10
11
cmds "github.com/ipfs/go-ipfs/commands"
12
core "github.com/ipfs/go-ipfs/core"
13
- corenet "github.com/ipfs/go-ipfs/core/corenet"
13
+ corenet "github.com/ipfs/go-ipfs/corenet"
14
+ cnet "github.com/ipfs/go-ipfs/corenet/net"
15
16
peerstore "gx/ipfs/QmNUVzEjq3XWJ89hegahPvyfJbTXgTaom48pLb7YBD9gHQ/go-libp2p-peerstore"
17
net "gx/ipfs/QmVHSBsn8LEeay8m5ERebgUVuhzw838PsyTttCmP6GMJkg/go-libp2p-net"
18
ma "gx/ipfs/QmcyqRMCAXVtYPS4DiBrA7sezL9rRGfW8Ctx7cywL4TXJj/go-multiaddr"
18
- peer "gx/ipfs/QmdS9KpbDyPrieswibZhkod1oXqRwZJrUPzxCofAMWpFGq/go-libp2p-peer"
19
manet "gx/ipfs/Qmf1Gq7N45Rpuw7ev47uWgH6dLPtdnvcMRNPkVBwqjLJg2/go-multiaddr-net"
20
)
21
22
+// CorenetAppInfoOutput is output type of ls command
23
type CorenetAppInfoOutput struct {
24
Protocol string
25
Address string
26
}
27
28
+// CorenetStreamInfoOutput is output type of streams command
29
type CorenetStreamInfoOutput struct {
30
HandlerID string
31
Protocol string
@@ -33,116 +35,16 @@ type CorenetStreamInfoOutput struct {
35
RemoteAddress string
36
}
37
38
+// CorenetLsOutput is output type of ls command
39
type CorenetLsOutput struct {
40
Apps []CorenetAppInfoOutput
41
}
42
43
+// CorenetStreamsOutput is output type of streams command
44
type CorenetStreamsOutput struct {
45
Streams []CorenetStreamInfoOutput
46
}
47
44
-// cnAppInfo holds information on a local application protocol listener service.
45
-type cnAppInfo struct {
46
- // Application protocol identifier.
47
- protocol string
48
-
49
- // Node identity
50
- identity peer.ID
51
-
52
- // Local protocol stream address.
53
- address ma.Multiaddr
54
-
55
- // Local protocol stream listener.
56
- closer io.Closer
57
-
58
- // Flag indicating whether we're still accepting incoming connections, or
59
- // whether this application listener has been shutdown.
60
- running bool
61
-}
62
-
63
-func (c *cnAppInfo) Close() error {
64
- apps.Deregister(c.protocol)
65
- c.closer.Close()
66
- return nil
67
-}
68
-
69
-// cnAppRegistry is a collection of local application protocol listeners.
70
-type cnAppRegistry struct {
71
- apps []*cnAppInfo
72
-}
73
-
74
-func (c *cnAppRegistry) Register(appInfo *cnAppInfo) {
75
- c.apps = append(c.apps, appInfo)
76
-}
77
-
78
-func (c *cnAppRegistry) Deregister(proto string) {
79
- foundAt := -1
80
- for i, a := range c.apps {
81
- if a.protocol == proto {
82
- foundAt = i
83
- break
84
- }
85
- }
86
-
87
- if foundAt != -1 {
88
- c.apps = append(c.apps[:foundAt], c.apps[foundAt+1:]...)
89
- }
90
-}
91
-
92
-// cnStreamInfo holds information on active incoming and outgoing protocol app streams.
93
-type cnStreamInfo struct {
94
- handlerID uint64
95
-
96
- protocol string
97
-
98
- localPeer peer.ID
99
- localAddr ma.Multiaddr
100
-
101
- remotePeer peer.ID
102
- remoteAddr ma.Multiaddr
103
-
104
- local io.ReadWriteCloser
105
- remote io.ReadWriteCloser
106
-}
107
-
108
-func (c *cnStreamInfo) Close() error {
109
- c.local.Close()
110
- c.remote.Close()
111
- streams.Deregister(c.handlerID)
112
- return nil
113
-}
114
-
115
-// cnStreamRegistry is a collection of active incoming and outgoing protocol app streams.
116
-type cnStreamRegistry struct {
117
- streams []*cnStreamInfo
118
-
119
- nextID uint64
120
-}
121
-
122
-func (c *cnStreamRegistry) Register(streamInfo *cnStreamInfo) {
123
- streamInfo.handlerID = c.nextID
124
- c.streams = append(c.streams, streamInfo)
125
- c.nextID++
126
-}
127
-
128
-func (c *cnStreamRegistry) Deregister(handlerID uint64) {
129
- foundAt := -1
130
- for i, s := range c.streams {
131
- if s.handlerID == handlerID {
132
- foundAt = i
133
- break
134
- }
135
- }
136
-
137
- if foundAt != -1 {
138
- c.streams = append(c.streams[:foundAt], c.streams[foundAt+1:]...)
139
- }
140
-}
141
-
142
-//TODO: Ideally I'd like to see these combined into a module in core.
143
-var apps cnAppRegistry
144
-var streams cnStreamRegistry
145
-
48
var CorenetCmd = &cmds.Command{
49
Helptext: cmds.HelpText{
50
Tagline: "Libp2p stream mounting.",
@@ -188,10 +90,10 @@ var CorenetLsCmd = &cmds.Command{
90
91
output := &CorenetLsOutput{}
92
191
- for _, a := range apps.apps {
93
+ for _, app := range n.Corenet.Apps.Apps {
94
output.Apps = append(output.Apps, CorenetAppInfoOutput{
193
- Protocol: a.protocol,
194
- Address: a.address.String(),
95
+ Protocol: app.Protocol,
96
+ Address: app.Address.String(),
97
})
98
}
99
@@ -245,17 +147,17 @@ var CorenetStreamsCmd = &cmds.Command{
147
148
output := &CorenetStreamsOutput{}
149
248
- for _, s := range streams.streams {
150
+ for _, s := range n.Corenet.Streams.Streams {
151
output.Streams = append(output.Streams, CorenetStreamInfoOutput{
250
- HandlerID: strconv.FormatUint(s.handlerID, 10),
152
+ HandlerID: strconv.FormatUint(s.HandlerID, 10),
153
252
- Protocol: s.protocol,
154
+ Protocol: s.Protocol,
155
254
- LocalPeer: s.localPeer.Pretty(),
255
- LocalAddress: s.localAddr.String(),
156
+ LocalPeer: s.LocalPeer.Pretty(),
157
+ LocalAddress: s.LocalAddr.String(),
158
257
- RemotePeer: s.remotePeer.Pretty(),
258
- RemoteAddress: s.remoteAddr.String(),
159
+ RemotePeer: s.RemotePeer.Pretty(),
160
+ RemoteAddress: s.RemoteAddr.String(),
161
})
162
}
163
@@ -320,23 +222,24 @@ var CorenetListenCmd = &cmds.Command{
222
return
223
}
224
323
- listener, err := corenet.Listen(n, proto)
225
+ listener, err := cnet.Listen(n, proto)
226
if err != nil {
227
res.SetError(err, cmds.ErrNormal)
228
return
229
}
230
329
- app := cnAppInfo{
330
- identity: n.Identity,
331
- protocol: proto,
332
- address: addr,
333
- closer: listener,
334
- running: true,
231
+ app := corenet.AppInfo{
232
+ Identity: n.Identity,
233
+ Protocol: proto,
234
+ Address: addr,
235
+ Closer: listener,
236
+ Running: true,
237
+ Registry: &n.Corenet.Apps,
238
}
239
337
- go acceptStreams(&app, listener)
240
+ go acceptStreams(n, &app, listener)
241
339
- apps.Register(&app)
242
+ n.Corenet.Apps.Register(&app)
243
244
// Successful response.
245
res.SetOutput(&CorenetAppInfoOutput{
@@ -356,47 +259,49 @@ func checkProtoExists(protos []string, proto string) bool {
259
return false
260
}
261
359
-func acceptStreams(app *cnAppInfo, listener corenet.Listener) {
360
- for app.running {
262
+func acceptStreams(n *core.IpfsNode, app *corenet.AppInfo, listener cnet.Listener) {
263
+ for app.Running {
264
remote, err := listener.Accept()
265
if err != nil {
266
listener.Close()
267
break
268
}
269
367
- local, err := manet.Dial(app.address)
270
+ local, err := manet.Dial(app.Address)
271
if err != nil {
272
remote.Close()
273
continue
274
}
275
373
- stream := cnStreamInfo{
374
- protocol: app.protocol,
276
+ stream := corenet.StreamInfo{
277
+ Protocol: app.Protocol,
278
+
279
+ LocalPeer: app.Identity,
280
+ LocalAddr: app.Address,
281
376
- localPeer: app.identity,
377
- localAddr: app.address,
282
+ RemotePeer: remote.Conn().RemotePeer(),
283
+ RemoteAddr: remote.Conn().RemoteMultiaddr(),
284
379
- remotePeer: remote.Conn().RemotePeer(),
380
- remoteAddr: remote.Conn().RemoteMultiaddr(),
285
+ Local: local,
286
+ Remote: remote,
287
382
- local: local,
383
- remote: remote,
288
+ Registry: &n.Corenet.Streams,
289
}
290
386
- streams.Register(&stream)
291
+ n.Corenet.Streams.Register(&stream)
292
startStreaming(&stream)
293
}
389
- apps.Deregister(app.protocol)
294
+ n.Corenet.Apps.Deregister(app.Protocol)
295
}
296
392
-func startStreaming(stream *cnStreamInfo) {
297
+func startStreaming(stream *corenet.StreamInfo) {
298
go func() {
394
- io.Copy(stream.local, stream.remote)
299
+ io.Copy(stream.Local, stream.Remote)
300
stream.Close()
301
}()
302
303
go func() {
399
- io.Copy(stream.remote, stream.local)
304
+ io.Copy(stream.Remote, stream.Local)
305
stream.Close()
306
}()
307
}
@@ -451,14 +356,14 @@ var CorenetDialCmd = &cmds.Command{
356
return
357
}
358
454
- app := cnAppInfo{
455
- identity: n.Identity,
456
- protocol: proto,
359
+ app := corenet.AppInfo{
360
+ Identity: n.Identity,
361
+ Protocol: proto,
362
}
363
364
n.Peerstore.AddAddr(peer, addr, peerstore.TempAddrTTL)
365
461
- remote, err := corenet.Dial(n, peer, proto)
366
+ remote, err := cnet.Dial(n, peer, proto)
367
if err != nil {
368
res.SetError(err, cmds.ErrNormal)
369
return
@@ -475,11 +380,11 @@ var CorenetDialCmd = &cmds.Command{
380
return
381
}
382
478
- app.address = listener.Multiaddr()
479
- app.closer = listener
480
- app.running = true
383
+ app.Address = listener.Multiaddr()
384
+ app.Closer = listener
385
+ app.Running = true
386
482
- go doAccept(&app, remote, listener)
387
+ go doAccept(n, &app, remote, listener)
388
389
default:
390
res.SetError(errors.New("unsupported protocol: "+lnet), cmds.ErrNormal)
@@ -487,15 +392,15 @@ var CorenetDialCmd = &cmds.Command{
392
}
393
394
output := CorenetAppInfoOutput{
490
- Protocol: app.protocol,
491
- Address: app.address.String(),
395
+ Protocol: app.Protocol,
396
+ Address: app.Address.String(),
397
}
398
399
res.SetOutput(&output)
400
},
401
}
402
498
-func doAccept(app *cnAppInfo, remote net.Stream, listener manet.Listener) {
403
+func doAccept(n *core.IpfsNode, app *corenet.AppInfo, remote net.Stream, listener manet.Listener) {
404
defer listener.Close()
405
406
local, err := listener.Accept()
@@ -503,20 +408,22 @@ func doAccept(app *cnAppInfo, remote net.Stream, listener manet.Listener) {
408
return
409
}
410
506
- stream := cnStreamInfo{
507
- protocol: app.protocol,
411
+ stream := corenet.StreamInfo{
412
+ Protocol: app.Protocol,
413
+
414
+ LocalPeer: app.Identity,
415
+ LocalAddr: app.Address,
416
509
- localPeer: app.identity,
510
- localAddr: app.address,
417
+ RemotePeer: remote.Conn().RemotePeer(),
418
+ RemoteAddr: remote.Conn().RemoteMultiaddr(),
419
512
- remotePeer: remote.Conn().RemotePeer(),
513
- remoteAddr: remote.Conn().RemoteMultiaddr(),
420
+ Local: local,
421
+ Remote: remote,
422
515
- local: local,
516
- remote: remote,
423
+ Registry: &n.Corenet.Streams,
424
}
425
519
- streams.Register(&stream)
426
+ n.Corenet.Streams.Register(&stream)
427
startStreaming(&stream)
428
}
429
@@ -571,8 +478,8 @@ var CorenetCloseCmd = &cmds.Command{
478
}
479
480
if closeAll || useHandlerID {
574
- for _, s := range streams.streams {
575
- if !closeAll && handlerID != s.handlerID {
481
+ for _, s := range n.Corenet.Streams.Streams {
482
+ if !closeAll && handlerID != s.HandlerID {
483
continue
484
}
485
s.Close()
@@ -583,8 +490,8 @@ var CorenetCloseCmd = &cmds.Command{
490
}
491
492
if closeAll || !useHandlerID {
586
- for _, a := range apps.apps {
587
- if !closeAll && a.protocol != proto {
493
+ for _, a := range n.Corenet.Apps.Apps {
494
+ if !closeAll && a.Protocol != proto {
495
continue
496
}
497
a.Close()
core/core.go
+4
@@ -23,6 +23,7 @@ import (
23
24
bstore "github.com/ipfs/go-ipfs/blocks/blockstore"
25
bserv "github.com/ipfs/go-ipfs/blockservice"
26
+ corenet "github.com/ipfs/go-ipfs/corenet"
27
exchange "github.com/ipfs/go-ipfs/exchange"
28
bitswap "github.com/ipfs/go-ipfs/exchange/bitswap"
29
bsnet "github.com/ipfs/go-ipfs/exchange/bitswap/network"
@@ -131,6 +132,7 @@ type IpfsNode struct {
132
IpnsRepub *ipnsrp.Republisher
133
134
Floodsub *floodsub.PubSub
135
+ Corenet *corenet.Corenet
136
137
proc goprocess.Process
138
ctx context.Context
@@ -246,6 +248,8 @@ func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption Routin
248
n.Floodsub = floodsub.NewFloodSub(ctx, peerhost)
249
}
250
251
+ n.Corenet = corenet.NewCorenet()
252
+
253
// setup local discovery
254
if do != nil {
255
service, err := do(ctx, n.PeerHost)
corenet/apps.go
new
+58
@@ -0,0 +1,58 @@
1
+package corenet
2
+
3
+import (
4
+ "io"
5
+
6
+ ma "gx/ipfs/QmcyqRMCAXVtYPS4DiBrA7sezL9rRGfW8Ctx7cywL4TXJj/go-multiaddr"
7
+ peer "gx/ipfs/QmdS9KpbDyPrieswibZhkod1oXqRwZJrUPzxCofAMWpFGq/go-libp2p-peer"
8
+)
9
+
10
+// AppInfo holds information on a local application protocol listener service.
11
+type AppInfo struct {
12
+ // Application protocol identifier.
13
+ Protocol string
14
+
15
+ // Node identity
16
+ Identity peer.ID
17
+
18
+ // Local protocol stream address.
19
+ Address ma.Multiaddr
20
+
21
+ // Local protocol stream listener.
22
+ Closer io.Closer
23
+
24
+ // Flag indicating whether we're still accepting incoming connections, or
25
+ // whether this application listener has been shutdown.
26
+ Running bool
27
+
28
+ Registry *AppRegistry
29
+}
30
+
31
+func (c *AppInfo) Close() error {
32
+ c.Registry.Deregister(c.Protocol)
33
+ c.Closer.Close()
34
+ return nil
35
+}
36
+
37
+// AppRegistry is a collection of local application protocol listeners.
38
+type AppRegistry struct {
39
+ Apps []*AppInfo
40
+}
41
+
42
+func (c *AppRegistry) Register(appInfo *AppInfo) {
43
+ c.Apps = append(c.Apps, appInfo)
44
+}
45
+
46
+func (c *AppRegistry) Deregister(proto string) {
47
+ foundAt := -1
48
+ for i, a := range c.Apps {
49
+ if a.Protocol == proto {
50
+ foundAt = i
51
+ break
52
+ }
53
+ }
54
+
55
+ if foundAt != -1 {
56
+ c.Apps = append(c.Apps[:foundAt], c.Apps[foundAt+1:]...)
57
+ }
58
+}
corenet/corenet.go
new
+10
@@ -0,0 +1,10 @@
1
+package corenet
2
+
3
+type Corenet struct {
4
+ Apps AppRegistry
5
+ Streams StreamRegistry
6
+}
7
+
8
+func NewCorenet() *Corenet {
9
+ return &Corenet{}
10
+}
corenet/net/net.go
renamed
+1
-1
@@ -1,4 +1,4 @@
1
-package corenet
1
+package net
2
3
import (
4
"time"
corenet/streams.go
new
+60
@@ -0,0 +1,60 @@
1
+package corenet
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 protocol app 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
+func (c *StreamInfo) Close() error {
29
+ c.Local.Close()
30
+ c.Remote.Close()
31
+ c.Registry.Deregister(c.HandlerID)
32
+ return nil
33
+}
34
+
35
+// StreamRegistry is a collection of active incoming and outgoing protocol app streams.
36
+type StreamRegistry struct {
37
+ Streams []*StreamInfo
38
+
39
+ nextID uint64
40
+}
41
+
42
+func (c *StreamRegistry) Register(streamInfo *StreamInfo) {
43
+ streamInfo.HandlerID = c.nextID
44
+ c.Streams = append(c.Streams, streamInfo)
45
+ c.nextID++
46
+}
47
+
48
+func (c *StreamRegistry) Deregister(handlerID uint64) {
49
+ foundAt := -1
50
+ for i, s := range c.Streams {
51
+ if s.HandlerID == handlerID {
52
+ foundAt = i
53
+ break
54
+ }
55
+ }
56
+
57
+ if foundAt != -1 {
58
+ c.Streams = append(c.Streams[:foundAt], c.Streams[foundAt+1:]...)
59
+ }
60
+}