-
Notifications
You must be signed in to change notification settings - Fork 0
/
client.go
164 lines (147 loc) · 5.01 KB
/
client.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
package supernode
import (
"bytes"
"context"
"errors"
"time"
proxy "github.com/ipfs/go-ipfs/routing/supernode/proxy"
routing "gx/ipfs/QmNdaQ8itUU9jEZUwTsG4gHMaPmRfi6FEe89QjQAFbep3M/go-libp2p-routing"
dhtpb "gx/ipfs/QmRmroYSdievxnjiuy99C8BzShNstdEWcEF3LQHF7fUbez/go-libp2p-kad-dht/pb"
logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
"gx/ipfs/QmUywuGNZoUKV8B9iyvup9bPkLiMrhTsyVMkeSXW5VxAfC/go-libp2p-host"
loggables "gx/ipfs/QmVesPmqbPp7xRGyY96tnBwzDtVV1nqv4SCVxo5zCqKyH8/go-libp2p-loggables"
pb "gx/ipfs/QmWYCqr6UDqqD1bfRybaAPtbAqcN3TSJpveaBXMwbQ3ePZ/go-libp2p-record/pb"
pstore "gx/ipfs/QmXZSd1qR5BxZkPyuwfT5jpqQFScZccoZvDneXsKzCNHWX/go-libp2p-peerstore"
cid "gx/ipfs/QmYhQaCYEcaPPjxJX7YcPcVKkQfRy6sJ7B3XmGFk82XYdQ/go-cid"
proto "gx/ipfs/QmZ4Qi3GaRbjcx28Sme5eMH7RQjGkt8wHxt2a65oLaeFEV/gogo-protobuf/proto"
peer "gx/ipfs/QmdS9KpbDyPrieswibZhkod1oXqRwZJrUPzxCofAMWpFGq/go-libp2p-peer"
)
var log = logging.Logger("supernode")
type Client struct {
peerhost host.Host
peerstore pstore.Peerstore
proxy proxy.Proxy
local peer.ID
}
// TODO take in datastore/cache
func NewClient(px proxy.Proxy, h host.Host, ps pstore.Peerstore, local peer.ID) (*Client, error) {
return &Client{
proxy: px,
local: local,
peerstore: ps,
peerhost: h,
}, nil
}
func (c *Client) FindProvidersAsync(ctx context.Context, k *cid.Cid, max int) <-chan pstore.PeerInfo {
logging.ContextWithLoggable(ctx, loggables.Uuid("findProviders"))
defer log.EventBegin(ctx, "findProviders", k).Done()
ch := make(chan pstore.PeerInfo)
go func() {
defer close(ch)
request := dhtpb.NewMessage(dhtpb.Message_GET_PROVIDERS, k.KeyString(), 0)
response, err := c.proxy.SendRequest(ctx, request)
if err != nil {
log.Debug(err)
return
}
for _, p := range dhtpb.PBPeersToPeerInfos(response.GetProviderPeers()) {
select {
case <-ctx.Done():
log.Debug(ctx.Err())
return
case ch <- *p:
}
}
}()
return ch
}
func (c *Client) PutValue(ctx context.Context, k string, v []byte) error {
defer log.EventBegin(ctx, "putValue").Done()
r, err := makeRecord(c.peerstore, c.local, k, v)
if err != nil {
return err
}
pmes := dhtpb.NewMessage(dhtpb.Message_PUT_VALUE, string(k), 0)
pmes.Record = r
return c.proxy.SendMessage(ctx, pmes) // wrap to hide the remote
}
func (c *Client) GetValue(ctx context.Context, k string) ([]byte, error) {
defer log.EventBegin(ctx, "getValue").Done()
msg := dhtpb.NewMessage(dhtpb.Message_GET_VALUE, string(k), 0)
response, err := c.proxy.SendRequest(ctx, msg) // TODO wrap to hide the remote
if err != nil {
return nil, err
}
return response.Record.GetValue(), nil
}
func (c *Client) GetValues(ctx context.Context, k string, _ int) ([]routing.RecvdVal, error) {
defer log.EventBegin(ctx, "getValue").Done()
msg := dhtpb.NewMessage(dhtpb.Message_GET_VALUE, string(k), 0)
response, err := c.proxy.SendRequest(ctx, msg) // TODO wrap to hide the remote
if err != nil {
return nil, err
}
return []routing.RecvdVal{
{
Val: response.Record.GetValue(),
From: c.local,
},
}, nil
}
// Provide adds the given key 'k' to the content routing system. If 'brd' is
// true, it announces that content to the network. For the supernode client,
// setting 'brd' to false makes this call a no-op
func (c *Client) Provide(ctx context.Context, k *cid.Cid, brd bool) error {
if !brd {
return nil
}
defer log.EventBegin(ctx, "provide", k).Done()
msg := dhtpb.NewMessage(dhtpb.Message_ADD_PROVIDER, k.KeyString(), 0)
// FIXME how is connectedness defined for the local node
pri := []dhtpb.PeerRoutingInfo{
{
PeerInfo: pstore.PeerInfo{
ID: c.local,
Addrs: c.peerhost.Addrs(),
},
},
}
msg.ProviderPeers = dhtpb.PeerRoutingInfosToPBPeers(pri)
return c.proxy.SendMessage(ctx, msg) // TODO wrap to hide remote
}
func (c *Client) FindPeer(ctx context.Context, id peer.ID) (pstore.PeerInfo, error) {
defer log.EventBegin(ctx, "findPeer", id).Done()
request := dhtpb.NewMessage(dhtpb.Message_FIND_NODE, string(id), 0)
response, err := c.proxy.SendRequest(ctx, request) // hide remote
if err != nil {
return pstore.PeerInfo{}, err
}
for _, p := range dhtpb.PBPeersToPeerInfos(response.GetCloserPeers()) {
if p.ID == id {
return *p, nil
}
}
return pstore.PeerInfo{}, errors.New("could not find peer")
}
// creates and signs a record for the given key/value pair
func makeRecord(ps pstore.Peerstore, p peer.ID, k string, v []byte) (*pb.Record, error) {
blob := bytes.Join([][]byte{[]byte(k), v, []byte(p)}, []byte{})
sig, err := ps.PrivKey(p).Sign(blob)
if err != nil {
return nil, err
}
return &pb.Record{
Key: proto.String(string(k)),
Value: v,
Author: proto.String(string(p)),
Signature: sig,
}, nil
}
func (c *Client) Ping(ctx context.Context, id peer.ID) (time.Duration, error) {
defer log.EventBegin(ctx, "ping", id).Done()
return time.Nanosecond, errors.New("supernode routing does not support the ping method")
}
func (c *Client) Bootstrap(ctx context.Context) error {
return c.proxy.Bootstrap(ctx)
}
var _ routing.IpfsRouting = &Client{}