swarm/network: Bundle syncer in swarm.go

swarm/network/kademlia.go

@@ -329,6 +329,12 @@ func (k *Kademlia) Off(p OverlayConn) {
 	}
 }
 
+func (k *Kademlia) EachBin(base []byte, pof pot.Pof, o int, eachBinFunc func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool) {
+	k.lock.RLock()
+	defer k.lock.RUnlock()
+	k.conns.EachBin(base, pof, o, eachBinFunc)
+}
+
 // EachConn is an iterator with args (base, po, f) applies f to each live peer
 // that has proximity order po or less as measured from the base
 // if base is nil, kademlia base address is used

swarm/network/protocol.go

@@ -113,9 +113,11 @@ type BzzConfig struct {
 // Bzz is the swarm protocol bundle
 type Bzz struct {
 	*Hive
-	localAddr  *BzzAddr
-	mtx        sync.Mutex
-	handshakes map[discover.NodeID]*HandshakeMsg
+	localAddr    *BzzAddr
+	mtx          sync.Mutex
+	handshakes   map[discover.NodeID]*HandshakeMsg
+	streamerSpec *protocols.Spec
+	streamerRun  func(*BzzPeer) error
 }
 
 // NewBzz is the swarm protocol constructor
@@ -123,11 +125,13 @@ type Bzz struct {
 // * bzz config
 // * overlay driver
 // * peer store
-func NewBzz(config *BzzConfig, kad Overlay, store state.Store) *Bzz {
+func NewBzz(config *BzzConfig, kad Overlay, store state.Store, streamerSpec *protocols.Spec, streamerRun func(*BzzPeer) error) *Bzz {
 	return &Bzz{
-		Hive:       NewHive(config.HiveParams, kad, store),
-		localAddr:  &BzzAddr{config.OverlayAddr, config.UnderlayAddr},
-		handshakes: make(map[discover.NodeID]*HandshakeMsg),
+		Hive:         NewHive(config.HiveParams, kad, store),
+		localAddr:    &BzzAddr{config.OverlayAddr, config.UnderlayAddr},
+		handshakes:   make(map[discover.NodeID]*HandshakeMsg),
+		streamerRun:  streamerRun,
+		streamerSpec: streamerSpec,
 	}
 }
 
@@ -166,6 +170,12 @@ func (b *Bzz) Protocols() []p2p.Protocol {
 			NodeInfo: b.Hive.NodeInfo,
 			PeerInfo: b.Hive.PeerInfo,
 		},
+		{
+			Name:    b.streamerSpec.Name,
+			Version: b.streamerSpec.Version,
+			Length:  b.streamerSpec.Length(),
+			Run:     b.RunProtocol(b.streamerSpec, b.streamerRun),
+		},
 	}
 }
 

swarm/network/stream/messages.go

@@ -61,6 +61,23 @@ type SubscribeMsg struct {
 	Priority uint8  // delivered on priority channel
 }
 
+// RequestSubscriptionMsg is the protocol msg for a node to request subscription to a
+// specific stream
+type RequestSubscriptionMsg struct {
+	Stream   Stream
+	History  *Range `rlp:"nil"`
+	Priority uint8  // delivered on priority channel
+}
+
+func (p *Peer) handleRequestSubscription(req *RequestSubscriptionMsg) (err error) {
+	log.Debug(fmt.Sprintf("handleRequestSubscription: streamer %s to subscribe to %s with stream %s", p.streamer.addr.ID(), p.ID(), req.Stream))
+	err = p.streamer.Subscribe(p.ID(), req.Stream, req.History, req.Priority)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) {
 	defer func() {
 		if err != nil {

swarm/network/stream/stream.go

@@ -22,12 +22,12 @@ import (
 	"math"
 	"sync"
 
-	"github.com/ethereum/go-ethereum/p2p"
-	"github.com/ethereum/go-ethereum/rpc"
-
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/protocols"
+	"github.com/ethereum/go-ethereum/pot"
+	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
 	"github.com/ethereum/go-ethereum/swarm/state"
@@ -123,6 +123,26 @@ func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte, bool) (Serv
 	return f, nil
 }
 
+func (r *Registry) RequestSubscription(peerId discover.NodeID, s Stream, h *Range, prio uint8) error {
+	// check if the stream is registered
+	if _, err := r.GetClientFunc(s.Name); err != nil {
+		return err
+	}
+
+	peer := r.getPeer(peerId)
+	if peer == nil {
+		return fmt.Errorf("peer not found %v", peerId)
+	}
+
+	msg := &RequestSubscriptionMsg{
+		Stream:   s,
+		History:  h,
+		Priority: prio,
+	}
+	log.Debug("RequestSubscription ", "peer", peerId, "stream", s, "history", h)
+	return peer.Send(msg)
+}
+
 // Subscribe initiates the streamer
 func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priority uint8) error {
 	// check if the stream is registered
@@ -225,12 +245,71 @@ func (r *Registry) peersCount() (c int) {
 }
 
 // Run protocol run function
-func (r *Registry) run(p *protocols.Peer) error {
-	sp := NewPeer(p, r)
+func (r *Registry) Run(p *network.BzzPeer) error {
+	sp := NewPeer(p.Peer, r)
 	r.setPeer(sp)
 	defer r.deletePeer(sp)
 	defer close(sp.quit)
 	defer sp.close()
+
+	var kadDepth int
+
+	r.delivery.overlay.EachConn(nil, 256, func(addr network.OverlayConn, po int, nn bool) bool {
+		// TODO: stop or expose by kademlia
+		if nn {
+			kadDepth = po
+		}
+		return true
+	})
+
+	kad, ok := r.delivery.overlay.(*network.Kademlia)
+	if !ok {
+		return fmt.Errorf("Not a Kademlia!")
+	}
+
+	var startPo int
+	var endPo int
+	var i int
+	var err error
+
+	//iterate over each bin and solicit needed subscription to bins
+	kad.EachBin(r.addr.Over(), pot.DefaultPof(256), 0, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool {
+
+		//identify begin and start index of the bin(s) we want to subscribe to
+		if po < kadDepth {
+			//not nn
+			endPo = po
+			if i > 0 {
+				startPo = endPo + 1
+			}
+		} else if endPo < kadDepth || endPo == 0 {
+			if po == 0 && kadDepth == 0 {
+				startPo = endPo
+			} else {
+				startPo = endPo + 1
+			}
+			endPo = maxPO
+		}
+
+		// now iterate and subscribe
+		for bin := po - startPo; bin <= endPo; bin++ {
+
+			f(func(val pot.Val, i int) bool {
+				// a := val.(network.OverlayPeer)
+				log.Debug(fmt.Sprintf("Requesting subscription by: registry %s from peer %s for bin: %d", r.addr.ID(), p.ID(), bin))
+
+				err = r.RequestSubscription(p.ID(), NewStream("SYNC", []byte{uint8(bin)}, true), &Range{}, Top)
+				if err != nil {
+					log.Error(fmt.Sprintf("Error in RequestSubsciption! %v", err))
+					return false
+				}
+				return true
+			})
+		}
+		i++
+		return true
+	})
+
 	return sp.Run(sp.HandleMsg)
 }
 
@@ -239,7 +318,7 @@ func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error {
 	bzzPeer := network.NewBzzTestPeer(peer, r.addr)
 	r.delivery.overlay.On(bzzPeer)
 	defer r.delivery.overlay.Off(bzzPeer)
-	return r.run(peer)
+	return r.Run(bzzPeer)
 }
 
 // HandleMsg is the message handler that delegates incoming messages
@@ -270,6 +349,9 @@ func (p *Peer) HandleMsg(msg interface{}) error {
 	case *RetrieveRequestMsg:
 		return p.streamer.delivery.handleRetrieveRequestMsg(p, msg)
 
+	case *RequestSubscriptionMsg:
+		return p.handleRequestSubscription(msg)
+
 	default:
 		return fmt.Errorf("unknown message type: %T", msg)
 	}
@@ -428,6 +510,7 @@ var Spec = &protocols.Spec{
 		RetrieveRequestMsg{},
 		ChunkDeliveryMsg{},
 		SubscribeErrorMsg{},
+		RequestSubscriptionMsg{},
 	},
 }
 
@@ -457,6 +540,7 @@ func (r *Registry) APIs() []rpc.API {
 
 func (r *Registry) Start(server *p2p.Server) error {
 	r.api.dpa.Start()
+	log.Info("Streamer started")
 	return nil
 }
 

swarm/swarm.go

@@ -157,7 +157,7 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.
 	stream.RegisterSwarmSyncerServer(self.streamer, db)
 	stream.RegisterSwarmSyncerClient(self.streamer, db)
 
-	self.bzz = network.NewBzz(bzzconfig, to, stateStore)
+	self.bzz = network.NewBzz(bzzconfig, to, stateStore, stream.Spec, self.streamer.Run)
 
 	// set up DPA, the cloud storage local access layer
 	dpaChunkStore := storage.NewNetStore(self.lstore, self.streamer.Retrieve)
@@ -378,6 +378,7 @@ func (self *Swarm) Start(srv *p2p.Server) error {
 	self.periodicallyUpdateGauges()
 
 	startCounter.Inc(1)
+	self.streamer.Start(srv)
 	return nil
 }
 
@@ -413,6 +414,7 @@ func (self *Swarm) Stop() error {
 	}
 	self.sfs.Stop()
 	stopCounter.Inc(1)
+	self.streamer.Stop()
 	return self.bzz.Stop()
 }