diff --git a/bzz/hive.go b/bzz/hive.go index be684f7b59..46962d4e72 100644 --- a/bzz/hive.go +++ b/bzz/hive.go @@ -16,6 +16,26 @@ import ( "time" ) +// This is a mock implementation with a fixed peer pool with no distinction between peers +type peerPool struct { + pool map[string]peer +} + +func (self *peerPool) addPeer(p peer) { + self.pool[p.peer.identity.Pubkey()] = p +} + +func (self *peerPool) removePeer(p peer) { + delete(self.pool, p.peer.identity.Pubkey) +} + +func (self *peerPool) getPeers(target Key) (peers []peer) { + for key, value := range self.pool { + peers = append(peers, value) + } + return +} + type Hive struct { dpa *DPA memstore *memStore @@ -104,10 +124,42 @@ func (self *Hive) addStoreRequest(req *storeRequestMsgData) (err error) { self.lock.Lock() defer self.lock.Unlock() - // TODO: + chunk, err := self.dpa.Get(req.Key) + // we assume that a returned chunk is the one stored in the memory cache + if err != nil { + s := new(storeRequestStatus) + chunk = &Chunk{ + Key: req.Key, + Data: req.Data, + Size: req.Size, + storeRequestStatus: s, + } + self.dpa.Put(chunk) + self.store(chunk) + } else { + // pending retrieval request + if chunk.Data != nil { + // update access counts not needed, Get takes care of it + return + } + chunk.Data = req.Data + chunk.Size = req.Size + // FIXME: breach of memstore contract data is put into storage without checking capacity + self.dpa.Put(chunk) + // only send responses once + if chunk.req.status == reqSearching { + chunk.req.status = reqFound + self.propagateResponse(chunk) + } + } + return } +func (self *Hive) propagateResponse(chunk *Chunk) { + // send chunk to first requesterCount peer of each Id +} + func (self *Hive) addRetrieveRequest(req *retrieveRequestMsgData) { self.lock.Lock() @@ -154,6 +206,19 @@ func (self *Hive) deliver(req *retrieveRequestMsgData, chunk *Chunk) { req.peer.store(storeReq) } +func (self *Hive) store(chunk) { + r := rand.New(rand.NewSource(time.Now().UnixNano())) + req := storeRequestMsgData{ + Key: chunk.Key, + Data: chunk.Data, + Id: r.Int63(), + Size: chunk.Size, + } + for _, peer := range self.peerPool.GetPeers(chunk.Key) { + go peer.store(req) + } +} + func (self *Hive) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout time.Time) { peersData := &peersMsgData{ Peers: []*peerAddr{}, // get proximity bin from cademlia routing table diff --git a/bzz/netstore.go b/bzz/netstore.go deleted file mode 100644 index 9dec6e673f..0000000000 --- a/bzz/netstore.go +++ /dev/null @@ -1,55 +0,0 @@ -package bzz - -/* -DHT implements the chunk store that directly communicates with the bzz protocol on the one hand and the kademlia node table on the other. -It does forwarding for incoming requests and handles expiry/timeout. -*/ - -import ( - "math/rand" - "time" -) - -// This is a mock implementation with a fixed peer pool with no distinction between peers -type peerPool struct { - pool map[string]peer -} - -func (self *peerPool) addPeer(p peer) { - self.pool[p.peer.identity.Pubkey()] = p -} - -func (self *peerPool) removePeer(p peer) { - delete(self.pool, p.peer.identity.Pubkey) -} - -func (self *peerPool) GetPeers(target Key) (peers []peer) { - for key, value := range self.pool { - peers = append(peers, value) - } - return -} - -// it implements the ChunkStore interface -type netStore struct { - peerPool peerPool - // cademlia -} - -func (self *netStore) Put(chunk *Chunk) { - r := rand.New(rand.NewSource(time.Now().UnixNano())) - req := storeRequestMsgData{ - Key: chunk.Key, - Data: chunk.Data, - Id: r.Int63(), - Size: chunk.Size, - } - for _, peer := range self.peerPool.GetPeers(chunk.Key) { - go peer.store(req) - } - return -} - -func (self *DPA) Get(key Key) (chunk *Chunk, err error) { - return -}