a little more work on message handling stuff
Jeromy Johnson committed
Jul 30, 2014 at 17:46 UTC
87739b3af5bb9a59db757fcca44324afe1841b3b
4 files changed
+89
-20
routing/dht/dht.go
+65
-10
@@ -1,10 +1,12 @@
1
package dht
2
3
import (
4
+ "sync"
5
+
6
+ peer "github.com/jbenet/go-ipfs/peer"
7
swarm "github.com/jbenet/go-ipfs/swarm"
8
u "github.com/jbenet/go-ipfs/util"
9
"code.google.com/p/goprotobuf/proto"
7
- "sync"
10
)
11
12
// TODO. SEE https://github.com/jbenet/node-ipfs/blob/master/submodules/ipfs-dht/index.js
@@ -16,6 +18,9 @@ type IpfsDHT struct {
18
19
network *swarm.Swarm
20
21
+ // local data (TEMPORARY: until we formalize data storage with datastore)
22
+ data map[string][]byte
23
+
24
// map of channels waiting for reply messages
25
listeners map[uint64]chan *swarm.Message
26
listenLock sync.RWMutex
@@ -38,22 +43,28 @@ func (dht *IpfsDHT) handleMessages() {
43
}
44
45
// Note: not sure if this is the correct place for this
41
- dht.listenLock.RLock()
42
- ch, ok := dht.listeners[pmes.GetId()]
43
- dht.listenLock.RUnlock()
44
- if ok {
45
- ch <- mes
46
+ if pmes.GetResponse() {
47
+ dht.listenLock.RLock()
48
+ ch, ok := dht.listeners[pmes.GetId()]
49
+ dht.listenLock.RUnlock()
50
+ if ok {
51
+ ch <- mes
52
+ }
53
+
54
+ // this is expected behaviour during a timeout
55
+ u.DOut("Received response with nobody listening...")
56
+ continue
57
}
58
//
59
49
- // Do something else with the messages?
60
switch pmes.GetType() {
51
- case DHTMessage_ADD_PROVIDER:
61
+ case DHTMessage_GET_VALUE:
62
+ dht.handleGetValue(mes.Peer, pmes)
63
+ case DHTMessage_PUT_VALUE:
64
case DHTMessage_FIND_NODE:
65
+ case DHTMessage_ADD_PROVIDER:
66
case DHTMessage_GET_PROVIDERS:
54
- case DHTMessage_GET_VALUE:
67
case DHTMessage_PING:
56
- case DHTMessage_PUT_VALUE:
68
}
69
70
case <-dht.shutdown:
@@ -62,6 +73,44 @@ func (dht *IpfsDHT) handleMessages() {
73
}
74
}
75
76
+func (dht *IpfsDHT) handleGetValue(p *peer.Peer, pmes *DHTMessage) {
77
+ val, found := dht.data[pmes.GetKey()]
78
+ if found {
79
+ isResponse := true
80
+ resp := new(DHTMessage)
81
+ resp.Response = &isResponse
82
+ resp.Id = pmes.Id
83
+ resp.Key = pmes.Key
84
+ resp.Value = val
85
+ } else {
86
+ // Find closest node(s) to desired key and reply with that info
87
+ // TODO: this will need some other metadata in the protobuf message
88
+ // to signal to the querying node that the data its receiving
89
+ // is actually a list of other nodes
90
+ }
91
+}
92
+
93
+func (dht *IpfsDHT) handlePutValue(p *peer.Peer, pmes *DHTMessage) {
94
+ panic("Not implemented.")
95
+}
96
+
97
+func (dht *IpfsDHT) handleFindNode(p *peer.Peer, pmes *DHTMessage) {
98
+ panic("Not implemented.")
99
+}
100
+
101
+func (dht *IpfsDHT) handlePing(p *peer.Peer, pmes *DHTMessage) {
102
+ isResponse := true
103
+ resp := new(DHTMessage)
104
+ resp.Id = pmes.Id
105
+ resp.Response = &isResponse
106
+
107
+ mes := new(swarm.Message)
108
+ mes.Peer = p
109
+ mes.Data = []byte(resp.String())
110
+ dht.network.Chan.Outgoing <- mes
111
+}
112
+
113
+
114
// Register a handler for a specific message ID, used for getting replies
115
// to certain messages (i.e. response to a GET_VALUE message)
116
func (dht *IpfsDHT) ListenFor(mesid uint64) <-chan *swarm.Message {
@@ -71,3 +120,9 @@ func (dht *IpfsDHT) ListenFor(mesid uint64) <-chan *swarm.Message {
120
dht.listenLock.Unlock()
121
return lchan
122
}
123
+
124
+// Stop all communications from this node and shut down
125
+func (dht *IpfsDHT) Halt() {
126
+ dht.shutdown <- struct{}{}
127
+ dht.network.Close()
128
+}
routing/dht/messages.pb.go
+15
-5
@@ -66,11 +66,14 @@ func (x *DHTMessage_MessageType) UnmarshalJSON(data []byte) error {
66
}
67
68
type DHTMessage struct {
69
- Type *DHTMessage_MessageType `protobuf:"varint,1,req,name=type,enum=dht.DHTMessage_MessageType" json:"type,omitempty"`
70
- Key *string `protobuf:"bytes,2,opt,name=key" json:"key,omitempty"`
71
- Value []byte `protobuf:"bytes,3,opt,name=value" json:"value,omitempty"`
72
- Id *uint64 `protobuf:"varint,4,req,name=id" json:"id,omitempty"`
73
- XXX_unrecognized []byte `json:"-"`
69
+ Type *DHTMessage_MessageType `protobuf:"varint,1,req,name=type,enum=dht.DHTMessage_MessageType" json:"type,omitempty"`
70
+ Key *string `protobuf:"bytes,2,opt,name=key" json:"key,omitempty"`
71
+ Value []byte `protobuf:"bytes,3,opt,name=value" json:"value,omitempty"`
72
+ // Unique ID of this message, used to match queries with responses
73
+ Id *uint64 `protobuf:"varint,4,req,name=id" json:"id,omitempty"`
74
+ // Signals whether or not this message is a response to another message
75
+ Response *bool `protobuf:"varint,5,opt,name=response" json:"response,omitempty"`
76
+ XXX_unrecognized []byte `json:"-"`
77
}
78
79
func (m *DHTMessage) Reset() { *m = DHTMessage{} }
@@ -105,6 +108,13 @@ func (m *DHTMessage) GetId() uint64 {
108
return 0
109
}
110
111
+func (m *DHTMessage) GetResponse() bool {
112
+ if m != nil && m.Response != nil {
113
+ return *m.Response
114
+ }
115
+ return false
116
+}
117
+
118
func init() {
119
proto.RegisterEnum("dht.DHTMessage_MessageType", DHTMessage_MessageType_name, DHTMessage_MessageType_value)
120
}
routing/dht/messages.proto
+5
@@ -15,5 +15,10 @@ message DHTMessage {
15
required MessageType type = 1;
16
optional string key = 2;
17
optional bytes value = 3;
18
+
19
+ // Unique ID of this message, used to match queries with responses
20
required uint64 id = 4;
21
+
22
+ // Signals whether or not this message is a response to another message
23
+ optional bool response = 5;
24
}
routing/dht/routing.go
+4
-5
@@ -1,15 +1,17 @@
1
package dht
2
3
import (
4
+ "math/rand"
5
+ "time"
6
+
7
peer "github.com/jbenet/go-ipfs/peer"
8
swarm "github.com/jbenet/go-ipfs/swarm"
9
u "github.com/jbenet/go-ipfs/util"
7
- "time"
10
)
11
12
// TODO: determine a way of creating and managing message IDs
13
func GenerateMessageID() uint64 {
12
- return 4
14
+ return uint64(rand.Uint32()) << 32 & uint64(rand.Uint32())
15
}
16
17
// This file implements the Routing interface for the IpfsDHT struct.
@@ -68,9 +70,6 @@ func (s *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
70
case resp := <-response_chan:
71
return resp.Data, nil
72
}
71
-
72
- // Should never be hit
73
- return nil, nil
73
}
74
75
// Value provider layer of indirection.