use a notification type strategy for the query events
Jeromy committed
Jan 22, 2015 at 18:18 UTC
528eb5a55a5068af0bda8381f0363ecf3ef51df9
4 files changed
+97
-67
core/commands/dht.go
+13
-10
@@ -8,6 +8,7 @@ import (
8
"time"
9
10
cmds "github.com/jbenet/go-ipfs/commands"
11
+ notif "github.com/jbenet/go-ipfs/notifications"
12
ipdht "github.com/jbenet/go-ipfs/routing/dht"
13
u "github.com/jbenet/go-ipfs/util"
14
)
@@ -46,16 +47,18 @@ var queryDhtCmd = &cmds.Command{
47
return nil, errors.New("Routing service was not a dht")
48
}
49
49
- events := make(chan *ipdht.QueryEvent)
50
- closestPeers, err := dht.GetClosestPeers(req.Context().Context, u.Key(req.Arguments()[0]), events)
50
+ events := make(chan *notif.QueryEvent)
51
+ ctx := notif.RegisterForQueryEvents(req.Context().Context, events)
52
+
53
+ closestPeers, err := dht.GetClosestPeers(ctx, u.Key(req.Arguments()[0]))
54
55
go func() {
56
defer close(events)
57
for p := range closestPeers {
55
- events <- &ipdht.QueryEvent{
58
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
59
ID: p,
57
- Type: ipdht.FinalPeer,
58
- }
60
+ Type: notif.FinalPeer,
61
+ })
62
}
63
}()
64
@@ -76,7 +79,7 @@ var queryDhtCmd = &cmds.Command{
79
}
80
81
marshal := func(v interface{}) (io.Reader, error) {
79
- obj, ok := v.(*ipdht.QueryEvent)
82
+ obj, ok := v.(*notif.QueryEvent)
83
if !ok {
84
return nil, u.ErrCast()
85
}
@@ -84,15 +87,15 @@ var queryDhtCmd = &cmds.Command{
87
buf := new(bytes.Buffer)
88
fmt.Fprintf(buf, "%s: ", time.Now().Format("15:04:05.000"))
89
switch obj.Type {
87
- case ipdht.FinalPeer:
90
+ case notif.FinalPeer:
91
fmt.Fprintf(buf, "%s\n", obj.ID)
89
- case ipdht.PeerResponse:
92
+ case notif.PeerResponse:
93
fmt.Fprintf(buf, "* %s says use ", obj.ID)
94
for _, p := range obj.Responses {
95
fmt.Fprintf(buf, "%s ", p.ID)
96
}
97
fmt.Fprintln(buf)
95
- case ipdht.SendingQuery:
98
+ case notif.SendingQuery:
99
fmt.Fprintf(buf, "* querying %s\n", obj.ID)
100
}
101
return buf, nil
@@ -104,5 +107,5 @@ var queryDhtCmd = &cmds.Command{
107
}, nil
108
},
109
},
107
- Type: ipdht.QueryEvent{},
110
+ Type: notif.QueryEvent{},
111
}
notifications/query.go
new
+73
@@ -0,0 +1,73 @@
1
+package notifications
2
+
3
+import (
4
+ "encoding/json"
5
+
6
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
7
+ peer "github.com/jbenet/go-ipfs/p2p/peer"
8
+)
9
+
10
+const RoutingQueryKey = "RoutingQueryEvent"
11
+
12
+type QueryEventType int
13
+
14
+const (
15
+ SendingQuery QueryEventType = iota
16
+ PeerResponse
17
+ FinalPeer
18
+)
19
+
20
+type QueryEvent struct {
21
+ ID peer.ID
22
+ Type QueryEventType
23
+ Responses []*peer.PeerInfo
24
+}
25
+
26
+func RegisterForQueryEvents(ctx context.Context, ch chan<- *QueryEvent) context.Context {
27
+ return context.WithValue(ctx, RoutingQueryKey, ch)
28
+}
29
+
30
+func PublishQueryEvent(ctx context.Context, ev *QueryEvent) {
31
+ ich := ctx.Value(RoutingQueryKey)
32
+ if ich == nil {
33
+ return
34
+ }
35
+
36
+ ch, ok := ich.(chan<- *QueryEvent)
37
+ if !ok {
38
+ return
39
+ }
40
+
41
+ select {
42
+ case ch <- ev:
43
+ case <-ctx.Done():
44
+ }
45
+}
46
+
47
+func (qe *QueryEvent) MarshalJSON() ([]byte, error) {
48
+ out := make(map[string]interface{})
49
+ out["ID"] = peer.IDB58Encode(qe.ID)
50
+ out["Type"] = int(qe.Type)
51
+ out["Responses"] = qe.Responses
52
+ return json.Marshal(out)
53
+}
54
+
55
+func (qe *QueryEvent) UnmarshalJSON(b []byte) error {
56
+ temp := struct {
57
+ ID string
58
+ Type int
59
+ Responses []*peer.PeerInfo
60
+ }{}
61
+ err := json.Unmarshal(b, &temp)
62
+ if err != nil {
63
+ return err
64
+ }
65
+ pid, err := peer.IDB58Decode(temp.ID)
66
+ if err != nil {
67
+ return err
68
+ }
69
+ qe.ID = pid
70
+ qe.Type = QueryEventType(temp.Type)
71
+ qe.Responses = temp.Responses
72
+ return nil
73
+}
routing/dht/lookup.go
+9
-55
@@ -1,10 +1,9 @@
1
package dht
2
3
import (
4
- "encoding/json"
5
-
4
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
5
6
+ notif "github.com/jbenet/go-ipfs/notifications"
7
peer "github.com/jbenet/go-ipfs/p2p/peer"
8
kb "github.com/jbenet/go-ipfs/routing/kbucket"
9
u "github.com/jbenet/go-ipfs/util"
@@ -12,20 +11,7 @@ import (
11
pset "github.com/jbenet/go-ipfs/util/peerset"
12
)
13
15
-type QueryEventType int
16
-
17
-const (
18
- SendingQuery QueryEventType = iota
19
- PeerResponse
20
- FinalPeer
21
-)
22
-
23
-type QueryEvent struct {
24
- ID peer.ID
25
- Type QueryEventType
26
- Responses []*peer.PeerInfo
27
-}
28
-
14
+// Required in order for proper JSON marshaling
15
func pointerizePeerInfos(pis []peer.PeerInfo) []*peer.PeerInfo {
16
out := make([]*peer.PeerInfo, len(pis))
17
for i, p := range pis {
@@ -37,7 +23,7 @@ func pointerizePeerInfos(pis []peer.PeerInfo) []*peer.PeerInfo {
23
24
// Kademlia 'node lookup' operation. Returns a channel of the K closest peers
25
// to the given key
40
-func (dht *IpfsDHT) GetClosestPeers(ctx context.Context, key u.Key, events chan<- *QueryEvent) (<-chan peer.ID, error) {
26
+func (dht *IpfsDHT) GetClosestPeers(ctx context.Context, key u.Key) (<-chan peer.ID, error) {
27
e := log.EventBegin(ctx, "getClosestPeers", &key)
28
tablepeers := dht.routingTable.NearestPeers(kb.ConvertKey(key), AlphaValue)
29
if len(tablepeers) == 0 {
@@ -58,12 +44,10 @@ func (dht *IpfsDHT) GetClosestPeers(ctx context.Context, key u.Key, events chan<
44
45
query := dht.newQuery(key, func(ctx context.Context, p peer.ID) (*dhtQueryResult, error) {
46
// For DHT query command
61
- select {
62
- case events <- &QueryEvent{
63
- Type: SendingQuery,
47
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
48
+ Type: notif.SendingQuery,
49
ID: p,
65
- }:
66
- }
50
+ })
51
52
closer, err := dht.closerPeersSingle(ctx, key, p)
53
if err != nil {
@@ -86,13 +70,11 @@ func (dht *IpfsDHT) GetClosestPeers(ctx context.Context, key u.Key, events chan<
70
log.Errorf("filtered: %v", filtered)
71
72
// For DHT query command
89
- select {
90
- case events <- &QueryEvent{
91
- Type: PeerResponse,
73
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
74
+ Type: notif.PeerResponse,
75
ID: p,
76
Responses: pointerizePeerInfos(filtered),
94
- }:
95
- }
77
+ })
78
79
return &dhtQueryResult{closerPeers: filtered}, nil
80
})
@@ -126,31 +108,3 @@ func (dht *IpfsDHT) closerPeersSingle(ctx context.Context, key u.Key, p peer.ID)
108
}
109
return out, nil
110
}
129
-
130
-func (qe *QueryEvent) MarshalJSON() ([]byte, error) {
131
- out := make(map[string]interface{})
132
- out["ID"] = peer.IDB58Encode(qe.ID)
133
- out["Type"] = int(qe.Type)
134
- out["Responses"] = qe.Responses
135
- return json.Marshal(out)
136
-}
137
-
138
-func (qe *QueryEvent) UnmarshalJSON(b []byte) error {
139
- temp := struct {
140
- ID string
141
- Type int
142
- Responses []*peer.PeerInfo
143
- }{}
144
- err := json.Unmarshal(b, &temp)
145
- if err != nil {
146
- return err
147
- }
148
- pid, err := peer.IDB58Decode(temp.ID)
149
- if err != nil {
150
- return err
151
- }
152
- qe.ID = pid
153
- qe.Type = QueryEventType(temp.Type)
154
- qe.Responses = temp.Responses
155
- return nil
156
-}
routing/dht/routing.go
+2
-2
@@ -48,7 +48,7 @@ func (dht *IpfsDHT) PutValue(ctx context.Context, key u.Key, value []byte) error
48
return err
49
}
50
51
- pchan, err := dht.GetClosestPeers(ctx, key, nil)
51
+ pchan, err := dht.GetClosestPeers(ctx, key)
52
if err != nil {
53
return err
54
}
@@ -134,7 +134,7 @@ func (dht *IpfsDHT) Provide(ctx context.Context, key u.Key) error {
134
// add self locally
135
dht.providers.AddProvider(key, dht.self)
136
137
- peers, err := dht.GetClosestPeers(ctx, key, nil)
137
+ peers, err := dht.GetClosestPeers(ctx, key)
138
if err != nil {
139
return err
140
}