diff options
| author | Viktor Trón <viktor.tron@gmail.com> | 2018-09-12 17:24:56 +0800 |
|---|---|---|
| committer | Balint Gabor <balint.g@gmail.com> | 2018-09-12 17:24:56 +0800 |
| commit | bfce00385f1c8dab222b7ddab6c336177a5ae731 (patch) | |
| tree | 8b3d2b2ce30e8b5eaf6db5b89a6c5570c3997cff /swarm/network/stream/stream.go | |
| parent | b06ff563a1f1095407612e04a1707e13d2dc20da (diff) | |
| download | dexon-bfce00385f1c8dab222b7ddab6c336177a5ae731.tar.gz dexon-bfce00385f1c8dab222b7ddab6c336177a5ae731.tar.zst dexon-bfce00385f1c8dab222b7ddab6c336177a5ae731.zip | |
Kademlia refactor (#17641)
* swarm/network: simplify kademlia/hive; rid interfaces
* swarm, swarm/network/stream, swarm/netork/simulations,, swarm/pss: adapt to new Kad API
* swarm/network: minor changes re review; add missing lock to NeighbourhoodDepthC
Diffstat (limited to 'swarm/network/stream/stream.go')
| -rw-r--r-- | swarm/network/stream/stream.go | 18 |
1 files changed, 8 insertions, 10 deletions
diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index cd0580a0c..deffdfc3f 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -130,7 +130,7 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i // wait for kademlia table to be healthy time.Sleep(options.SyncUpdateDelay) - kad := streamer.delivery.overlay.(*network.Kademlia) + kad := streamer.delivery.kad depthC := latestIntC(kad.NeighbourhoodDepthC()) addressBookSizeC := latestIntC(kad.AddrCountC()) @@ -398,9 +398,7 @@ func (r *Registry) Run(p *network.BzzPeer) error { // and they are no longer required after iteration, request to Quit // them will be send to appropriate peers. func (r *Registry) updateSyncing() { - // if overlay in not Kademlia, panic - kad := r.delivery.overlay.(*network.Kademlia) - + kad := r.delivery.kad // map of all SYNC streams for all peers // used at the and of the function to remove servers // that are not needed anymore @@ -421,8 +419,7 @@ func (r *Registry) updateSyncing() { r.peersMu.RUnlock() // request subscriptions for all nodes and bins - kad.EachBin(r.addr.Over(), pot.DefaultPof(256), 0, func(conn network.OverlayConn, bin int) bool { - p := conn.(network.Peer) + kad.EachBin(r.addr.Over(), pot.DefaultPof(256), 0, func(p *network.Peer, bin int) bool { log.Debug(fmt.Sprintf("Requesting subscription by: registry %s from peer %s for bin: %d", r.addr.ID(), p.ID(), bin)) // bin is always less then 256 and it is safe to convert it to type uint8 @@ -461,10 +458,11 @@ func (r *Registry) updateSyncing() { func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error { peer := protocols.NewPeer(p, rw, Spec) - bzzPeer := network.NewBzzTestPeer(peer, r.addr) - r.delivery.overlay.On(bzzPeer) - defer r.delivery.overlay.Off(bzzPeer) - return r.Run(bzzPeer) + bp := network.NewBzzPeer(peer, r.addr) + np := network.NewPeer(bp, r.delivery.kad) + r.delivery.kad.On(np) + defer r.delivery.kad.Off(np) + return r.Run(bp) } // HandleMsg is the message handler that delegates incoming messages |