handleGetValue
Juan Batiz-Benet committed
Sep 16, 2014 at 00:57 UTC
2f2808e9d8f618d05244ac8fc49bfbdaebff68c1
1 file changed
+77
-54
routing/dht/dht.go
+77
-54
@@ -185,75 +185,98 @@ func (dht *IpfsDHT) handlerForMsgType(t Message_MessageType) dhtHandler {
185
}
186
187
func (dht *IpfsDHT) putValueToNetwork(p *peer.Peer, key string, value []byte) error {
188
- pmes := Message{
189
- Type: PBDHTMessage_PUT_VALUE,
190
- Key: key,
188
+ typ := Message_PUT_VALUE
189
+ pmes := &Message{
190
+ Type: &typ,
191
+ Key: &key,
192
Value: value,
192
- ID: swarm.GenerateMessageID(),
193
}
194
195
- mes := swarm.NewMessage(p, pmes.ToProtobuf())
196
- dht.netChan.Outgoing <- mes
197
- return nil
195
+ mes, err := msg.FromObject(p, pmes)
196
+ if err != nil {
197
+ return err
198
+ }
199
+ return dht.sender.SendMessage(context.TODO(), mes)
200
}
201
200
-func (dht *IpfsDHT) handleGetValue(p *peer.Peer, pmes *PBDHTMessage) {
202
+func (dht *IpfsDHT) handleGetValue(p *peer.Peer, pmes *Message) (*Message, error) {
203
u.DOut("handleGetValue for key: %s\n", pmes.GetKey())
202
- dskey := ds.NewKey(pmes.GetKey())
204
+
205
+ // setup response
206
resp := &Message{
204
- Response: true,
205
- ID: pmes.GetId(),
206
- Key: pmes.GetKey(),
207
+ Type: pmes.Type,
208
+ Key: pmes.Key,
209
+ }
210
+
211
+ // first, is the key even a key?
212
+ key := pmes.GetKey()
213
+ if key == "" {
214
+ return nil, errors.New("handleGetValue but no key was provided.")
215
}
216
+
217
+ // let's first check if we have the value locally.
218
+ dskey := ds.NewKey(pmes.GetKey())
219
iVal, err := dht.datastore.Get(dskey)
220
+
221
+ // if we got an unexpected error, bail.
222
+ if err != ds.ErrNotFound {
223
+ return nil, err
224
+ }
225
+
226
+ // if we have the value, respond with it!
227
if err == nil {
228
u.DOut("handleGetValue success!\n")
211
- resp.Success = true
212
- resp.Value = iVal.([]byte)
213
- } else if err == ds.ErrNotFound {
214
- // Check if we know any providers for the requested value
215
- provs := dht.providers.GetProviders(u.Key(pmes.GetKey()))
216
- if len(provs) > 0 {
217
- u.DOut("handleGetValue returning %d provider[s]\n", len(provs))
218
- resp.Peers = provs
219
- resp.Success = true
220
- } else {
221
- // No providers?
222
- // Find closest peer on given cluster to desired key and reply with that info
223
-
224
- level := 0
225
- if len(pmes.GetValue()) < 1 {
226
- // TODO: maybe return an error? Defaulting isnt a good idea IMO
227
- u.PErr("handleGetValue: no routing level specified, assuming 0\n")
228
- } else {
229
- level = int(pmes.GetValue()[0]) // Using value field to specify cluster level
230
- }
231
- u.DOut("handleGetValue searching level %d clusters\n", level)
232
-
233
- closer := dht.routingTables[level].NearestPeer(kb.ConvertKey(u.Key(pmes.GetKey())))
229
235
- if closer.ID.Equal(dht.self.ID) {
236
- u.DOut("Attempted to return self! this shouldnt happen...\n")
237
- resp.Peers = nil
238
- goto out
239
- }
240
- // If this peer is closer than the one from the table, return nil
241
- if kb.Closer(dht.self.ID, closer.ID, u.Key(pmes.GetKey())) {
242
- resp.Peers = nil
243
- u.DOut("handleGetValue could not find a closer node than myself.\n")
244
- } else {
245
- u.DOut("handleGetValue returning a closer peer: '%s'\n", closer.ID.Pretty())
246
- resp.Peers = []*peer.Peer{closer}
247
- }
230
+ byts, ok := iVal.([]byte)
231
+ if !ok {
232
+ return nil, fmt.Errorf("datastore had non byte-slice value for %v", dskey)
233
}
249
- } else {
250
- //temp: what other errors can a datastore return?
251
- panic(err)
234
+
235
+ resp.Value = byts
236
+ return resp, nil
237
}
238
254
-out:
255
- mes := swarm.NewMessage(p, resp.ToProtobuf())
256
- dht.netChan.Outgoing <- mes
239
+ // if we know any providers for the requested value, return those.
240
+ provs := dht.providers.GetProviders(u.Key(pmes.GetKey()))
241
+ if len(provs) > 0 {
242
+ u.DOut("handleGetValue returning %d provider[s]\n", len(provs))
243
+ resp.ProviderPeers = provs
244
+ return resp, nil
245
+ }
246
+
247
+ // Find closest peer on given cluster to desired key and reply with that info
248
+ // TODO: this should probably be decomposed.
249
+
250
+ // stored levels are > 1, to distinguish missing levels.
251
+ level := pmes.GetClusterLevel()
252
+ if level < 0 {
253
+ // TODO: maybe return an error? Defaulting isnt a good idea IMO
254
+ u.PErr("handleGetValue: no routing level specified, assuming 0\n")
255
+ level = 0
256
+ }
257
+ u.DOut("handleGetValue searching level %d clusters\n", level)
258
+
259
+ ck := kb.ConvertKey(u.Key(pmes.GetKey()))
260
+ closer := dht.routingTables[level].NearestPeer(ck)
261
+
262
+ // if closer peer is self, return nil
263
+ if closer.ID.Equal(dht.self.ID) {
264
+ u.DOut("Attempted to return self! this shouldnt happen...\n")
265
+ resp.CloserPeers = nil
266
+ return resp, nil
267
+ }
268
+
269
+ // if self is closer than the one from the table, return nil
270
+ if kb.Closer(dht.self.ID, closer.ID, u.Key(pmes.GetKey())) {
271
+ u.DOut("handleGetValue could not find a closer node than myself.\n")
272
+ resp.CloserPeers = nil
273
+ return resp, nil
274
+ }
275
+
276
+ // we got a closer peer, it seems. return it.
277
+ u.DOut("handleGetValue returning a closer peer: '%s'\n", closer.ID.Pretty())
278
+ resp.CloserPeers = []*peer.Peer{closer}
279
+ return resp, nil
280
}
281
282
// Store a value in this peer local storage