coreapi: implement swarm api
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Mar 10, 2018 at 19:02 UTC
8358c8d0410c9da896ade974c9d08212b1f2b39a
4 files changed
+158
-3
core/coreapi/coreapi.go
+5
@@ -67,3 +67,8 @@ func (api *CoreAPI) Pin() coreiface.PinAPI {
67
func (api *CoreAPI) Dht() coreiface.DhtAPI {
68
return (*DhtAPI)(api)
69
}
70
+
71
+// Swarm returns the SwarmAPI interface implementation backed by the go-ipfs node
72
+func (api *CoreAPI) Swarm() coreiface.SwarmAPI {
73
+ return &SwarmAPI{api}
74
+}
core/coreapi/interface/coreapi.go
+3
@@ -34,6 +34,9 @@ type CoreAPI interface {
34
// Dht returns an implementation of Dht API
35
Dht() DhtAPI
36
37
+ // Swarm returns an implementation of Swarm API
38
+ Swarm() SwarmAPI
39
+
40
// ResolvePath resolves the path using Unixfs resolver
41
ResolvePath(context.Context, Path) (ResolvedPath, error)
42
core/coreapi/interface/swarm.go
+3
-3
@@ -1,9 +1,9 @@
1
package iface
2
3
import (
4
+ "context"
5
"time"
6
6
- "context"
7
ma "gx/ipfs/QmWWQ2Txc2c6tqjsBpzg5Ar652cHPGNsQQp2SejkNmkUMb/go-multiaddr"
8
peer "gx/ipfs/QmZoWKhxUmZ2seW4BzX6fJkNR8hh9PsGModr7q171yq2SS/go-libp2p-peer"
9
)
@@ -17,11 +17,11 @@ type PeerInfo interface {
17
Address() ma.Multiaddr
18
19
// Latency returns last known round trip time to the peer
20
- Latency() time.Duration
20
+ Latency(context.Context) (time.Duration, error)
21
22
// Streams returns list of streams established with the peer
23
// TODO: should this return multicodecs?
24
- Streams() []string
24
+ Streams(context.Context) ([]string, error)
25
}
26
27
// SwarmAPI specifies the interface to libp2p swarm
core/coreapi/swarm.go
new
+147
@@ -0,0 +1,147 @@
1
+package coreapi
2
+
3
+import (
4
+ "context"
5
+ "errors"
6
+ "fmt"
7
+ "time"
8
+
9
+ coreiface "github.com/ipfs/go-ipfs/core/coreapi/interface"
10
+
11
+ iaddr "gx/ipfs/QmQViVWBHbU6HmYjXcdNq7tVASCNgdg64ZGcauuDkLCivW/go-ipfs-addr"
12
+ swarm "gx/ipfs/QmSwZMWwFZSUpe5muU2xgTUwppH24KfMwdPXiwbEp2c6G5/go-libp2p-swarm"
13
+ ma "gx/ipfs/QmWWQ2Txc2c6tqjsBpzg5Ar652cHPGNsQQp2SejkNmkUMb/go-multiaddr"
14
+ pstore "gx/ipfs/QmXauCuJzmzapetmC6W4TuDJLL1yFFrVzSHoWv8YdbmnxH/go-libp2p-peerstore"
15
+ net "gx/ipfs/QmXfkENeeBvh3zYA51MaSdGUdBjhQ99cP5WQe8zgr6wchG/go-libp2p-net"
16
+ peer "gx/ipfs/QmZoWKhxUmZ2seW4BzX6fJkNR8hh9PsGModr7q171yq2SS/go-libp2p-peer"
17
+)
18
+
19
+type SwarmAPI struct {
20
+ *CoreAPI
21
+}
22
+
23
+type connInfo struct {
24
+ api *CoreAPI
25
+ conn net.Conn
26
+
27
+ addr ma.Multiaddr
28
+ peer peer.ID
29
+ muxer string
30
+}
31
+
32
+func (api *SwarmAPI) Connect(ctx context.Context, addr ma.Multiaddr) error {
33
+ if api.node.PeerHost == nil {
34
+ return coreiface.ErrOffline
35
+ }
36
+
37
+ snet, ok := api.node.PeerHost.Network().(*swarm.Network)
38
+ if !ok {
39
+ return fmt.Errorf("peerhost network was not swarm")
40
+ }
41
+
42
+ swrm := snet.Swarm()
43
+
44
+ ia, err := iaddr.ParseMultiaddr(ma.Multiaddr(addr))
45
+ if err != nil {
46
+ return err
47
+ }
48
+
49
+ pi := pstore.PeerInfo{
50
+ ID: ia.ID(),
51
+ Addrs: []ma.Multiaddr{ia.Transport()},
52
+ }
53
+
54
+ swrm.Backoff().Clear(pi.ID)
55
+
56
+ return api.node.PeerHost.Connect(ctx, pi)
57
+}
58
+
59
+func (api *SwarmAPI) Disconnect(ctx context.Context, addr ma.Multiaddr) error {
60
+ if api.node.PeerHost == nil {
61
+ return coreiface.ErrOffline
62
+ }
63
+
64
+ ia, err := iaddr.ParseMultiaddr(ma.Multiaddr(addr))
65
+ if err != nil {
66
+ return err
67
+ }
68
+
69
+ taddr := ia.Transport()
70
+
71
+ found := false
72
+ conns := api.node.PeerHost.Network().ConnsToPeer(ia.ID())
73
+ for _, conn := range conns {
74
+ if !conn.RemoteMultiaddr().Equal(taddr) {
75
+ continue
76
+ }
77
+
78
+ if err := conn.Close(); err != nil {
79
+ return err
80
+ }
81
+ found = true
82
+ break
83
+ }
84
+
85
+ if !found {
86
+ return errors.New("conn not found")
87
+ }
88
+
89
+ return nil
90
+}
91
+
92
+func (api *SwarmAPI) Peers(context.Context) ([]coreiface.PeerInfo, error) {
93
+ if api.node.PeerHost == nil {
94
+ return nil, coreiface.ErrOffline
95
+ }
96
+
97
+ conns := api.node.PeerHost.Network().Conns()
98
+
99
+ var out []coreiface.PeerInfo
100
+ for _, c := range conns {
101
+ pid := c.RemotePeer()
102
+ addr := c.RemoteMultiaddr()
103
+
104
+ ci := &connInfo{
105
+ api: api.CoreAPI,
106
+ conn: c,
107
+
108
+ addr: addr,
109
+ peer: pid,
110
+ }
111
+
112
+ swcon, ok := c.(*swarm.Conn)
113
+ if ok {
114
+ ci.muxer = fmt.Sprintf("%T", swcon.StreamConn().Conn())
115
+ }
116
+
117
+ out = append(out, ci)
118
+ }
119
+
120
+ return out, nil
121
+}
122
+
123
+func (ci *connInfo) ID() peer.ID {
124
+ return ci.ID()
125
+}
126
+
127
+func (ci *connInfo) Address() ma.Multiaddr {
128
+ return ci.addr
129
+}
130
+
131
+func (ci *connInfo) Latency(context.Context) (time.Duration, error) {
132
+ return ci.api.node.Peerstore.LatencyEWMA(peer.ID(ci.ID())), nil
133
+}
134
+
135
+func (ci *connInfo) Streams(context.Context) ([]string, error) {
136
+ streams, err := ci.conn.GetStreams()
137
+ if err != nil {
138
+ return nil, err
139
+ }
140
+
141
+ out := make([]string, len(streams))
142
+ for i, s := range streams {
143
+ out[i] = string(s.Protocol())
144
+ }
145
+
146
+ return out, nil
147
+}