NetStore exported and constructed in BZZ protocol definition.

bzz/hive.go

@@ -10,6 +10,12 @@ type hive struct {
 	pool map[string]peer
 }
 
+func newHive() *hive {
+	return &hive{
+		pool: make(map[string]peer),
+	}
+}
+
 func (self *hive) addPeer(p peer) {
 	self.pool[string(p.pubkey)] = p
 }

bzz/netstore.go

@@ -6,7 +6,7 @@ import (
 	"time"
 )
 
-type netStore struct {
+type NetStore struct {
 	localStore *localStore
 	lock       sync.Mutex
 	hive       *hive
@@ -42,7 +42,18 @@ type requestStatus struct {
 	C          chan bool
 }
 
-func (self *netStore) Put(entry *Chunk) {
+func NewNetStore(path string) *NetStore {
+	dbStore, _ := newDbStore(path)
+	return &NetStore{
+		localStore: &localStore{
+			memStore: newMemStore(dbStore),
+			dbStore:  dbStore,
+		},
+		hive: newHive(),
+	}
+}
+
+func (self *NetStore) Put(entry *Chunk) {
 	chunk, err := self.localStore.Get(entry.Key)
 	if err != nil {
 		chunk = entry
@@ -55,7 +66,7 @@ func (self *netStore) Put(entry *Chunk) {
 	self.put(chunk)
 }
 
-func (self *netStore) put(entry *Chunk) {
+func (self *NetStore) put(entry *Chunk) {
 	self.localStore.Put(entry)
 	self.store(entry)
 	// only send responses once
@@ -66,7 +77,7 @@ func (self *netStore) put(entry *Chunk) {
 	}
 }
 
-func (self *netStore) addStoreRequest(req *storeRequestMsgData) {
+func (self *NetStore) addStoreRequest(req *storeRequestMsgData) {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	chunk, err := self.localStore.Get(req.Key)
@@ -87,7 +98,7 @@ func (self *netStore) addStoreRequest(req *storeRequestMsgData) {
 }
 
 // waits for response or times  out
-func (self *netStore) Get(key Key) (chunk *Chunk, err error) {
+func (self *NetStore) Get(key Key) (chunk *Chunk, err error) {
 	chunk = self.get(key)
 	id := generateId()
 	timeout := time.Now().Add(searchTimeout)
@@ -103,7 +114,7 @@ func (self *netStore) Get(key Key) (chunk *Chunk, err error) {
 	return
 }
 
-func (self *netStore) get(key Key) (chunk *Chunk) {
+func (self *NetStore) get(key Key) (chunk *Chunk) {
 	var err error
 	chunk, err = self.localStore.Get(key)
 	// we assume that a returned chunk is the one stored in the memory cache
@@ -121,7 +132,7 @@ func (self *netStore) get(key Key) (chunk *Chunk) {
 	return
 }
 
-func (self *netStore) addRetrieveRequest(req *retrieveRequestMsgData) {
+func (self *NetStore) addRetrieveRequest(req *retrieveRequestMsgData) {
 
 	self.lock.Lock()
 	defer self.lock.Unlock()
@@ -143,7 +154,7 @@ func (self *netStore) addRetrieveRequest(req *retrieveRequestMsgData) {
 }
 
 // it's assumed that caller holds the lock
-func (self *netStore) startSearch(chunk *Chunk, id int64, timeout time.Time) {
+func (self *NetStore) startSearch(chunk *Chunk, id int64, timeout time.Time) {
 	chunk.req.status = reqSearching
 	peers := self.hive.getPeers(chunk.Key)
 	req := &retrieveRequestMsgData{
@@ -166,7 +177,7 @@ adds a new peer to an existing open request
 only add if less than requesterCount peers forwarded the same request id so far
 note this is done irrespective of status (searching or found/timedOut)
 */
-func (self *netStore) addRequester(rs *requestStatus, req *retrieveRequestMsgData) {
+func (self *NetStore) addRequester(rs *requestStatus, req *retrieveRequestMsgData) {
 	list := rs.requesters[req.Id]
 	rs.requesters[req.Id] = append(list, req)
 }
@@ -183,7 +194,7 @@ this is the most simplistic implementation:
  - respond with peers and timeout if still searching
 ! in the last case as well, we should respond with reject if already got requesterCount peers with that exact id
 */
-func (self *netStore) strategyUpdateRequest(rs *requestStatus, req *retrieveRequestMsgData) (msgTyp int, timeout *time.Time) {
+func (self *NetStore) strategyUpdateRequest(rs *requestStatus, req *retrieveRequestMsgData) (msgTyp int, timeout *time.Time) {
 
 	switch rs.status {
 	case reqSearching:
@@ -198,7 +209,7 @@ func (self *netStore) strategyUpdateRequest(rs *requestStatus, req *retrieveRequ
 
 }
 
-func (self *netStore) propagateResponse(chunk *Chunk) {
+func (self *NetStore) propagateResponse(chunk *Chunk) {
 	for id, requesters := range chunk.req.requesters {
 		counter := requesterCount
 		msg := &storeRequestMsgData{
@@ -219,7 +230,7 @@ func (self *netStore) propagateResponse(chunk *Chunk) {
 	}
 }
 
-func (self *netStore) deliver(req *retrieveRequestMsgData, chunk *Chunk) {
+func (self *NetStore) deliver(req *retrieveRequestMsgData, chunk *Chunk) {
 	storeReq := &storeRequestMsgData{
 		Key:            req.Key,
 		Id:             req.Id,
@@ -232,7 +243,7 @@ func (self *netStore) deliver(req *retrieveRequestMsgData, chunk *Chunk) {
 	req.peer.store(storeReq)
 }
 
-func (self *netStore) store(chunk *Chunk) {
+func (self *NetStore) store(chunk *Chunk) {
 	id := generateId()
 	req := &storeRequestMsgData{
 		Key:  chunk.Key,
@@ -245,7 +256,7 @@ func (self *netStore) store(chunk *Chunk) {
 	}
 }
 
-func (self *netStore) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout time.Time) {
+func (self *NetStore) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout time.Time) {
 	peersData := &peersMsgData{
 		Peers:   []*peerAddr{}, // get proximity bin from cademlia routing table
 		Key:     req.Key,
@@ -255,7 +266,7 @@ func (self *netStore) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout t
 	req.peer.peers(peersData)
 }
 
-func (self *netStore) searchTimeout(rs *requestStatus, req *retrieveRequestMsgData) (timeout *time.Time) {
+func (self *NetStore) searchTimeout(rs *requestStatus, req *retrieveRequestMsgData) (timeout *time.Time) {
 	t := time.Now().Add(searchTimeout)
 	if req.Timeout.Before(t) {
 		return &req.Timeout

bzz/protocol.go

@@ -32,8 +32,7 @@ const (
 // bzzProtocol represents the swarm wire protocol
 // instance is running on each peer
 type bzzProtocol struct {
-	netStore *netStore
+	netStore *NetStore
-	hive     *hive
 	peer     *p2p.Peer
 	rw       p2p.MsgReadWriter
 }
@@ -141,23 +140,22 @@ main entrypoint, wrappers starting a server running the bzz protocol
 use this constructor to attach the protocol ("class") to server caps
 the Dev p2p layer then runs the protocol instance on each peer
 */
-func BzzProtocol(netStore *netStore, hive *hive) p2p.Protocol {
+func BzzProtocol(netStore *NetStore) p2p.Protocol {
 	return p2p.Protocol{
 		Name:    "bzz",
 		Version: Version,
 		Length:  ProtocolLength,
 		Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
-			return runBzzProtocol(netStore, hive, p, rw)
+			return runBzzProtocol(netStore, p, rw)
 		},
 	}
 }
 
 // the main loop that handles incoming messages
 // note RemovePeer in the post-disconnect hook
-func runBzzProtocol(netStore *netStore, hive *hive, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
+func runBzzProtocol(netStore *NetStore, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
 	self := &bzzProtocol{
 		netStore: netStore,
-		hive:     hive,
 		rw:       rw,
 		peer:     p,
 	}
@@ -166,7 +164,7 @@ func runBzzProtocol(netStore *netStore, hive *hive, p *p2p.Peer, rw p2p.MsgReadW
 		for {
 			err = self.handle()
 			if err != nil {
-				self.hive.removePeer(peer{bzzProtocol: self})
+				self.netStore.hive.removePeer(peer{bzzProtocol: self})
 				break
 			}
 		}
@@ -216,7 +214,7 @@ func (self *bzzProtocol) handle() error {
 			return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
 		}
 		req.peer = peer{bzzProtocol: self}
-		self.hive.addPeers(&req)
+		self.netStore.hive.addPeers(&req)
 
 	default:
 		return self.protoError(ErrInvalidMsgCode, "%v", msg.Code)
@@ -269,7 +267,7 @@ func (self *bzzProtocol) handleStatus() (err error) {
 
 	self.peer.Infof("Peer is [bzz] capable (%d/%d)\n", status.Version, status.NetworkId)
 
-	self.hive.addPeer(peer{bzzProtocol: self, pubkey: status.NodeID})
+	self.netStore.hive.addPeer(peer{bzzProtocol: self, pubkey: status.NodeID})
 
 	return nil
 }

eth/backend.go

@@ -138,7 +138,7 @@ func New(config *Config) (*Ethereum, error) {
 	protocols := []p2p.Protocol{
 		ethProto,
 		eth.whisper.Protocol(),
-		bzz.BzzProtocol(nil, nil),
+		bzz.BzzProtocol(bzz.NewNetStore(config.DataDir + "/bzz")),
 	}
 
 	nat, err := p2p.ParseNAT(config.NATType, config.PMPGateway)