reorganise initialisation and start of bzz components so that self node is available (only after p2p.Server.Start()

bzz/dpa.go

@@ -3,8 +3,6 @@ package bzz
 import (
 	"errors"
 	"sync"
-	// "time"
-	// "fmt"
 
 	ethlogger "github.com/ethereum/go-ethereum/logger"
 )

bzz/hive.go

@@ -3,7 +3,6 @@ package bzz
 import (
 	// "fmt"
 
-	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/kademlia"
 )
 
@@ -29,15 +28,15 @@ type hive struct {
 	path string
 }
 
-func newHive(address common.Hash, hivepath string) *hive {
+func newHive(hivepath string) *hive {
 	return &hive{
 		path: hivepath,
-		kad:  kademlia.New(kademlia.Address(address)),
+		kad:  kademlia.New(),
 	}
 }
 
-func (self *hive) start() (err error) {
+func (self *hive) start(address kademlia.Address) (err error) {
-	self.kad.Start()
+	self.kad.Start(address)
 	err = self.kad.Load(self.path)
 	if err != nil {
 		dpaLogger.Warnf("Warning: error reading kademlia node db (skipping): %v", err)

bzz/netstore.go

@@ -6,13 +6,15 @@ import (
 	"sync"
 	"time"
 
-	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/kademlia"
+	"github.com/ethereum/go-ethereum/p2p/discover"
 )
 
 type NetStore struct {
 	localStore *localStore
 	lock       sync.Mutex
 	hive       *hive
+	self       *discover.Node
 }
 
 /*
@@ -45,19 +47,24 @@ type requestStatus struct {
 	C          chan bool
 }
 
-func NewNetStore(addr common.Hash, path, hivepath string) (netstore *NetStore, err error) {
+func NewNetStore(path, hivepath string) (netstore *NetStore, err error) {
 	dbStore, err := newDbStore(path)
 	if err != nil {
 		return
 	}
-	hive := newHive(addr, hivepath)
+	hive := newHive(hivepath)
 	netstore = &NetStore{
 		localStore: &localStore{
 			memStore: newMemStore(dbStore),
 			dbStore:  dbStore,
 		}, hive: hive,
 	}
-	err = hive.start()
+	return
+}
+
+func (self *NetStore) Start(node *discover.Node) (err error) {
+	self.self = node
+	err = self.hive.start(kademlia.Address(node.Sha()))
 	if err != nil {
 		return
 	}

bzz/protocol.go

@@ -58,7 +58,6 @@ var errorToString = map[int]string{
 // bzzProtocol represents the swarm wire protocol
 // instance is running on each peer
 type bzzProtocol struct {
-	self     func() *discover.Node
 	node     *discover.Node
 	netStore *NetStore
 	peer     *p2p.Peer
@@ -188,22 +187,21 @@ 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, self func() *discover.Node) 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, self, 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, selfF func() *discover.Node, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
+func runBzzProtocol(netStore *NetStore, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
 	self := &bzzProtocol{
-		self:     selfF,
 		netStore: netStore,
 		rw:       rw,
 		peer:     p,
@@ -284,12 +282,12 @@ func (self *bzzProtocol) handle() error {
 
 func (self *bzzProtocol) handleStatus() (err error) {
 	// send precanned status message
-	sliceNodeID := self.self().ID
+	sliceNodeID := self.netStore.self.ID
 	handshake := &statusMsgData{
 		Version:   uint64(Version),
 		ID:        "honey",
 		NodeID:    sliceNodeID[:],
-		Addr:      newPeerAddrFromNode(self.self()),
+		Addr:      newPeerAddrFromNode(self.netStore.self),
 		NetworkId: uint64(NetworkId),
 		Caps:      []p2p.Cap{},
 	}

common/kademlia/kademlia.go

@@ -77,12 +77,10 @@ type kadDB struct {
 	Nodes   [][]*NodeRecord `json:nodes`
 }
 
-// public constructor with compulsory arguments
+// public constructor
 // hash is a byte slice of length equal to self.HashBytes
-func New(a Address) *Kademlia {
+func New() *Kademlia {
-	return &Kademlia{
+	return &Kademlia{}
-		addr: a, // compulsory fields without default
-	}
 }
 
 // accessor for KAD self address
@@ -97,12 +95,13 @@ func (self *Kademlia) Count() int {
 
 // Start brings up a pool of entries potentially from an offline persisted source
 // and sets default values for optional parameters
-func (self *Kademlia) Start() error {
+func (self *Kademlia) Start(addr Address) error {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	if self.quitC != nil {
 		return nil
 	}
+	self.addr = addr
 	if self.MaxProx == 0 {
 		self.MaxProx = maxProx
 	}
@@ -454,6 +453,7 @@ bit first).
 proximity(x, y) counts the common zeros in the front of this distance measure.
 */
 func proximity(one, other Address) (ret int) {
+	fmt.Printf("%08b ^  %08b = %08b\n", one[0], other[0], one[0]^other[0])
 	for i := 0; i < len(one); i++ {
 		oxo := one[i] ^ other[i]
 		for j := 0; j < 8; j++ {

common/kademlia/kademlia_test.go

@@ -61,8 +61,8 @@ func TestAddNode(t *testing.T) {
 	if !ok {
 		t.Errorf("oops")
 	}
-	kad := New(addr)
+	kad := New()
-	kad.Start()
+	kad.Start(addr)
 	err := kad.AddNode(&testNode{addr: other})
 	_ = err
 }
@@ -73,9 +73,9 @@ func TestGetNodes(t *testing.T) {
 
 	test := func(test *getNodesTest) bool {
 		// for any node kad.le, Target and N
-		kad := New(test.Self)
+		kad := New()
 		kad.MaxProx = 10
-		kad.Start()
+		kad.Start(test.Self)
 		var err error
 		t.Logf("getNodesTest %v: %v\n", len(test.All), test)
 		for _, node := range test.All {
@@ -152,9 +152,9 @@ func TestProxAdjust(t *testing.T) {
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
 	self := gen(Address{}, r).(Address)
 
-	kad := New(self)
+	kad := New()
 	kad.MaxProx = 10
-	kad.Start()
+	kad.Start(self)
 	var err error
 	for i := 0; i < 100; i++ {
 		a := gen(Address{}, r).(Address)
@@ -186,7 +186,7 @@ func TestProxAdjust(t *testing.T) {
 func TestCallback(t *testing.T) {
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
 	self := gen(Address{}, r).(Address)
-	kad := New(self)
+	kad := New()
 	var bucket int
 	var called chan bool
 	kad.MaxProx = 5
@@ -195,7 +195,7 @@ func TestCallback(t *testing.T) {
 		bucket = b
 		close(called)
 	}
-	kad.Start()
+	kad.Start(self)
 	var err error
 
 	for i := 0; i < 100; i++ {
@@ -232,9 +232,9 @@ func TestSaveLoad(t *testing.T) {
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
 	addresses := gen([]Address{}, r).([]Address)
 	self := addresses[0]
-	kad := New(self)
+	kad := New()
 	kad.MaxProx = 10
-	kad.Start()
+	kad.Start(self)
 	var err error
 	for _, a := range addresses[1:] {
 		err = kad.AddNode(&testNode{addr: a})
@@ -246,8 +246,8 @@ func TestSaveLoad(t *testing.T) {
 	nodes := kad.GetNodes(self, 100)
 	path := "/tmp/bzz.peers"
 	kad.Stop(path)
-	kad = New(self)
+	kad = New()
-	kad.Start()
+	kad.Start(self)
 	kad.Load(path)
 	for _, b := range kad.DB() {
 		for _, node := range b {

eth/backend.go

@@ -184,6 +184,8 @@ type Ethereum struct {
 	pow             *ethash.Ethash
 	protocolManager *ProtocolManager
 	downloader      *downloader.Downloader
+	DPA             *bzz.DPA
+	netStore        *bzz.NetStore
 
 	net      *p2p.Server
 	eventMux *event.TypeMux
@@ -285,12 +287,30 @@ func New(config *Config) (*Ethereum, error) {
 
 	protocols := []p2p.Protocol{eth.protocolManager.SubProtocol}
 
+	if config.Bzz {
+		eth.netStore, err = bzz.NewNetStore(path.Join(config.DataDir, "bzz"), path.Join(config.DataDir, "bzzpeers.json"))
+		if err != nil {
+			glog.V(logger.Warn).Infof("BZZ: error creating net store: %v. Protocol skipped", err)
+		} else {
+			chunker := &bzz.TreeChunker{}
+			chunker.Init()
+			eth.DPA = &bzz.DPA{
+				Chunker:    chunker,
+				ChunkStore: eth.netStore,
+			}
+			protocols = append(protocols, bzz.BzzProtocol(eth.netStore))
+		}
+	}
+
+	if config.Shh {
+		protocols = append(protocols, eth.whisper.Protocol())
+	}
 	eth.net = &p2p.Server{
 		PrivateKey:      netprv,
 		Name:            config.Name,
 		MaxPeers:        config.MaxPeers,
 		MaxPendingPeers: config.MaxPendingPeers,
-		// Protocols:       protocols,
+		Protocols:       protocols,
 		NAT:             config.NAT,
 		NoDial:          !config.Dial,
 		BootstrapNodes:  config.parseBootNodes(),
@@ -301,26 +321,6 @@ func New(config *Config) (*Ethereum, error) {
 	if len(config.Port) > 0 {
 		eth.net.ListenAddr = ":" + config.Port
 	}
-	if config.Bzz {
-		netStore, err := bzz.NewNetStore(eth.net.Self().Sha(), path.Join(config.DataDir, "bzz"), path.Join(config.DataDir, "bzzpeers.json"))
-		if err != nil {
-			glog.V(logger.Warn).Infof("BZZ: error creating net store: %v. Protocol skipped", err)
-		} else {
-			chunker := &bzz.TreeChunker{}
-			chunker.Init()
-			dpa := &bzz.DPA{
-				Chunker:    chunker,
-				ChunkStore: netStore,
-			}
-			dpa.Start()
-			protocols = append(protocols, bzz.BzzProtocol(netStore, eth.net.Self))
-			go bzz.StartHttpServer(dpa)
-		}
-	}
-
-	if config.Shh {
-		protocols = append(protocols, eth.whisper.Protocol())
-	}
 
 	eth.net.Protocols = protocols
 
@@ -466,6 +466,12 @@ func (s *Ethereum) Start() error {
 		s.whisper.Start()
 	}
 
+	if s.DPA != nil {
+		s.DPA.Start()
+		s.netStore.Start(s.net.Self())
+		go bzz.StartHttpServer(s.DPA)
+	}
+
 	// broadcast transactions
 	s.txSub = s.eventMux.Subscribe(core.TxPreEvent{})
 	go s.txBroadcastLoop()