swarm/swap: swap implemented as protocols/protocol.Peer extension

swarm/network/stream/delivery.go

@@ -30,7 +30,6 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/spancontext"
 	"github.com/ethereum/go-ethereum/swarm/storage"
-	"github.com/ethereum/go-ethereum/swarm/swap"
 	opentracing "github.com/opentracing/opentracing-go"
 )
 
@@ -140,8 +139,8 @@ type RetrieveRequestMsg struct {
 }
 
 //TODO: what is the correct price
-func (rrm *RetrieveRequestMsg) GetMsgPrice() (*big.Int, swap.EntryDirection) {
+func (rrm *RetrieveRequestMsg) GetMsgPrice() *big.Int {
-	return big.NewInt(int64(4096)), swap.CreditEntry
+	return big.NewInt(int64(4096))
 }
 
 func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *RetrieveRequestMsg) error {

swarm/network/stream/peer.go

@@ -31,6 +31,7 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/spancontext"
 	"github.com/ethereum/go-ethereum/swarm/state"
 	"github.com/ethereum/go-ethereum/swarm/storage"
+	"github.com/ethereum/go-ethereum/swarm/swap"
 	opentracing "github.com/opentracing/opentracing-go"
 )
 
@@ -53,7 +54,8 @@ var ErrMaxPeerServers = errors.New("max peer servers")
 
 // Peer is the Peer extension for the streaming protocol
 type Peer struct {
-	*protocols.Peer
+	//*protocols.Peer
+	*swap.SwapPeer
 	streamer *Registry
 	pq       *pq.PriorityQueue
 	serverMu sync.RWMutex
@@ -75,7 +77,7 @@ type WrappedPriorityMsg struct {
 // NewPeer is the constructor for Peer
 func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer {
 	p := &Peer{
-		Peer:         peer,
+		SwapPeer:     swap.NewSwapPeer(peer, streamer.swap),
 		pq:           pq.New(int(PriorityQueue), PriorityQueueCap),
 		streamer:     streamer,
 		servers:      make(map[Stream]*server),

swarm/network/stream/stream.go

@@ -385,7 +385,7 @@ func (r *Registry) Run(p *network.BzzPeer) error {
 		}
 	}
 
-	return sp.Run(sp.HandleAccountedMsg)
+	return sp.RunAccountedProtocol(sp.HandleMsg)
 }
 
 // updateSyncing subscribes to SYNC streams by iterating over the
@@ -460,16 +460,8 @@ func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error {
 	return r.Run(bp)
 }
 
-func (p *Peer) HandleAccountedMsg(ctx context.Context, msg interface{}) error {
-	err := p.handleMsg(ctx, msg)
-	if _, ok := msg.(swap.SwapAccountedMsgType); ok && err == nil {
-		p.streamer.swap.AccountForMsg(ctx, msg, p.ID())
-	}
-	return err
-}
-
 // HandleMsg is the message handler that delegates incoming messages
-func (p *Peer) handleMsg(ctx context.Context, msg interface{}) error {
+func (p *Peer) HandleMsg(ctx context.Context, msg interface{}) error {
 	switch msg := msg.(type) {
 
 	case *SubscribeMsg:

swarm/swap/protocol.go

@@ -35,25 +35,25 @@ const (
 
 type SwapProtocol struct {
 	peersMu sync.RWMutex
-	peers   map[discover.NodeID]*SwapPeer
+	peers   map[discover.NodeID]*SwapProtocolPeer
 }
 
 // Peer is the Peer extension for the streaming protocol
-type SwapPeer struct {
+type SwapProtocolPeer struct {
 	*protocols.Peer
 	swapProtocol *SwapProtocol
 }
 
 func NewSwapProtocol() *SwapProtocol {
 	proto := &SwapProtocol{
-		peers: make(map[discover.NodeID]*SwapPeer),
+		peers: make(map[discover.NodeID]*SwapProtocolPeer),
 	}
 	return proto
 }
 
 // NewPeer is the constructor for Peer
-func NewPeer(peer *protocols.Peer, swap *SwapProtocol) *SwapPeer {
+func NewPeer(peer *protocols.Peer, swap *SwapProtocol) *SwapProtocolPeer {
-	p := &SwapPeer{
+	p := &SwapProtocolPeer{
 		Peer:         peer,
 		swapProtocol: swap,
 	}
@@ -100,19 +100,6 @@ func (p *SwapProtocol) Protocols() []p2p.Protocol {
 	}
 }
 
-func (swap *SwapProtocol) DebitByteCount(peer *SwapPeer, numberOfBytes int) error {
-	return nil
-}
-
-func (swap *SwapProtocol) Run(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
-	p := protocols.NewPeer(peer, rw, swapSpec)
-	sp := NewPeer(p, swap)
-	swap.setPeer(sp)
-	defer swap.deletePeer(sp)
-	defer swap.Close()
-	return sp.Run(sp.handleSwapMsg)
-}
-
 func (p *SwapProtocol) APIs() []rpc.API {
 	apis := []rpc.API{
 		{
@@ -125,7 +112,17 @@ func (p *SwapProtocol) APIs() []rpc.API {
 	return apis
 }
 
-//--------------------
+/////////////////////////////////////////////////////////////////////
+// SECTION: p2p.protocol interface
+/////////////////////////////////////////////////////////////////////
+func (swap *SwapProtocol) Run(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
+	p := protocols.NewPeer(peer, rw, swapSpec)
+	sp := NewPeer(p, swap)
+	swap.setPeer(sp)
+	defer swap.deletePeer(sp)
+	defer swap.Close()
+	return sp.Run(sp.handleSwapMsg)
+}
 
 func (swap *SwapProtocol) NodeInfo() interface{} {
 	return nil
@@ -135,18 +132,23 @@ func (swap *SwapProtocol) PeerInfo(id discover.NodeID) interface{} {
 	return nil
 }
 
-func (swap *SwapProtocol) Close() error {
+//------------------------------------------------------------------------------------------
+
+func (swap *SwapProtocol) DebitByteCount(peer *SwapProtocolPeer, numberOfBytes int) error {
 	return nil
 }
 
-func (swap *SwapProtocol) getPeer(peerId discover.NodeID) *SwapPeer {
+func (swap *SwapProtocol) Close() {
+}
+
+func (swap *SwapProtocol) getPeer(peerId discover.NodeID) *SwapProtocolPeer {
 	swap.peersMu.RLock()
 	defer swap.peersMu.RUnlock()
 
 	return swap.peers[peerId]
 }
 
-func (swap *SwapProtocol) setPeer(peer *SwapPeer) {
+func (swap *SwapProtocol) setPeer(peer *SwapProtocolPeer) {
 	swap.peersMu.Lock()
 	defer swap.peersMu.Unlock()
 
@@ -154,7 +156,7 @@ func (swap *SwapProtocol) setPeer(peer *SwapPeer) {
 	metrics.GetOrRegisterGauge("registry.peers", nil).Update(int64(len(swap.peers)))
 }
 
-func (swap *SwapProtocol) deletePeer(peer *SwapPeer) {
+func (swap *SwapProtocol) deletePeer(peer *SwapProtocolPeer) {
 	swap.peersMu.Lock()
 	defer swap.peersMu.Unlock()
 
@@ -170,7 +172,7 @@ func (swap *SwapProtocol) peersCount() (c int) {
 	return
 }
 
-func (p *SwapPeer) handleSwapMsg(ctx context.Context, msg interface{}) error {
+func (p *SwapProtocolPeer) handleSwapMsg(ctx context.Context, msg interface{}) error {
 	switch msg := msg.(type) {
 
 	case *IssueChequeMsg:
@@ -190,10 +192,10 @@ func (p *SwapPeer) handleSwapMsg(ctx context.Context, msg interface{}) error {
 	return nil
 }
 
-func (sp *SwapPeer) handleIssueChequeMsg(ctx context.Context, msg interface{}) (err error) {
+func (sp *SwapProtocolPeer) handleIssueChequeMsg(ctx context.Context, msg interface{}) (err error) {
 	return err
 }
 
-func (sp *SwapPeer) handleRedeemChequeMsg(ctx context.Context, msg interface{}) (err error) {
+func (sp *SwapProtocolPeer) handleRedeemChequeMsg(ctx context.Context, msg interface{}) (err error) {
 	return err
 }

swarm/swap/swap.go

@@ -33,6 +33,7 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/p2p/discover"
+	"github.com/ethereum/go-ethereum/p2p/protocols"
 	"github.com/ethereum/go-ethereum/swarm/log"
 	"github.com/ethereum/go-ethereum/swarm/state"
 	whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
@@ -67,10 +68,19 @@ const (
 type Swap struct {
 	stateStore state.Store
 	lock       sync.RWMutex
-	peers      map[discover.NodeID]*swapPeer
+	peers      map[discover.NodeID]*SwapPeer
 	local      *Params // local peer's swap parameters
 }
 
+type SwapPeer struct {
+	*protocols.Peer
+	lock        sync.RWMutex
+	swapAccount *Swap
+	handlerFunc func(context.Context, interface{}) error
+	balance     *big.Int
+	storeID     string
+}
+
 type EntryDirection bool
 
 const (
@@ -79,26 +89,20 @@ const (
 )
 
 type SwapAccountedMsgType interface {
-	GetMsgPrice() (*big.Int, EntryDirection)
+	GetMsgPrice() *big.Int
 }
 
-func (swap *Swap) AccountForMsg(ctx context.Context, msg interface{}, peer discover.NodeID) error {
+func (sp *SwapPeer) RunAccountedProtocol(protocolHandler func(ctx context.Context, msg interface{}) error) error {
+	sp.handlerFunc = protocolHandler
+
+	return sp.Run(sp.handleAccountedMsg)
+}
+
+func (sp *SwapPeer) doAccountMsg(ctx context.Context, msg interface{}, direction EntryDirection) error {
 	if accounted, ok := msg.(SwapAccountedMsgType); ok {
-		if _, exists := swap.peers[peer]; !exists {
+		price := accounted.GetMsgPrice()
-			balance := big.NewInt(0)
-			swap.stateStore.Get(peer.String()[:24]+"-swap", &balance)
-			swap.lock.Lock()
-			swap.peers[peer] = &swapPeer{
-				peer:        peer,
-				swapAccount: swap,
-				balance:     balance,
-				storeID:     peer.String()[:24] + "-swap",
-			}
-			swap.lock.Unlock()
-		}
-		price, direction := accounted.GetMsgPrice()
 		//TODO: Calculate total price and account
-		swap.peers[peer].AccountMsgForPeer(price, direction)
+		sp.AccountMsgForPeer(price, direction)
 	}
 	return nil
 }
@@ -110,6 +114,61 @@ func (swap *Swap) GetPeerBalance(peer discover.NodeID) *big.Int {
 	return nil
 }
 
+func (sp *SwapPeer) handleAccountedMsg(ctx context.Context, msg interface{}) error {
+	err := sp.handlerFunc(ctx, msg)
+	if _, ok := msg.(SwapAccountedMsgType); ok && err == nil {
+		sp.doAccountMsg(ctx, msg, CreditEntry)
+	}
+	return err
+}
+
+func (sp *SwapPeer) Send(ctx context.Context, msg interface{}) error {
+	err := sp.Peer.Send(ctx, msg)
+	if _, ok := msg.(SwapAccountedMsgType); ok && err == nil {
+		sp.doAccountMsg(ctx, msg, DebitEntry)
+	}
+	return err
+}
+
+//The balance is accounted from the point of view of the local node
+//Thus, we credit the balance and increase it when the amount is in favor of the local node
+//We debit the balance and decrease it when the amount is in favor of the remote peer
+func (sp *SwapPeer) AccountMsgForPeer(price *big.Int, direction EntryDirection) {
+	sp.lock.Lock()
+	defer sp.lock.Unlock()
+	//local node is being credited (in its favor), so its balance increases
+	if direction == CreditEntry {
+		sp.balance = sp.balance.Add(sp.balance, price)
+		//local node is being debited (in favor of remote peer), so its balance decreases
+	} else if direction == DebitEntry {
+		sp.balance = sp.balance.Sub(sp.balance, price)
+	}
+	//TODO: save to store here? init store?
+	sp.swapAccount.stateStore.Put(sp.storeID, sp.balance)
+	if sp.balance.Cmp(payAt) > -1 {
+		//TODO: Issue Cheque
+	}
+	if sp.balance.Cmp(dropAt) < 0 {
+		//TODO: Drop peer
+	}
+	log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.ID(), sp.balance.String()))
+}
+
+func NewSwapPeer(peer *protocols.Peer, swap *Swap) *SwapPeer {
+	balance := big.NewInt(0)
+	swap.stateStore.Get(peer.String()[:24]+"-swap", &balance)
+	sp := &SwapPeer{
+		Peer:        peer,
+		swapAccount: swap,
+		balance:     balance,
+		storeID:     peer.String()[:24] + "-swap",
+	}
+	swap.lock.Lock()
+	defer swap.lock.Unlock()
+	swap.peers[peer.ID()] = sp
+	return sp
+}
+
 // Profile - public swap profile
 // public parameters for SWAP, serializable config struct passed in handshake
 type Profile struct {
@@ -168,42 +227,13 @@ type PayProfile struct {
 	lock        sync.RWMutex
 }
 
-type swapPeer struct {
-	lock        sync.RWMutex
-	peer        discover.NodeID
-	swapAccount *Swap
-	balance     *big.Int
-	storeID     string
-}
-
-func (sp *swapPeer) AccountMsgForPeer(price *big.Int, direction EntryDirection) {
-	sp.lock.Lock()
-	defer sp.lock.Unlock()
-	//the peer is being credited (in its favor), so its balance increases
-	if direction == CreditEntry {
-		sp.balance = sp.balance.Add(sp.balance, price)
-		//the peer is being debited (in local favor), so its balance decreases
-	} else if direction == DebitEntry {
-		sp.balance = sp.balance.Sub(sp.balance, price)
-	}
-	//TODO: save to store here? init store?
-	sp.swapAccount.stateStore.Put(sp.storeID, sp.balance)
-	if sp.balance.Cmp(payAt) > -1 {
-		//TODO: Issue Cheque
-	}
-	if sp.balance.Cmp(dropAt) < 0 {
-		//TODO: Drop peer
-	}
-	log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.peer, sp.balance.String()))
-}
-
 // New - swap constructor
 func NewSwap(local *Params, stateStore state.Store) (swap *Swap, err error) {
 
 	swap = &Swap{
 		local:      local,
 		stateStore: stateStore,
-		peers:      make(map[discover.NodeID]*swapPeer),
+		peers:      make(map[discover.NodeID]*SwapPeer),
 	}
 
 	//swap.SetParams(local)