diff --git a/routing/dht/dht.go b/routing/dht/dht.go index a1f63f106b0..0f55ba45b0e 100644 --- a/routing/dht/dht.go +++ b/routing/dht/dht.go @@ -3,9 +3,12 @@ package dht import ( "sync" - peer "github.com/jbenet/go-ipfs/peer" - swarm "github.com/jbenet/go-ipfs/swarm" - u "github.com/jbenet/go-ipfs/util" + peer "github.com/jbenet/go-ipfs/peer" + swarm "github.com/jbenet/go-ipfs/swarm" + u "github.com/jbenet/go-ipfs/util" + + ds "github.com/jbenet/datastore.go" + "code.google.com/p/goprotobuf/proto" ) @@ -18,8 +21,11 @@ type IpfsDHT struct { network *swarm.Swarm - // local data (TEMPORARY: until we formalize data storage with datastore) - data map[string][]byte + // Local peer (yourself) + self *peer.Peer + + // Local data + datastore ds.Datastore // map of channels waiting for reply messages listeners map[uint64]chan *swarm.Message @@ -29,6 +35,15 @@ type IpfsDHT struct { shutdown chan struct{} } +func NewDHT(p *peer.Peer) *IpfsDHT { + dht := new(IpfsDHT) + dht.self = p + dht.network = swarm.NewSwarm(p) + dht.listeners = make(map[uint64]chan *swarm.Message) + dht.shutdown = make(chan struct{}) + return dht +} + // Read in all messages from swarm and handle them appropriately // NOTE: this function is just a quick sketch func (dht *IpfsDHT) handleMessages() { @@ -61,10 +76,13 @@ func (dht *IpfsDHT) handleMessages() { case DHTMessage_GET_VALUE: dht.handleGetValue(mes.Peer, pmes) case DHTMessage_PUT_VALUE: + dht.handlePutValue(mes.Peer, pmes) case DHTMessage_FIND_NODE: + dht.handleFindNode(mes.Peer, pmes) case DHTMessage_ADD_PROVIDER: case DHTMessage_GET_PROVIDERS: case DHTMessage_PING: + dht.handleFindNode(mes.Peer, pmes) } case <-dht.shutdown: @@ -74,15 +92,22 @@ func (dht *IpfsDHT) handleMessages() { } func (dht *IpfsDHT) handleGetValue(p *peer.Peer, pmes *DHTMessage) { - val, found := dht.data[pmes.GetKey()] - if found { + dskey := ds.NewKey(pmes.GetKey()) + i_val, err := dht.datastore.Get(dskey) + if err == nil { isResponse := true resp := new(DHTMessage) resp.Response = &isResponse resp.Id = pmes.Id resp.Key = pmes.Key + + val := i_val.([]byte) resp.Value = val - } else { + + mes := new(swarm.Message) + mes.Peer = p + mes.Data = []byte(resp.String()) + } else if err == ds.ErrNotFound { // Find closest node(s) to desired key and reply with that info // TODO: this will need some other metadata in the protobuf message // to signal to the querying node that the data its receiving @@ -90,8 +115,14 @@ func (dht *IpfsDHT) handleGetValue(p *peer.Peer, pmes *DHTMessage) { } } +// Store a value in this nodes local storage func (dht *IpfsDHT) handlePutValue(p *peer.Peer, pmes *DHTMessage) { - panic("Not implemented.") + dskey := ds.NewKey(pmes.GetKey()) + err := dht.datastore.Put(dskey, pmes.GetValue()) + if err != nil { + // For now, just panic, handle this better later maybe + panic(err) + } } func (dht *IpfsDHT) handleFindNode(p *peer.Peer, pmes *DHTMessage) { @@ -121,6 +152,17 @@ func (dht *IpfsDHT) ListenFor(mesid uint64) <-chan *swarm.Message { return lchan } +func (dht *IpfsDHT) Unlisten(mesid uint64) { + dht.listenLock.Lock() + ch, ok := dht.listeners[mesid] + if ok { + delete(dht.listeners, mesid) + } + dht.listenLock.Unlock() + close(ch) +} + + // Stop all communications from this node and shut down func (dht *IpfsDHT) Halt() { dht.shutdown <- struct{}{}