implement dht findprovs and add error output to dht query
Jeromy committed
Jan 22, 2015 at 22:18 UTC
56a5e72760ed3fca5684ec18cc9318a253d1e9e0
4 files changed
+100
-6
core/commands/dht.go
+82
-1
@@ -9,6 +9,7 @@ import (
9
10
cmds "github.com/jbenet/go-ipfs/commands"
11
notif "github.com/jbenet/go-ipfs/notifications"
12
+ peer "github.com/jbenet/go-ipfs/p2p/peer"
13
ipdht "github.com/jbenet/go-ipfs/routing/dht"
14
u "github.com/jbenet/go-ipfs/util"
15
)
@@ -20,7 +21,8 @@ var DhtCmd = &cmds.Command{
21
},
22
23
Subcommands: map[string]*cmds.Command{
23
- "query": queryDhtCmd,
24
+ "query": queryDhtCmd,
25
+ "findprovs": findProvidersDhtCmd,
26
},
27
}
28
@@ -97,6 +99,10 @@ var queryDhtCmd = &cmds.Command{
99
fmt.Fprintln(buf)
100
case notif.SendingQuery:
101
fmt.Fprintf(buf, "* querying %s\n", obj.ID)
102
+ case notif.QueryError:
103
+ fmt.Fprintf(buf, "error: %s\n", obj.Extra)
104
+ default:
105
+ fmt.Fprintf(buf, "unrecognized event type: %d\n", obj.Type)
106
}
107
return buf, nil
108
}
@@ -109,3 +115,78 @@ var queryDhtCmd = &cmds.Command{
115
},
116
Type: notif.QueryEvent{},
117
}
118
+
119
+var findProvidersDhtCmd = &cmds.Command{
120
+ Helptext: cmds.HelpText{
121
+ Tagline: "Run a 'FindProviders' query through the DHT",
122
+ ShortDescription: `
123
+FindProviders will return a list of peers who are able to provide the value requested.
124
+`,
125
+ },
126
+
127
+ Arguments: []cmds.Argument{
128
+ cmds.StringArg("key", true, true, "The key to find providers for"),
129
+ },
130
+ Options: []cmds.Option{
131
+ cmds.BoolOption("verbose", "v", "Write extra information"),
132
+ },
133
+ Run: func(req cmds.Request) (interface{}, error) {
134
+ n, err := req.Context().GetNode()
135
+ if err != nil {
136
+ return nil, err
137
+ }
138
+
139
+ dht, ok := n.Routing.(*ipdht.IpfsDHT)
140
+ if !ok {
141
+ return nil, errors.New("Routing service was not a dht")
142
+ }
143
+
144
+ numProviders := 20
145
+
146
+ outChan := make(chan interface{})
147
+ pchan := dht.FindProvidersAsync(req.Context().Context, u.B58KeyDecode(req.Arguments()[0]), numProviders)
148
+
149
+ go func() {
150
+ defer close(outChan)
151
+ for p := range pchan {
152
+ np := p
153
+ outChan <- &np
154
+ }
155
+ }()
156
+ return outChan, nil
157
+ },
158
+ Marshalers: cmds.MarshalerMap{
159
+ cmds.Text: func(res cmds.Response) (io.Reader, error) {
160
+ outChan, ok := res.Output().(<-chan interface{})
161
+ if !ok {
162
+ return nil, u.ErrCast()
163
+ }
164
+
165
+ marshal := func(v interface{}) (io.Reader, error) {
166
+ obj, ok := v.(*peer.PeerInfo)
167
+ if !ok {
168
+ return nil, u.ErrCast()
169
+ }
170
+
171
+ verbose, _, err := res.Request().Option("v").Bool()
172
+ if err != nil {
173
+ return nil, err
174
+ }
175
+
176
+ buf := new(bytes.Buffer)
177
+ if verbose {
178
+ fmt.Fprintf(buf, "%s\n", obj.ID.Pretty())
179
+ } else {
180
+ fmt.Fprintf(buf, "%s\n", obj.ID)
181
+ }
182
+ return buf, nil
183
+ }
184
+
185
+ return &cmds.ChannelMarshaler{
186
+ Channel: outChan,
187
+ Marshaler: marshal,
188
+ }, nil
189
+ },
190
+ },
191
+ Type: peer.PeerInfo{},
192
+}
notifications/query.go
+11
-4
@@ -15,12 +15,14 @@ const (
15
SendingQuery QueryEventType = iota
16
PeerResponse
17
FinalPeer
18
+ QueryError
19
)
20
21
type QueryEvent struct {
22
ID peer.ID
23
Type QueryEventType
24
Responses []*peer.PeerInfo
25
+ Extra string
26
}
27
28
func RegisterForQueryEvents(ctx context.Context, ch chan<- *QueryEvent) context.Context {
@@ -49,6 +51,7 @@ func (qe *QueryEvent) MarshalJSON() ([]byte, error) {
51
out["ID"] = peer.IDB58Encode(qe.ID)
52
out["Type"] = int(qe.Type)
53
out["Responses"] = qe.Responses
54
+ out["Extra"] = qe.Extra
55
return json.Marshal(out)
56
}
57
@@ -57,17 +60,21 @@ func (qe *QueryEvent) UnmarshalJSON(b []byte) error {
60
ID string
61
Type int
62
Responses []*peer.PeerInfo
63
+ Extra string
64
}{}
65
err := json.Unmarshal(b, &temp)
66
if err != nil {
67
return err
68
}
65
- pid, err := peer.IDB58Decode(temp.ID)
66
- if err != nil {
67
- return err
69
+ if len(temp.ID) > 0 {
70
+ pid, err := peer.IDB58Decode(temp.ID)
71
+ if err != nil {
72
+ return err
73
+ }
74
+ qe.ID = pid
75
}
69
- qe.ID = pid
76
qe.Type = QueryEventType(temp.Type)
77
qe.Responses = temp.Responses
78
+ qe.Extra = temp.Extra
79
return nil
80
}
routing/dht/lookup.go
-1
@@ -67,7 +67,6 @@ func (dht *IpfsDHT) GetClosestPeers(ctx context.Context, key u.Key) (<-chan peer
67
filtered = append(filtered, dht.peerstore.PeerInfo(clp))
68
}
69
}
70
- log.Errorf("filtered: %v", filtered)
70
71
// For DHT query command
72
notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
routing/dht/query.go
+7
@@ -3,6 +3,7 @@ package dht
3
import (
4
"sync"
5
6
+ notif "github.com/jbenet/go-ipfs/notifications"
7
peer "github.com/jbenet/go-ipfs/p2p/peer"
8
queue "github.com/jbenet/go-ipfs/p2p/peer/queue"
9
"github.com/jbenet/go-ipfs/routing"
@@ -230,6 +231,12 @@ func (r *dhtQueryRunner) queryPeer(cg ctxgroup.ContextGroup, p peer.ID) {
231
pi := peer.PeerInfo{ID: p}
232
if err := r.query.dht.host.Connect(cg.Context(), pi); err != nil {
233
log.Debugf("Error connecting: %s", err)
234
+
235
+ notif.PublishQueryEvent(cg.Context(), ¬if.QueryEvent{
236
+ Type: notif.QueryError,
237
+ Extra: err.Error(),
238
+ })
239
+
240
r.Lock()
241
r.errs = append(r.errs, err)
242
r.Unlock()