respect verbose option a bit, and show query events for other commands
Jeromy committed
Jan 24, 2015 at 00:59 UTC
d68efad06c874352b22e661c7822ee723ad0871a
4 files changed
+137
-32
core/commands/dht.go
+122
-31
@@ -14,6 +14,8 @@ import (
14
u "github.com/jbenet/go-ipfs/util"
15
)
16
17
+var ErrNotDHT = errors.New("routing service is not a DHT")
18
+
19
var DhtCmd = &cmds.Command{
20
Helptext: cmds.HelpText{
21
Tagline: "Issue commands directly through the DHT",
@@ -87,19 +89,27 @@ var queryDhtCmd = &cmds.Command{
89
return nil, u.ErrCast()
90
}
91
92
+ verbose, _, _ := res.Request().Option("v").Bool()
93
+
94
buf := new(bytes.Buffer)
91
- fmt.Fprintf(buf, "%s: ", time.Now().Format("15:04:05.000"))
95
+ if verbose {
96
+ fmt.Fprintf(buf, "%s: ", time.Now().Format("15:04:05.000"))
97
+ }
98
switch obj.Type {
99
case notif.FinalPeer:
100
fmt.Fprintf(buf, "%s\n", obj.ID)
101
case notif.PeerResponse:
96
- fmt.Fprintf(buf, "* %s says use ", obj.ID)
97
- for _, p := range obj.Responses {
98
- fmt.Fprintf(buf, "%s ", p.ID)
102
+ if verbose {
103
+ fmt.Fprintf(buf, "* %s says use ", obj.ID)
104
+ for _, p := range obj.Responses {
105
+ fmt.Fprintf(buf, "%s ", p.ID)
106
+ }
107
+ fmt.Fprintln(buf)
108
}
100
- fmt.Fprintln(buf)
109
case notif.SendingQuery:
102
- fmt.Fprintf(buf, "* querying %s\n", obj.ID)
110
+ if verbose {
111
+ fmt.Fprintf(buf, "* querying %s\n", obj.ID)
112
+ }
113
case notif.QueryError:
114
fmt.Fprintf(buf, "error: %s\n", obj.Extra)
115
default:
@@ -139,19 +149,31 @@ FindProviders will return a list of peers who are able to provide the value requ
149
150
dht, ok := n.Routing.(*ipdht.IpfsDHT)
151
if !ok {
142
- return nil, errors.New("Routing service was not a dht")
152
+ return nil, ErrNotDHT
153
}
154
155
numProviders := 20
156
157
outChan := make(chan interface{})
148
- pchan := dht.FindProvidersAsync(req.Context().Context, u.B58KeyDecode(req.Arguments()[0]), numProviders)
158
+ events := make(chan *notif.QueryEvent)
159
+ ctx := notif.RegisterForQueryEvents(req.Context().Context, events)
160
161
+ pchan := dht.FindProvidersAsync(ctx, u.B58KeyDecode(req.Arguments()[0]), numProviders)
162
go func() {
163
defer close(outChan)
164
+ for e := range events {
165
+ outChan <- e
166
+ }
167
+ }()
168
+
169
+ go func() {
170
+ defer close(events)
171
for p := range pchan {
172
np := p
154
- outChan <- &np
173
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
174
+ Type: notif.Provider,
175
+ Responses: []*peer.PeerInfo{&np},
176
+ })
177
}
178
}()
179
return outChan, nil
@@ -163,22 +185,41 @@ FindProviders will return a list of peers who are able to provide the value requ
185
return nil, u.ErrCast()
186
}
187
188
+ verbose, _, _ := res.Request().Option("v").Bool()
189
+
190
marshal := func(v interface{}) (io.Reader, error) {
167
- obj, ok := v.(*peer.PeerInfo)
191
+ obj, ok := v.(*notif.QueryEvent)
192
if !ok {
193
return nil, u.ErrCast()
194
}
195
172
- verbose, _, err := res.Request().Option("v").Bool()
173
- if err != nil {
174
- return nil, err
175
- }
176
-
196
buf := new(bytes.Buffer)
197
if verbose {
198
+ fmt.Fprintf(buf, "%s: ", time.Now().Format("15:04:05.000"))
199
+ }
200
+ switch obj.Type {
201
+ case notif.FinalPeer:
202
+ if verbose {
203
+ fmt.Fprintf(buf, "* closest peer %s\n", obj.ID)
204
+ }
205
+ case notif.Provider:
206
fmt.Fprintf(buf, "%s\n", obj.ID.Pretty())
180
- } else {
181
- fmt.Fprintf(buf, "%s\n", obj.ID)
207
+ case notif.PeerResponse:
208
+ if verbose {
209
+ fmt.Fprintf(buf, "* %s says use ", obj.ID)
210
+ for _, p := range obj.Responses {
211
+ fmt.Fprintf(buf, "%s ", p.ID)
212
+ }
213
+ fmt.Fprintln(buf)
214
+ }
215
+ case notif.SendingQuery:
216
+ if verbose {
217
+ fmt.Fprintf(buf, "* querying %s\n", obj.ID)
218
+ }
219
+ case notif.QueryError:
220
+ fmt.Fprintf(buf, "error: %s\n", obj.Extra)
221
+ default:
222
+ fmt.Fprintf(buf, "unrecognized event type: %d\n", obj.Type)
223
}
224
return buf, nil
225
}
@@ -209,7 +250,7 @@ var findPeerDhtCmd = &cmds.Command{
250
251
dht, ok := n.Routing.(*ipdht.IpfsDHT)
252
if !ok {
212
- return nil, errors.New("Routing service was not a dht")
253
+ return nil, ErrNotDHT
254
}
255
256
pid, err := peer.IDB58Decode(req.Arguments()[0])
@@ -217,29 +258,79 @@ var findPeerDhtCmd = &cmds.Command{
258
return nil, err
259
}
260
220
- pi, err := dht.FindPeer(req.Context().Context, pid)
221
- if err != nil {
222
- return nil, err
223
- }
261
+ outChan := make(chan interface{})
262
+ events := make(chan *notif.QueryEvent)
263
+ ctx := notif.RegisterForQueryEvents(req.Context().Context, events)
264
225
- return &pi, nil
265
+ go func() {
266
+ defer close(outChan)
267
+ for v := range events {
268
+ outChan <- v
269
+ }
270
+ }()
271
+
272
+ go func() {
273
+ defer close(events)
274
+ pi, err := dht.FindPeer(ctx, pid)
275
+ if err != nil {
276
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
277
+ Type: notif.QueryError,
278
+ Extra: err.Error(),
279
+ })
280
+ return
281
+ }
282
+
283
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
284
+ Type: notif.FinalPeer,
285
+ Responses: []*peer.PeerInfo{&pi},
286
+ })
287
+ }()
288
+
289
+ return outChan, nil
290
},
291
Marshalers: cmds.MarshalerMap{
292
cmds.Text: func(res cmds.Response) (io.Reader, error) {
229
- pinfo, ok := res.Output().(*peer.PeerInfo)
293
+ outChan, ok := res.Output().(<-chan interface{})
294
if !ok {
295
return nil, u.ErrCast()
296
}
297
234
- buf := new(bytes.Buffer)
235
- fmt.Fprintf(buf, "found peer: %s\n", pinfo.ID)
236
- fmt.Fprintf(buf, "reported addresses:\n")
237
- for _, addr := range pinfo.Addrs {
238
- fmt.Fprintf(buf, "\t%s\n", addr)
298
+ marshal := func(v interface{}) (io.Reader, error) {
299
+ obj, ok := v.(*notif.QueryEvent)
300
+ if !ok {
301
+ return nil, u.ErrCast()
302
+ }
303
+
304
+ buf := new(bytes.Buffer)
305
+ fmt.Fprintf(buf, "%s: ", time.Now().Format("15:04:05.000"))
306
+ switch obj.Type {
307
+ case notif.FinalPeer:
308
+ pi := obj.Responses[0]
309
+ fmt.Fprintf(buf, "%s\n", pi.ID)
310
+ for _, a := range pi.Addrs {
311
+ fmt.Fprintf(buf, "\t%s\n", a)
312
+ }
313
+ case notif.PeerResponse:
314
+ fmt.Fprintf(buf, "* %s says use ", obj.ID)
315
+ for _, p := range obj.Responses {
316
+ fmt.Fprintf(buf, "%s ", p.ID)
317
+ }
318
+ fmt.Fprintln(buf)
319
+ case notif.SendingQuery:
320
+ fmt.Fprintf(buf, "* querying %s\n", obj.ID)
321
+ case notif.QueryError:
322
+ fmt.Fprintf(buf, "error: %s\n", obj.Extra)
323
+ default:
324
+ fmt.Fprintf(buf, "unrecognized event type: %d\n", obj.Type)
325
+ }
326
+ return buf, nil
327
}
328
241
- return buf, nil
329
+ return &cmds.ChannelMarshaler{
330
+ Channel: outChan,
331
+ Marshaler: marshal,
332
+ }, nil
333
},
334
},
244
- Type: peer.PeerInfo{},
335
+ Type: notif.QueryEvent{},
336
}
notifications/query.go
+1
@@ -16,6 +16,7 @@ const (
16
PeerResponse
17
FinalPeer
18
QueryError
19
+ Provider
20
)
21
22
type QueryEvent struct {
routing/dht/lookup.go
-1
@@ -60,7 +60,6 @@ func (dht *IpfsDHT) GetClosestPeers(ctx context.Context, key u.Key) (<-chan peer
60
if kb.Closer(clp, dht.self, key) && peerset.TryAdd(clp) {
61
select {
62
case out <- clp:
63
- log.Error("Sending out peer: %s", clp.Pretty())
63
case <-ctx.Done():
64
return nil, ctx.Err()
65
}
routing/dht/routing.go
+14
@@ -7,6 +7,7 @@ import (
7
8
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
9
10
+ notif "github.com/jbenet/go-ipfs/notifications"
11
inet "github.com/jbenet/go-ipfs/p2p/net"
12
peer "github.com/jbenet/go-ipfs/p2p/peer"
13
"github.com/jbenet/go-ipfs/routing"
@@ -242,6 +243,10 @@ func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key u.Key, co
243
_, err := query.Run(ctx, peers)
244
if err != nil {
245
log.Errorf("Query error: %s", err)
246
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
247
+ Type: notif.QueryError,
248
+ Extra: err.Error(),
249
+ })
250
}
251
}
252
@@ -269,6 +274,10 @@ func (dht *IpfsDHT) FindPeer(ctx context.Context, id peer.ID) (peer.PeerInfo, er
274
275
// setup the Query
276
query := dht.newQuery(u.Key(id), func(ctx context.Context, p peer.ID) (*dhtQueryResult, error) {
277
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
278
+ Type: notif.SendingQuery,
279
+ ID: p,
280
+ })
281
282
pmes, err := dht.findPeerSingle(ctx, p, id)
283
if err != nil {
@@ -288,6 +297,11 @@ func (dht *IpfsDHT) FindPeer(ctx context.Context, id peer.ID) (peer.PeerInfo, er
297
}
298
}
299
300
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
301
+ Type: notif.PeerResponse,
302
+ Responses: pointerizePeerInfos(clpeerInfos),
303
+ })
304
+
305
return &dhtQueryResult{closerPeers: clpeerInfos}, nil
306
})
307