add put and get dht commands to cli
Jeromy committed
Feb 21, 2015 at 16:20 UTC
7c0c3c4511898e638fcd7e91e6e4f5fe037006ed
5 files changed
+242
-2
core/commands/dht.go
+223
@@ -26,6 +26,8 @@ var DhtCmd = &cmds.Command{
26
"query": queryDhtCmd,
27
"findprovs": findProvidersDhtCmd,
28
"findpeer": findPeerDhtCmd,
29
+ "get": getValueDhtCmd,
30
+ "put": putValueDhtCmd,
31
},
32
}
33
@@ -352,3 +354,224 @@ var findPeerDhtCmd = &cmds.Command{
354
},
355
Type: notif.QueryEvent{},
356
}
357
+
358
+var getValueDhtCmd = &cmds.Command{
359
+ Helptext: cmds.HelpText{
360
+ Tagline: "Run a 'GetValue' query through the DHT",
361
+ ShortDescription: `
362
+GetValue will return the value stored in the dht at the given key.
363
+`,
364
+ },
365
+
366
+ Arguments: []cmds.Argument{
367
+ cmds.StringArg("key", true, true, "The key to find a value for"),
368
+ },
369
+ Options: []cmds.Option{
370
+ cmds.BoolOption("verbose", "v", "Write extra information"),
371
+ },
372
+ Run: func(req cmds.Request, res cmds.Response) {
373
+ n, err := req.Context().GetNode()
374
+ if err != nil {
375
+ res.SetError(err, cmds.ErrNormal)
376
+ return
377
+ }
378
+
379
+ dht, ok := n.Routing.(*ipdht.IpfsDHT)
380
+ if !ok {
381
+ res.SetError(ErrNotDHT, cmds.ErrNormal)
382
+ return
383
+ }
384
+
385
+ outChan := make(chan interface{})
386
+ res.SetOutput((<-chan interface{})(outChan))
387
+
388
+ events := make(chan *notif.QueryEvent)
389
+ ctx := notif.RegisterForQueryEvents(req.Context().Context, events)
390
+
391
+ go func() {
392
+ defer close(outChan)
393
+ for e := range events {
394
+ outChan <- e
395
+ }
396
+ }()
397
+
398
+ go func() {
399
+ defer close(events)
400
+ val, err := dht.GetValue(ctx, u.B58KeyDecode(req.Arguments()[0]))
401
+ if err != nil {
402
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
403
+ Type: notif.QueryError,
404
+ Extra: err.Error(),
405
+ })
406
+ } else {
407
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
408
+ Type: notif.Value,
409
+ Extra: string(val),
410
+ })
411
+ }
412
+ }()
413
+ },
414
+ Marshalers: cmds.MarshalerMap{
415
+ cmds.Text: func(res cmds.Response) (io.Reader, error) {
416
+ outChan, ok := res.Output().(<-chan interface{})
417
+ if !ok {
418
+ return nil, u.ErrCast()
419
+ }
420
+
421
+ verbose, _, _ := res.Request().Option("v").Bool()
422
+
423
+ marshal := func(v interface{}) (io.Reader, error) {
424
+ obj, ok := v.(*notif.QueryEvent)
425
+ if !ok {
426
+ return nil, u.ErrCast()
427
+ }
428
+
429
+ buf := new(bytes.Buffer)
430
+ if verbose {
431
+ fmt.Fprintf(buf, "%s: ", time.Now().Format("15:04:05.000"))
432
+ }
433
+ switch obj.Type {
434
+ case notif.FinalPeer:
435
+ if verbose {
436
+ fmt.Fprintf(buf, "* closest peer %s\n", obj.ID)
437
+ }
438
+ case notif.PeerResponse:
439
+ if verbose {
440
+ fmt.Fprintf(buf, "* %s says use ", obj.ID)
441
+ for _, p := range obj.Responses {
442
+ fmt.Fprintf(buf, "%s ", p.ID)
443
+ }
444
+ fmt.Fprintln(buf)
445
+ }
446
+ case notif.SendingQuery:
447
+ if verbose {
448
+ fmt.Fprintf(buf, "* querying %s\n", obj.ID)
449
+ }
450
+ case notif.Value:
451
+ fmt.Fprintf(buf, "got value: '%s'\n", obj.Extra)
452
+ case notif.QueryError:
453
+ fmt.Fprintf(buf, "error: %s\n", obj.Extra)
454
+ default:
455
+ fmt.Fprintf(buf, "unrecognized event type: %d\n", obj.Type)
456
+ }
457
+ return buf, nil
458
+ }
459
+
460
+ return &cmds.ChannelMarshaler{
461
+ Channel: outChan,
462
+ Marshaler: marshal,
463
+ }, nil
464
+ },
465
+ },
466
+ Type: notif.QueryEvent{},
467
+}
468
+
469
+var putValueDhtCmd = &cmds.Command{
470
+ Helptext: cmds.HelpText{
471
+ Tagline: "Run a 'PutValue' query through the DHT",
472
+ ShortDescription: `
473
+PutValue will store the given key value pair in the dht.
474
+`,
475
+ },
476
+
477
+ Arguments: []cmds.Argument{
478
+ cmds.StringArg("key", true, false, "The key to store the value at"),
479
+ cmds.StringArg("value", true, false, "The value to store").EnableStdin(),
480
+ },
481
+ Options: []cmds.Option{
482
+ cmds.BoolOption("verbose", "v", "Write extra information"),
483
+ },
484
+ Run: func(req cmds.Request, res cmds.Response) {
485
+ n, err := req.Context().GetNode()
486
+ if err != nil {
487
+ res.SetError(err, cmds.ErrNormal)
488
+ return
489
+ }
490
+
491
+ dht, ok := n.Routing.(*ipdht.IpfsDHT)
492
+ if !ok {
493
+ res.SetError(ErrNotDHT, cmds.ErrNormal)
494
+ return
495
+ }
496
+
497
+ outChan := make(chan interface{})
498
+ res.SetOutput((<-chan interface{})(outChan))
499
+
500
+ events := make(chan *notif.QueryEvent)
501
+ ctx := notif.RegisterForQueryEvents(req.Context().Context, events)
502
+
503
+ key := u.B58KeyDecode(req.Arguments()[0])
504
+ data := req.Arguments()[1]
505
+
506
+ go func() {
507
+ defer close(outChan)
508
+ for e := range events {
509
+ outChan <- e
510
+ }
511
+ }()
512
+
513
+ go func() {
514
+ defer close(events)
515
+ err := dht.PutValue(ctx, key, []byte(data))
516
+ if err != nil {
517
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
518
+ Type: notif.QueryError,
519
+ Extra: err.Error(),
520
+ })
521
+ }
522
+ }()
523
+ },
524
+ Marshalers: cmds.MarshalerMap{
525
+ cmds.Text: func(res cmds.Response) (io.Reader, error) {
526
+ outChan, ok := res.Output().(<-chan interface{})
527
+ if !ok {
528
+ return nil, u.ErrCast()
529
+ }
530
+
531
+ verbose, _, _ := res.Request().Option("v").Bool()
532
+
533
+ marshal := func(v interface{}) (io.Reader, error) {
534
+ obj, ok := v.(*notif.QueryEvent)
535
+ if !ok {
536
+ return nil, u.ErrCast()
537
+ }
538
+
539
+ buf := new(bytes.Buffer)
540
+ if verbose {
541
+ fmt.Fprintf(buf, "%s: ", time.Now().Format("15:04:05.000"))
542
+ }
543
+ switch obj.Type {
544
+ case notif.FinalPeer:
545
+ if verbose {
546
+ fmt.Fprintf(buf, "* closest peer %s\n", obj.ID)
547
+ }
548
+ case notif.PeerResponse:
549
+ if verbose {
550
+ fmt.Fprintf(buf, "* %s says use ", obj.ID)
551
+ for _, p := range obj.Responses {
552
+ fmt.Fprintf(buf, "%s ", p.ID)
553
+ }
554
+ fmt.Fprintln(buf)
555
+ }
556
+ case notif.SendingQuery:
557
+ if verbose {
558
+ fmt.Fprintf(buf, "* querying %s\n", obj.ID)
559
+ }
560
+ case notif.QueryError:
561
+ fmt.Fprintf(buf, "error: %s\n", obj.Extra)
562
+ case notif.Value:
563
+ fmt.Fprintf(buf, "storing value at %s\n", obj.ID)
564
+ default:
565
+ fmt.Fprintf(buf, "unrecognized event type: %d\n", obj.Type)
566
+ }
567
+ return buf, nil
568
+ }
569
+
570
+ return &cmds.ChannelMarshaler{
571
+ Channel: outChan,
572
+ Marshaler: marshal,
573
+ }, nil
574
+ },
575
+ },
576
+ Type: notif.QueryEvent{},
577
+}
notifications/query.go
+1
@@ -17,6 +17,7 @@ const (
17
FinalPeer
18
QueryError
19
Provider
20
+ Value
21
)
22
23
type QueryEvent struct {
routing/dht/dht.go
+1
-1
@@ -322,7 +322,7 @@ func (dht *IpfsDHT) betterPeersToQuery(pmes *pb.Message, p peer.ID, count int) [
322
// == to self? thats bad
323
for _, p := range closer {
324
if p == dht.self {
325
- log.Debug("Attempted to return self! this shouldnt happen...")
325
+ log.Error("Attempted to return self! this shouldnt happen...")
326
return nil
327
}
328
}
routing/dht/handlers.go
+1
-1
@@ -83,7 +83,7 @@ func (dht *IpfsDHT) handleGetValue(ctx context.Context, p peer.ID, pmes *pb.Mess
83
84
// Find closest peer on given cluster to desired key and reply with that info
85
closer := dht.betterPeersToQuery(pmes, p, CloserPeerCount)
86
- if closer != nil {
86
+ if len(closer) > 0 {
87
closerinfos := peer.PeerInfos(dht.peerstore, closer)
88
for _, pi := range closerinfos {
89
log.Debugf("handleGetValue returning closer peer: '%s'", pi.ID)
routing/dht/routing.go
+16
@@ -59,6 +59,11 @@ func (dht *IpfsDHT) PutValue(ctx context.Context, key u.Key, value []byte) error
59
wg.Add(1)
60
go func(p peer.ID) {
61
defer wg.Done()
62
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
63
+ Type: notif.Value,
64
+ ID: p,
65
+ })
66
+
67
err := dht.putValueToPeer(ctx, p, key, rec)
68
if err != nil {
69
log.Debugf("failed putting value to peer: %s", err)
@@ -92,6 +97,11 @@ func (dht *IpfsDHT) GetValue(ctx context.Context, key u.Key) ([]byte, error) {
97
98
// setup the Query
99
query := dht.newQuery(key, func(ctx context.Context, p peer.ID) (*dhtQueryResult, error) {
100
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
101
+ Type: notif.SendingQuery,
102
+ ID: p,
103
+ })
104
+
105
val, peers, err := dht.getValueOrPeers(ctx, p, key)
106
if err != nil {
107
return nil, err
@@ -102,6 +112,12 @@ func (dht *IpfsDHT) GetValue(ctx context.Context, key u.Key) ([]byte, error) {
112
res.success = true
113
}
114
115
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
116
+ Type: notif.PeerResponse,
117
+ ID: p,
118
+ Responses: pointerizePeerInfos(peers),
119
+ })
120
+
121
return res, nil
122
})
123