network/retrieval: add balances to the retrieval protocol (ethersphere#1758)

Author: janos

network/retrieval/retrieve.go

@@ -38,6 +38,7 @@ import (
 	"github.com/ethersphere/swarm/p2p/protocols"
 	"github.com/ethersphere/swarm/spancontext"
 	"github.com/ethersphere/swarm/storage"
+	"github.com/ethersphere/swarm/swap"
 	opentracing "github.com/opentracing/opentracing-go"
 	olog "github.com/opentracing/opentracing-go/log"
 )
@@ -53,7 +54,7 @@ var (
 
 	retrievalPeers = metrics.GetOrRegisterGauge("network.retrieve.peers", nil)
 
-	Spec = &protocols.Spec{
+	spec = &protocols.Spec{
 		Name:       "bzz-retrieve",
 		Version:    1,
 		MaxMsgSize: 10 * 1024 * 1024,
@@ -79,11 +80,11 @@ func (p *RetrievalPrices) Price(msg interface{}) *protocols.Price {
 }
 
 func (p *RetrievalPrices) retrieveRequestPrice() uint64 {
-	return uint64(1)
+	return swap.RetrieveRequestPrice
 }
 
 func (p *RetrievalPrices) chunkDeliveryPrice() uint64 {
-	return uint64(1)
+	return swap.ChunkDeliveryPrice
 }
 
 // createPriceOracle sets up a matrix which can be queried to get
@@ -111,21 +112,27 @@ type Retrieval struct {
 	netStore *storage.NetStore
 	kad      *network.Kademlia
 	peers    map[enode.ID]*Peer
+	spec     *protocols.Spec
 	prices   protocols.Prices
 	logger   log.Logger
 	quit     chan struct{}
 }
 
 // New returns a new instance of the retrieval protocol handler
-func New(kad *network.Kademlia, ns *storage.NetStore, baseKey []byte) *Retrieval {
+func New(kad *network.Kademlia, ns *storage.NetStore, baseKey []byte, balance protocols.Balance) *Retrieval {
 	r := &Retrieval{
 		kad:      kad,
 		peers:    make(map[enode.ID]*Peer),
+		spec:     spec,
 		netStore: ns,
 		logger:   log.New("base", hex.EncodeToString(baseKey)[:16]),
 		quit:     make(chan struct{}),
 	}
 	r.createPriceOracle()
+	if balance != nil && !reflect.ValueOf(balance).IsNil() {
+		// swap is enabled, so setup the hook
+		r.spec.Hook = protocols.NewAccounting(balance, r.prices)
+	}
 	return r
 }
 
@@ -226,7 +233,7 @@ func (r *Retrieval) findPeer(ctx context.Context, req *storage.Request) (retPeer
 	r.kad.EachConn(req.Addr[:], 255, func(p *network.Peer, po int) bool {
 		id := p.ID()
 
-		if !p.HasCap(Spec.Name) {
+		if !p.HasCap(r.spec.Name) {
 			return true
 		}
 
@@ -453,16 +460,16 @@ func (r *Retrieval) Stop() error {
 func (r *Retrieval) Protocols() []p2p.Protocol {
 	return []p2p.Protocol{
 		{
-			Name:    Spec.Name,
-			Version: Spec.Version,
-			Length:  Spec.Length(),
+			Name:    r.spec.Name,
+			Version: r.spec.Version,
+			Length:  r.spec.Length(),
 			Run:     r.runProtocol,
 		},
 	}
 }
 
 func (r *Retrieval) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error {
-	peer := protocols.NewPeer(p, rw, Spec)
+	peer := protocols.NewPeer(p, rw, r.spec)
 	bp := network.NewBzzPeer(peer)
 
 	return r.Run(bp)
@@ -471,3 +478,7 @@ func (r *Retrieval) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error {
 func (r *Retrieval) APIs() []rpc.API {
 	return nil
 }
+
+func (r *Retrieval) Spec() *protocols.Spec {
+	return r.spec
+}

network/retrieval/retrieve_test.go

@@ -242,7 +242,7 @@ func TestRequestFromPeers(t *testing.T) {
 
 	to.On(peer)
 
-	s := New(to, nil, to.BaseAddr())
+	s := New(to, nil, to.BaseAddr(), nil)
 
 	req := storage.NewRequest(storage.Address(hash0[:]))
 	id, err := s.findPeer(context.Background(), req)
@@ -273,7 +273,7 @@ func TestRequestFromPeersWithLightNode(t *testing.T) {
 
 	to.On(peer)
 
-	r := New(to, nil, to.BaseAddr())
+	r := New(to, nil, to.BaseAddr(), nil)
 	req := storage.NewRequest(storage.Address(hash0[:]))
 
 	// making a request which should return with "no peer found"
@@ -288,7 +288,7 @@ func TestRequestFromPeersWithLightNode(t *testing.T) {
 func TestHasPriceImplementation(t *testing.T) {
 	addr := network.RandomAddr()
 	to := network.NewKademlia(addr.OAddr, network.NewKadParams())
-	r := New(to, nil, to.BaseAddr())
+	r := New(to, nil, to.BaseAddr(), nil)
 
 	if r.prices == nil {
 		t.Fatal("No prices implementation available for retrieve protocol")
@@ -336,7 +336,7 @@ func newBzzRetrieveWithLocalstore(ctx *adapters.ServiceContext, bucket *sync.Map
 		return nil, nil, err
 	}
 
-	r := New(kad, netStore, kad.BaseAddr())
+	r := New(kad, netStore, kad.BaseAddr(), nil)
 	netStore.RemoteGet = r.RequestFromPeers
 	bucket.Store(bucketKeyFileStore, fileStore)
 	bucket.Store(bucketKeyNetstore, netStore)

network/stream/v2/common_test.go

@@ -127,7 +127,7 @@ func newSyncSimServiceFunc(o *SyncSimServiceOptions) func(ctx *adapters.ServiceC
 		bucket.Store(bucketKeyFileStore, fileStore)
 		bucket.Store(bucketKeyLocalStore, localStore)
 
-		ret := retrieval.New(kad, netStore, kad.BaseAddr())
+		ret := retrieval.New(kad, netStore, kad.BaseAddr(), nil)
 		netStore.RemoteGet = ret.RequestFromPeers
 
 		if o.InitialChunkCount > 0 {

swarm.go

@@ -208,7 +208,7 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e
 		common.FromHex(config.BzzKey),
 		network.NewKadParams(),
 	)
-	self.retrieval = retrieval.New(to, self.netStore, bzzconfig.OverlayAddr) // nodeID.Bytes())
+	self.retrieval = retrieval.New(to, self.netStore, bzzconfig.OverlayAddr, self.swap) // nodeID.Bytes())
 	self.netStore.RemoteGet = self.retrieval.RequestFromPeers
 
 	feedsHandler.SetStore(self.netStore)
@@ -228,7 +228,7 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e
 
 	log.Debug("Setup local storage")
 
-	self.bzz = network.NewBzz(bzzconfig, to, self.stateStore, stream.Spec, retrieval.Spec, self.streamer.Run, self.retrieval.Run)
+	self.bzz = network.NewBzz(bzzconfig, to, self.stateStore, stream.Spec, self.retrieval.Spec(), self.streamer.Run, self.retrieval.Run)
 
 	self.bzzEth = bzzeth.New()