Merge ce63aec0c9 into 99e1a5e0fb

swarm/network/stream/common_test.go

@@ -109,7 +109,7 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora
 
 	db := storage.NewDBAPI(localStore)
 	delivery := NewDelivery(to, db)
-	streamer := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil)
+	streamer := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, nil)
 	teardown := func() {
 		streamer.Close()
 		removeDataDir()

swarm/network/stream/delivery.go

@@ -19,6 +19,7 @@ package stream
 import (
 	"context"
 	"errors"
+	"math/big"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -138,6 +139,11 @@ type RetrieveRequestMsg struct {
 	SkipCheck bool
 }
 
+//TODO: what is the correct price
+func (rrm *RetrieveRequestMsg) GetMsgPrice() *big.Int {
+	return big.NewInt(int64(4096))
+}
+
 func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *RetrieveRequestMsg) error {
 	log.Trace("received request", "peer", sp.ID(), "hash", req.Addr)
 	handleRetrieveRequestMsgCount.Inc(1)

swarm/network/stream/delivery_test.go

@@ -328,7 +328,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 			kad := network.NewKademlia(addr.Over(), network.NewKadParams())
 			delivery := NewDelivery(kad, db)
 
-			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{
+			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, &RegistryOptions{
 				SkipCheck: skipCheck,
 			})
 			bucket.Store(bucketKeyRegistry, r)
@@ -515,7 +515,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 			kad := network.NewKademlia(addr.Over(), network.NewKadParams())
 			delivery := NewDelivery(kad, db)
 
-			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{
+			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, &RegistryOptions{
 				SkipCheck:       skipCheck,
 				DoSync:          true,
 				SyncUpdateDelay: 0,

swarm/network/stream/intervals_test.go

@@ -74,7 +74,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
 			kad := network.NewKademlia(addr.Over(), network.NewKadParams())
 			delivery := NewDelivery(kad, db)
 
-			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{
+			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, &RegistryOptions{
 				SkipCheck: skipCheck,
 			})
 			bucket.Store(bucketKeyRegistry, r)

swarm/network/stream/peer.go

@@ -30,6 +30,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"
 )
 
@@ -50,7 +51,8 @@ func (e *notFoundError) Error() string {
 
 // 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
@@ -72,7 +74,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/snapshot_retrieval_test.go

@@ -133,7 +133,7 @@ func runFileRetrievalTest(nodeCount int) error {
 			kad := network.NewKademlia(addr.Over(), network.NewKadParams())
 			delivery := NewDelivery(kad, db)
 
-			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{
+			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, &RegistryOptions{
 				DoSync:          true,
 				SyncUpdateDelay: 3 * time.Second,
 			})
@@ -276,7 +276,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
 			kad := network.NewKademlia(addr.Over(), network.NewKadParams())
 			delivery := NewDelivery(kad, db)
 
-			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{
+			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, &RegistryOptions{
 				DoSync:          true,
 				SyncUpdateDelay: 0,
 			})

swarm/network/stream/snapshot_sync_test.go

@@ -139,7 +139,7 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) {
 			kad := network.NewKademlia(addr.Over(), network.NewKadParams())
 			delivery := NewDelivery(kad, db)
 
-			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{
+			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, &RegistryOptions{
 				DoSync:          true,
 				SyncUpdateDelay: 3 * time.Second,
 			})
@@ -297,7 +297,7 @@ func testSyncingViaDirectSubscribe(chunkCount int, nodeCount int) error {
 			kad := network.NewKademlia(addr.Over(), network.NewKadParams())
 			delivery := NewDelivery(kad, db)
 
-			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil)
+			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, nil)
 			bucket.Store(bucketKeyRegistry, r)
 
 			fileStore := storage.NewFileStore(storage.NewNetStore(localStore, nil), storage.NewFileStoreParams())

swarm/network/stream/stream.go

@@ -35,6 +35,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"
 )
 
@@ -50,6 +51,7 @@ const (
 
 // Registry registry for outgoing and incoming streamer constructors
 type Registry struct {
+	swap           *swap.Swap
 	api            *API
 	addr           *network.BzzAddr
 	skipCheck      bool
@@ -73,7 +75,7 @@ type RegistryOptions struct {
 }
 
 // NewRegistry is Streamer constructor
-func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, intervalsStore state.Store, options *RegistryOptions) *Registry {
+func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, intervalsStore state.Store, swap *swap.Swap, options *RegistryOptions) *Registry {
 	if options == nil {
 		options = &RegistryOptions{}
 	}
@@ -101,6 +103,8 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i
 	RegisterSwarmSyncerServer(streamer, db)
 	RegisterSwarmSyncerClient(streamer, db)
 
+	streamer.swap = swap
+
 	if options.DoSync {
 		// latestIntC function ensures that
 		//   - receiving from the in chan is not blocked by processing inside the for loop
@@ -390,7 +394,7 @@ func (r *Registry) Run(p *network.BzzPeer) error {
 		}
 	}
 
-	return sp.Run(sp.HandleMsg)
+	return sp.RunAccountedProtocol(sp.HandleMsg)
 }
 
 // updateSyncing subscribes to SYNC streams by iterating over the

swarm/network/stream/syncer_test.go

@@ -108,7 +108,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 			delivery := NewDelivery(kad, db)
 			bucket.Store(bucketKeyDelivery, delivery)
 
-			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{
+			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil, &RegistryOptions{
 				SkipCheck: skipCheck,
 			})
 

swarm/network_test.go

@@ -21,6 +21,7 @@ import (
 	"flag"
 	"fmt"
 	"io/ioutil"
+	"math/big"
 	"math/rand"
 	"os"
 	"sync"
@@ -41,9 +42,12 @@ import (
 )
 
 var (
-	loglevel     = flag.Int("loglevel", 2, "verbosity of logs")
+	loglevel       = flag.Int("loglevel", 2, "verbosity of logs")
-	longrunning  = flag.Bool("longrunning", false, "do run long-running tests")
+	longrunning    = flag.Bool("longrunning", false, "do run long-running tests")
-	waitKademlia = flag.Bool("waitkademlia", false, "wait for healthy kademlia before checking files availability")
+	waitKademlia   = flag.Bool("waitkademlia", false, "wait for healthy kademlia before checking files availability")
+	printStats     = flag.Bool("printstats", false, "print accounting stats to STDOUT")
+	bucketKeySwap  = simulation.BucketKey("swap")
+	bucketKeySwarm = simulation.BucketKey("swarm")
 )
 
 func init() {
@@ -373,6 +377,143 @@ func testSwarmNetwork(t *testing.T, o *testSwarmNetworkOptions, steps ...testSwa
 	}
 }
 
+func TestSwapNetworkSymmetricFileUpload(t *testing.T) {
+	nodeCount := 16
+
+	sim := simulation.New(map[string]simulation.ServiceFunc{
+		"swarm": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
+			config := api.NewConfig()
+
+			dir, err := ioutil.TempDir("", "swap-network-test-node")
+			if err != nil {
+				return nil, nil, err
+			}
+			cleanup = func() {
+				err := os.RemoveAll(dir)
+				if err != nil {
+					log.Error("cleaning up swarm temp dir", "err", err)
+				}
+			}
+
+			config.Path = dir
+
+			privkey, err := crypto.GenerateKey()
+			if err != nil {
+				return nil, cleanup, err
+			}
+
+			config.Init(privkey)
+
+			swarm, err := NewSwarm(config, nil)
+			if err != nil {
+				return nil, cleanup, err
+			}
+			bucket.Store(bucketKeySwarm, swarm)
+			log.Info("new swarm", "bzzKey", config.BzzKey, "baseAddr", fmt.Sprintf("%x", swarm.bzz.BaseAddr()))
+			return swarm, cleanup, nil
+		},
+	})
+	defer sim.Close()
+
+	ctx := context.Background()
+	files := make([]file, 0)
+
+	var checkStatusM sync.Map
+	var nodeStatusM sync.Map
+	var totalFoundCount uint64
+
+	_, err := sim.AddNodesAndConnectChain(nodeCount)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
+		nodeIDs := sim.UpNodeIDs()
+		shuffle(len(nodeIDs), func(i, j int) {
+			nodeIDs[i], nodeIDs[j] = nodeIDs[j], nodeIDs[i]
+		})
+		for _, id := range nodeIDs {
+			key, data, err := uploadFile(sim.Service("swarm", id).(*Swarm))
+			if err != nil {
+				return err
+			}
+			log.Trace("file uploaded", "node", id, "key", key.String())
+			files = append(files, file{
+				addr:   key,
+				data:   data,
+				nodeID: id,
+			})
+		}
+
+		if _, err := sim.WaitTillHealthy(ctx, 2); err != nil {
+			return err
+		}
+
+		// File retrieval check is repeated until all uploaded files are retrieved from all nodes
+		// or until the timeout is reached.
+		for {
+			if retrieve(sim, files, &checkStatusM, &nodeStatusM, &totalFoundCount) == 0 {
+				return nil
+			}
+		}
+	})
+
+	balancesMap := make(map[discover.NodeID]map[discover.NodeID]*big.Int)
+
+	for _, node := range sim.NodeIDs() {
+		item, ok := sim.NodeItem(node, bucketKeySwarm)
+		if !ok {
+			log.Error("No swarm")
+			return
+		}
+		swarm := item.(*Swarm)
+
+		subBalances := make(map[discover.NodeID]*big.Int)
+
+		for _, n := range sim.NodeIDs() {
+			if node == n {
+				continue
+			}
+			balance := swarm.swap.GetPeerBalance(n)
+			if balance != nil {
+				subBalances[n] = balance
+				log.Debug(fmt.Sprintf("Balance of node %s to node %s: %s", node.TerminalString(), n.TerminalString(), swarm.swap.GetPeerBalance(n).String()))
+			} else {
+				log.Debug(fmt.Sprintf("Node %s has no balance with node %s", node.TerminalString(), n.TerminalString()))
+			}
+		}
+		balancesMap[node] = subBalances
+	}
+
+	if *printStats {
+		for k, v := range balancesMap {
+			fmt.Println(fmt.Sprintf("node %s balances:", k.TerminalString()))
+			for kk, vv := range v {
+				fmt.Println(fmt.Sprintf(".........with node %s: balance %s", kk.TerminalString(), vv.String()))
+			}
+		}
+	}
+
+	for k, mapForK := range balancesMap {
+		for n, balanceKwithN := range mapForK {
+			for subK, mapForSubK := range balancesMap {
+				if n == subK {
+					log.Trace(fmt.Sprintf("balance of %s with %s: %s", k.TerminalString(), n.TerminalString(), balanceKwithN))
+					log.Trace(fmt.Sprintf("balance of %s with %s: %s", n.TerminalString(), k.TerminalString(), mapForSubK[k]))
+					if balanceKwithN.CmpAbs(mapForSubK[k]) != 0 && balanceKwithN.Cmp(big.NewInt(0)) != 0 {
+						log.Error("Expected balances to be |abs| = 0 AND balance1 != 0, but they are not")
+					}
+				}
+			}
+		}
+	}
+
+	if result.Error != nil {
+		t.Fatal(result.Error)
+	}
+	log.Debug("test terminated")
+}
+
 // uploadFile, uploads a short file to the swarm instance
 // using the api.Put method.
 func uploadFile(swarm *Swarm) (storage.Address, string, error) {

swarm/swap/api.go

@@ -0,0 +1,69 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package swap
+
+import (
+	"context"
+	"errors"
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/p2p/discover"
+)
+
+var (
+	ErrNoSuchPeerAccounting = errors.New("No accounting with that peer")
+)
+
+// Wrapper for receiving pss messages when using the pss API
+// providing access to sender of message
+type APIMsg struct {
+	Msg hexutil.Bytes
+}
+
+// Additional public methods accessible through API for pss
+type API struct {
+	*SwapProtocol
+}
+
+//TODO: define metrics
+type SwapMetrics struct {
+}
+
+func NewAPI(swap *SwapProtocol) *API {
+	return &API{SwapProtocol: swap}
+}
+
+func (swapapi *API) BalanceWithPeer(ctx context.Context, peer discover.NodeID) (balance *big.Int, err error) {
+	balance = swapapi.swap.peers[peer].balance
+	if balance == nil {
+		err = ErrNoSuchPeerAccounting
+	}
+	return
+}
+
+func (swapapi *API) Balance(ctx context.Context) (balance *big.Int, err error) {
+	balance = big.NewInt(0)
+	for _, peer := range swapapi.swap.peers {
+		balance.Add(balance, peer.balance)
+	}
+	return
+}
+
+func (swapapi *API) GetSwapMetrics() (*SwapMetrics, error) {
+	return nil, nil
+}

swarm/swap/chequemanager.go

@@ -0,0 +1,65 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package swap
+
+import (
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/p2p/discover"
+	"github.com/ethereum/go-ethereum/swarm/state"
+)
+
+type ChequeManager struct {
+	stateStore        state.Store
+	serialPerNode     map[discover.NodeID]uint64
+	openDebitCheques  map[discover.NodeID][]*Cheque
+	openCreditCheques map[discover.NodeID][]*Cheque
+}
+
+type Cheque struct {
+	serial       uint64
+	timeout      uint64
+	amount       *big.Int
+	sumCumulated *big.Int
+	beneficiary  discover.NodeID //this should probably be common.Address?
+}
+
+func NewChequeManager(stateStore state.Store) *ChequeManager {
+	return &ChequeManager{
+		stateStore: stateStore,
+		//TODO: restore from state store
+		serialPerNode:     make(map[discover.NodeID]uint64),
+		openDebitCheques:  make(map[discover.NodeID][]*Cheque),
+		openCreditCheques: make(map[discover.NodeID][]*Cheque),
+	}
+}
+
+func (mgr *ChequeManager) CreateCheque(beneficiary discover.NodeID, amount *big.Int) *Cheque {
+	mgr.serialPerNode[beneficiary]++
+	cheque := &Cheque{
+		serial:      mgr.serialPerNode[beneficiary],
+		beneficiary: beneficiary,
+		amount:      amount,
+	}
+	openCheques := mgr.openDebitCheques[beneficiary]
+	if openCheques == nil {
+		openCheques = make([]*Cheque, 0)
+	}
+	openCheques = append(openCheques, cheque)
+	mgr.openDebitCheques[beneficiary] = openCheques
+	return cheque
+}

swarm/swap/protocol.go

@@ -0,0 +1,201 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package swap
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/ethereum/go-ethereum/metrics"
+	"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/rpc"
+	"github.com/ethereum/go-ethereum/swarm/log"
+)
+
+const (
+	IsActiveProtocol = true
+)
+
+type SwapProtocol struct {
+	peersMu sync.RWMutex
+	peers   map[discover.NodeID]*SwapProtocolPeer
+	swap    *Swap
+}
+
+// Peer is the Peer extension for the streaming protocol
+type SwapProtocolPeer struct {
+	*protocols.Peer
+	swapProtocol *SwapProtocol
+}
+
+func NewSwapProtocol(swapAccount *Swap) *SwapProtocol {
+	proto := &SwapProtocol{
+		peers: make(map[discover.NodeID]*SwapProtocolPeer),
+		swap:  swapAccount,
+	}
+	return proto
+}
+
+// NewPeer is the constructor for Peer
+func NewPeer(peer *protocols.Peer, swap *SwapProtocol) *SwapProtocolPeer {
+	p := &SwapProtocolPeer{
+		Peer:         peer,
+		swapProtocol: swap,
+	}
+	return p
+}
+
+type IssueChequeMsg struct {
+	Cheque *Cheque
+}
+
+type RedeemChequeMsg struct {
+}
+
+/////////////////////////////////////////////////////////////////////
+// SECTION: node.Service interface
+/////////////////////////////////////////////////////////////////////
+
+func (p *SwapProtocol) Start(srv *p2p.Server) error {
+	log.Debug("Started swap")
+	return nil
+}
+
+func (p *SwapProtocol) Stop() error {
+	log.Info("swap shutting down")
+	return nil
+}
+
+var swapSpec = &protocols.Spec{
+	Name:       swapProtocolName,
+	Version:    swapVersion,
+	MaxMsgSize: defaultMaxMsgSize,
+	Messages: []interface{}{
+		SwapMsg{},
+	},
+}
+
+func (p *SwapProtocol) Protocols() []p2p.Protocol {
+	return []p2p.Protocol{
+		{
+			Name:    swapSpec.Name,
+			Version: swapSpec.Version,
+			Length:  swapSpec.Length(),
+			Run:     p.Run,
+		},
+	}
+}
+
+func (p *SwapProtocol) APIs() []rpc.API {
+	apis := []rpc.API{
+		{
+			Namespace: "swap",
+			Version:   "1.0",
+			Service:   NewAPI(p),
+			Public:    true,
+		},
+	}
+	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
+}
+
+func (swap *SwapProtocol) PeerInfo(id discover.NodeID) interface{} {
+	return nil
+}
+
+//------------------------------------------------------------------------------------------
+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 *SwapProtocolPeer) {
+	swap.peersMu.Lock()
+	defer swap.peersMu.Unlock()
+
+	swap.peers[peer.ID()] = peer
+	metrics.GetOrRegisterGauge("registry.peers", nil).Update(int64(len(swap.peers)))
+}
+
+func (swap *SwapProtocol) deletePeer(peer *SwapProtocolPeer) {
+	swap.peersMu.Lock()
+	defer swap.peersMu.Unlock()
+
+	delete(swap.peers, peer.ID())
+	metrics.GetOrRegisterGauge("registry.peers", nil).Update(int64(len(swap.peers)))
+	swap.peersMu.Unlock()
+}
+
+func (swap *SwapProtocol) peersCount() (c int) {
+	swap.peersMu.Lock()
+	c = len(swap.peers)
+	swap.peersMu.Unlock()
+	return
+}
+
+func (p *SwapProtocolPeer) handleSwapMsg(ctx context.Context, msg interface{}) error {
+	switch msg := msg.(type) {
+
+	case *IssueChequeMsg:
+		return p.handleIssueChequeMsg(ctx, msg)
+
+	case *RedeemChequeMsg:
+		return p.handleRedeemChequeMsg(ctx, msg)
+
+	/*
+		case *QuitMsg:
+			return p.handleQuitMsg(msg)
+	*/
+
+	default:
+		return fmt.Errorf("unknown message type: %T", msg)
+	}
+	return nil
+}
+
+func (sp *SwapProtocolPeer) handleIssueChequeMsg(ctx context.Context, msg interface{}) (err error) {
+	log.Debug("SwapProtocolPeer: handleIssueChequeMsg")
+	return err
+}
+
+func (sp *SwapProtocolPeer) handleRedeemChequeMsg(ctx context.Context, msg interface{}) (err error) {
+	log.Debug("SwapProtocolPeer: handleRedeemChequeMsg")
+	return err
+}

swarm/swap/swap.go

@@ -0,0 +1,587 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package swap
+
+import (
+	"context"
+	"crypto/ecdsa"
+	"errors"
+	"fmt"
+	"math/big"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/contracts/chequebook"
+	"github.com/ethereum/go-ethereum/contracts/chequebook/contract"
+	"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"
+)
+
+const (
+	defaultMaxMsgSize = 1024 * 1024
+	swapProtocolName  = "swap"
+	swapVersion       = 1
+)
+
+var (
+	autoCashInterval     = 300 * time.Second           // default interval for autocash
+	autoCashThreshold    = big.NewInt(50000000000000)  // threshold that triggers autocash (wei)
+	autoDepositInterval  = 300 * time.Second           // default interval for autocash
+	autoDepositThreshold = big.NewInt(50000000000000)  // threshold that triggers autodeposit (wei)
+	autoDepositBuffer    = big.NewInt(100000000000000) // buffer that is surplus for fork protection etc (wei)
+	buyAt                = big.NewInt(20000000000)     // maximum chunk price host is willing to pay (wei)
+	sellAt               = big.NewInt(20000000000)     // minimum chunk price host requires (wei)
+	payAt                = big.NewInt(-4096 * 10000)   // threshold that triggers payment {request} (bytes)
+	dropAt               = big.NewInt(-4096 * 12000)   // threshold that triggers disconnect (bytes)
+
+	ErrInsufficientFunds = errors.New("Insufficient funds")
+	ErrNotAccountedMsg   = errors.New("Message does not need accounting")
+)
+
+const (
+	chequebookDeployRetries = 5
+	chequebookDeployDelay   = 1 * time.Second // delay between retries
+)
+
+// SwAP Swarm Accounting Protocol with
+//      Swift Automatic  Payments
+// a peer to peer micropayment system
+type Swap struct {
+	chequeManager *ChequeManager
+	stateStore    state.Store
+	lock          sync.RWMutex
+	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 (
+	DebitEntry  EntryDirection = true
+	CreditEntry EntryDirection = false
+)
+
+type SwapAccountedMsgType interface {
+	GetMsgPrice() *big.Int
+}
+
+//Handler for received messages
+func (sp *SwapPeer) RunAccountedProtocol(protocolHandler func(ctx context.Context, msg interface{}) error) error {
+	//the `peer.Run` function is a loop, so in order to pre-/post-process a message with accounting,
+	//we need to save the actual handler
+	sp.handlerFunc = protocolHandler
+
+	//then run the handler loop function
+	return sp.Run(sp.handleAccountedMsg)
+}
+
+//get a peer's balance
+func (swap *Swap) GetPeerBalance(peer discover.NodeID) *big.Int {
+	if p, ok := swap.peers[peer]; ok {
+		return p.balance
+	}
+	return nil
+}
+
+//Handle a received message; this is the handler loop function.
+//Check if it needs accounting, and if yes, apply accounting logic:
+//Check for sufficient funds, perform operation, then account
+func (sp *SwapPeer) handleAccountedMsg(ctx context.Context, msg interface{}) error {
+	var err error
+	var price *big.Int
+
+	//the message is one which needs accounting...
+	if _, ok := msg.(SwapAccountedMsgType); ok {
+		//..so first check if there are enough funds for the operation available
+		//(for crediting, this means if we are not essentially "overdrafting", or crossing the threshold)
+		price, err = sp.checkAvailableFunds(ctx, msg, CreditEntry)
+		//if not (or some other error occured), return error
+		if err != nil {
+			//also, if the error is indeed insufficient funds, then disconnect the peer
+			if err == ErrInsufficientFunds {
+				log.Error("Insufficient funds, dropping peer")
+				sp.Drop(err)
+			}
+			return err
+		}
+		//at this point we know there are sufficient funds, so process the message
+		err = sp.handlerFunc(ctx, msg)
+		if err == nil {
+			//and if no errors occurred, finally book the entry
+			sp.AccountMsgForPeer(ctx, msg, price, CreditEntry)
+		}
+	} else {
+		//this message doesn't need accounting, so just process it
+		err = sp.handlerFunc(ctx, msg)
+	}
+	return err
+}
+
+//Send a message
+//Check if it needs accounting, and if yes, apply accounting logic:
+//Check for sufficient funds, perform operation, then account
+func (sp *SwapPeer) Send(ctx context.Context, msg interface{}) error {
+	var err error
+	var price *big.Int
+
+	//the message is one which needs accounting...
+	if _, ok := msg.(SwapAccountedMsgType); ok {
+		//..so first check if there are enough funds for the operation available
+		price, err = sp.checkAvailableFunds(ctx, msg, DebitEntry)
+		//if not (or some other error occured), return error
+		if err != nil {
+			//also, if the error is indeed insufficient funds, then disconnect the peer
+			if err == ErrInsufficientFunds {
+				log.Error("Insufficient funds, dropping peer")
+				sp.Drop(err)
+			}
+			return err
+		}
+		//at this point we know there are sufficient funds, so process the message
+		err = sp.Peer.Send(ctx, msg)
+		if err == nil {
+			//and if no errors occurred, finally book the entry
+			sp.AccountMsgForPeer(ctx, msg, price, DebitEntry)
+		}
+	} else {
+		//this message doesn't need accounting, so just process it
+		err = sp.Peer.Send(ctx, msg)
+	}
+	return err
+}
+
+//check that the operation has enough funds available
+func (sp *SwapPeer) checkAvailableFunds(ctx context.Context, msg interface{}, direction EntryDirection) (*big.Int, error) {
+	sp.lock.Lock()
+	defer sp.lock.Unlock()
+
+	if accounted, ok := msg.(SwapAccountedMsgType); ok {
+		price := accounted.GetMsgPrice()
+		//local node is being credited (in its favor), so check upper limit
+		if direction == CreditEntry {
+			//TODO: is there are a check needed here?
+			//It should actually have been done on the client side, the debitor!
+			//creditor could theoretically go over payAt, but if well done,
+			//should have been checked on the client side so this shouldn't happen?
+			checkBalance := sp.balance.Add(sp.balance, price)
+			//(checkBalance *Int) Cmp(payAt)
+			// -1 if checkBalance  <  payAt
+			//  0 if checkBalance  == payAt
+			// +1 if checkBalance  >  payAt
+			if checkBalance.Cmp(payAt) == 1 {
+				return nil, ErrInsufficientFunds
+			}
+		} else if direction == DebitEntry {
+			//NOTE: ErrInsufficientFunds should only be returned
+			//if the dropAt is exceeded, but should be ignored for payAt,
+			//as there is a "clemency" margin between triggering the check
+			//and actually disconnecting the peer
+
+			//(checkBalance *Int) Cmp(dropAt)
+			// -1 if checkBalance  <  dropAt
+			//  0 if checkBalance  ==	dropAt
+			// +1 if checkBalance  >  dropAt
+			checkBalance := sp.balance.Sub(sp.balance, price.Abs(price))
+			if checkBalance.Cmp(dropAt) == -1 {
+				return nil, ErrInsufficientFunds
+			}
+		}
+		return price, nil
+	}
+	return nil, ErrNotAccountedMsg
+}
+
+//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(ctx context.Context, msg interface{}, price *big.Int, direction EntryDirection) {
+	if _, ok := msg.(SwapAccountedMsgType); ok {
+		sp.lock.Lock()
+		defer sp.lock.Unlock()
+		//local node is being credited (in its favor), so its balance increases
+		if direction == CreditEntry {
+			//NOTE: do we need to check for sufficient funds again?
+			//operations are not atomic/transactional, so balance may have changed in the meanwhile!
+			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)
+		//(sp.balance *Int) Cmp(payAt)
+		// -1 if sp.balance <  payAt
+		//  0 if sp.balance == payAt
+		// +1 if sp.balance >  payAt
+		if sp.balance.Cmp(payAt) == -1 {
+			err := sp.issueCheque(ctx)
+			if err != nil {
+				//TODO: special error handling, as at this point the accounting has been done
+				//but the cheque could not be sent?
+			}
+		}
+		if sp.balance.Cmp(dropAt) == -1 {
+			sp.Drop(ErrInsufficientFunds)
+		}
+		log.Debug(fmt.Sprintf("balance for peer %s: %s", sp.ID(), sp.balance.String()))
+	}
+}
+
+func (sp *SwapPeer) issueCheque(ctx context.Context) error {
+	amount := big.NewInt(0)
+	cheque := sp.swapAccount.chequeManager.CreateCheque(sp.ID(), amount.Abs(payAt))
+	msg := IssueChequeMsg{
+		Cheque: cheque,
+	}
+	return sp.Send(ctx, msg)
+}
+
+//Create a new swap accounted peer
+func NewSwapPeer(peer *protocols.Peer, swap *Swap) *SwapPeer {
+	balance := big.NewInt(0)
+	//check if there is one already in the stateStore and load it
+	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 {
+	BuyAt  *big.Int // accepted max price for chunk
+	SellAt *big.Int // offered sale price for chunk
+	PayAt  *big.Int // threshold that triggers payment request
+	DropAt *big.Int // threshold that triggers disconnect
+}
+
+// Strategy encapsulates parameters relating to
+// automatic deposit and automatic cashing
+type Strategy struct {
+	AutoCashInterval     time.Duration // default interval for autocash
+	AutoCashThreshold    *big.Int      // threshold that triggers autocash (wei)
+	AutoDepositInterval  time.Duration // default interval for autocash
+	AutoDepositThreshold *big.Int      // threshold that triggers autodeposit (wei)
+	AutoDepositBuffer    *big.Int      // buffer that is surplus for fork protection etc (wei)
+}
+
+// SwapMsg encapsulates messages transported over pss.
+type SwapMsg struct {
+	To      []byte
+	Control []byte
+	Expire  uint32
+	Payload *whisper.Envelope
+}
+
+// Params extends the public profile with private parameters relating to
+// automatic deposit and automatic cashing
+type Params struct {
+	*Profile
+	*Strategy
+}
+
+// LocalProfile combines a PayProfile with *swap.Params
+type LocalProfile struct {
+	*Params
+	*PayProfile
+}
+
+// RemoteProfile combines a PayProfile with *swap.Profile
+type RemoteProfile struct {
+	*Profile
+	*PayProfile
+}
+
+// PayProfile is a container for relevant chequebook and beneficiary options
+type PayProfile struct {
+	PublicKey   string         // check against signature of promise
+	Contract    common.Address // address of chequebook contract
+	Beneficiary common.Address // recipient address for swarm sales revenue
+	privateKey  *ecdsa.PrivateKey
+	publicKey   *ecdsa.PublicKey
+	owner       common.Address
+	chbook      *chequebook.Chequebook
+	lock        sync.RWMutex
+}
+
+// New - swap constructor
+func NewSwap(local *Params, stateStore state.Store) (swap *Swap, err error) {
+
+	swap = &Swap{
+		chequeManager: NewChequeManager(stateStore),
+		local:         local,
+		stateStore:    stateStore,
+		peers:         make(map[discover.NodeID]*SwapPeer),
+	}
+
+	//swap.SetParams(local)
+	return
+}
+
+// NewDefaultSwapParams create params with default values
+func NewDefaultSwapParams() *LocalProfile {
+	return &LocalProfile{
+		PayProfile: &PayProfile{},
+		Params: &Params{
+			Profile: &Profile{
+				BuyAt:  buyAt,
+				SellAt: sellAt,
+				PayAt:  payAt,
+				DropAt: dropAt,
+			},
+			Strategy: &Strategy{
+				AutoCashInterval:     autoCashInterval,
+				AutoCashThreshold:    autoCashThreshold,
+				AutoDepositInterval:  autoDepositInterval,
+				AutoDepositThreshold: autoDepositThreshold,
+				AutoDepositBuffer:    autoDepositBuffer,
+			},
+		},
+	}
+}
+
+// Init this can only finally be set after all config options (file, cmd line, env vars)
+// have been evaluated
+func (lp *LocalProfile) Init(contract common.Address, prvkey *ecdsa.PrivateKey) {
+	pubkey := &prvkey.PublicKey
+
+	lp.PayProfile = &PayProfile{
+		PublicKey:   common.ToHex(crypto.FromECDSAPub(pubkey)),
+		Contract:    contract,
+		Beneficiary: crypto.PubkeyToAddress(*pubkey),
+		privateKey:  prvkey,
+		publicKey:   pubkey,
+		owner:       crypto.PubkeyToAddress(*pubkey),
+	}
+}
+
+// Chequebook get's chequebook from the localProfile
+func (lp *LocalProfile) Chequebook() *chequebook.Chequebook {
+	defer lp.lock.Unlock()
+	lp.lock.Lock()
+	return lp.chbook
+}
+
+// PrivateKey accessor
+func (lp *LocalProfile) PrivateKey() *ecdsa.PrivateKey {
+	return lp.privateKey
+}
+
+// func (self *LocalProfile) PublicKey() *ecdsa.PublicKey {
+// 	return self.publicKey
+// }
+
+// SetKey set's private and public key on localProfile
+func (lp *LocalProfile) SetKey(prvkey *ecdsa.PrivateKey) {
+	lp.privateKey = prvkey
+	lp.publicKey = &prvkey.PublicKey
+}
+
+// SetChequebook wraps the chequebook initialiser and sets up autoDeposit to cover spending.
+func (lp *LocalProfile) SetChequebook(ctx context.Context, backend chequebook.Backend, path string) error {
+	lp.lock.Lock()
+	swapContract := lp.Contract
+	lp.lock.Unlock()
+
+	valid, err := chequebook.ValidateCode(ctx, backend, swapContract)
+	if err != nil {
+		return err
+	} else if valid {
+		return lp.newChequebookFromContract(path, backend)
+	}
+	return lp.deployChequebook(ctx, backend, path)
+}
+
+// deployChequebook deploys the localProfile Chequebook
+func (lp *LocalProfile) deployChequebook(ctx context.Context, backend chequebook.Backend, path string) error {
+	opts := bind.NewKeyedTransactor(lp.privateKey)
+	opts.Value = lp.AutoDepositBuffer
+	opts.Context = ctx
+
+	log.Info(fmt.Sprintf("Deploying new chequebook (owner: %v)", opts.From.Hex()))
+	address, err := deployChequebookLoop(opts, backend)
+	if err != nil {
+		log.Error(fmt.Sprintf("unable to deploy new chequebook: %v", err))
+		return err
+	}
+	log.Info(fmt.Sprintf("new chequebook deployed at %v (owner: %v)", address.Hex(), opts.From.Hex()))
+
+	// need to save config at this point
+	lp.lock.Lock()
+	lp.Contract = address
+	err = lp.newChequebookFromContract(path, backend)
+	lp.lock.Unlock()
+	if err != nil {
+		log.Warn(fmt.Sprintf("error initialising cheque book (owner: %v): %v", opts.From.Hex(), err))
+	}
+	return err
+}
+
+// deployChequebookLoop repeatedly tries to deploy a chequebook.
+func deployChequebookLoop(opts *bind.TransactOpts, backend chequebook.Backend) (addr common.Address, err error) {
+	var tx *types.Transaction
+	for try := 0; try < chequebookDeployRetries; try++ {
+		if try > 0 {
+			time.Sleep(chequebookDeployDelay)
+		}
+		if _, tx, _, err = contract.DeployChequebook(opts, backend); err != nil {
+			log.Warn(fmt.Sprintf("can't send chequebook deploy tx (try %d): %v", try, err))
+			continue
+		}
+		if addr, err = bind.WaitDeployed(opts.Context, backend, tx); err != nil {
+			log.Warn(fmt.Sprintf("chequebook deploy error (try %d): %v", try, err))
+			continue
+		}
+		return addr, nil
+	}
+	return addr, err
+}
+
+// newChequebookFromContract - initialise the chequebook from a persisted json file or create a new one
+// caller holds the lock
+func (lp *LocalProfile) newChequebookFromContract(path string, backend chequebook.Backend) error {
+	hexkey := common.Bytes2Hex(lp.Contract.Bytes())
+	err := os.MkdirAll(filepath.Join(path, "chequebooks"), os.ModePerm)
+	if err != nil {
+		return fmt.Errorf("unable to create directory for chequebooks: %v", err)
+	}
+
+	chbookpath := filepath.Join(path, "chequebooks", hexkey+".json")
+	lp.chbook, err = chequebook.LoadChequebook(chbookpath, lp.privateKey, backend, true)
+
+	if err != nil {
+		lp.chbook, err = chequebook.NewChequebook(chbookpath, lp.Contract, lp.privateKey, backend)
+		if err != nil {
+			log.Warn(fmt.Sprintf("unable to initialise chequebook (owner: %v): %v", lp.owner.Hex(), err))
+			return fmt.Errorf("unable to initialise chequebook (owner: %v): %v", lp.owner.Hex(), err)
+		}
+	}
+
+	lp.chbook.AutoDeposit(lp.AutoDepositInterval, lp.AutoDepositThreshold, lp.AutoDepositBuffer)
+	log.Info(fmt.Sprintf("auto deposit ON for %v -> %v: interval = %v, threshold = %v, buffer = %v)", crypto.PubkeyToAddress(*(lp.publicKey)).Hex()[:8], lp.Contract.Hex()[:8], lp.AutoDepositInterval, lp.AutoDepositThreshold, lp.AutoDepositBuffer))
+
+	return nil
+}
+
+/*
+// Add (n)
+// n > 0 called when promised/provided n units of service
+// n < 0 called when used/requested n units of service
+func (swap *Swap) Add(n int) error {
+	//defer swap.lock.Unlock()
+	//swap.lock.Lock()
+	swap.balance += n
+	if !swap.Sells && swap.balance > 0 {
+		log.Trace(fmt.Sprintf("<%v> remote peer cannot have debt (balance: %v)", swap.proto, swap.balance))
+		swap.proto.Drop()
+		return fmt.Errorf("[SWAP] <%v> remote peer cannot have debt (balance: %v)", swap.proto, swap.balance)
+	}
+	if !swap.Buys && swap.balance < 0 {
+		log.Trace(fmt.Sprintf("<%v> we cannot have debt (balance: %v)", swap.proto, swap.balance))
+		return fmt.Errorf("[SWAP] <%v> we cannot have debt (balance: %v)", swap.proto, swap.balance)
+	}
+	if swap.balance >= int(swap.local.DropAt) {
+		log.Trace(fmt.Sprintf("<%v> remote peer has too much debt (balance: %v, disconnect threshold: %v)", swap.proto, swap.balance, swap.local.DropAt))
+		swap.proto.Drop()
+		return fmt.Errorf("[SWAP] <%v> remote peer has too much debt (balance: %v, disconnect threshold: %v)", swap.proto, swap.balance, swap.local.DropAt)
+	} else if swap.balance <= -int(swap.remote.PayAt) {
+		swap.send()
+	}
+	return nil
+}
+
+// Balance accessor
+func (swap *Swap) Balance() int {
+	//defer swap.lock.Unlock()
+	//swap.lock.Lock()
+	return swap.balance
+}
+
+/*
+// send (units) is called when payment is due
+// In case of insolvency no promise is issued and sent, safe against fraud
+// No return value: no error = payment is opportunistic = hang in till dropped
+func (swap *Swap) send() {
+	if swap.local.BuyAt != nil && swap.balance < 0 {
+		amount := big.NewInt(int64(-swap.balance))
+		amount.Mul(amount, swap.remote.SellAt)
+		promise, err := swap.Out.Issue(amount)
+		if err != nil {
+			log.Warn(fmt.Sprintf("<%v> cannot issue cheque (amount: %v, channel: %v): %v", swap.proto, amount, swap.Out, err))
+		} else {
+			log.Warn(fmt.Sprintf("<%v> cheque issued (amount: %v, channel: %v)", swap.proto, amount, swap.Out))
+			swap.proto.Pay(-swap.balance, promise)
+			swap.balance = 0
+		}
+	}
+}
+
+// Receive (units, promise) is called by the protocol when a payment msg is received
+// returns error if promise is invalid.
+func (swap *Swap) Receive(units int, promise Promise) error {
+	if units <= 0 {
+		return fmt.Errorf("invalid units: %v <= 0", units)
+	}
+
+	price := new(big.Int).SetInt64(int64(units))
+	price.Mul(price, swap.local.SellAt)
+
+	amount, err := swap.In.Receive(promise)
+
+	if err != nil {
+		err = fmt.Errorf("invalid promise: %v", err)
+	} else if price.Cmp(amount) != 0 {
+		// verify amount = units * unit sale price
+		return fmt.Errorf("invalid amount: %v = %v * %v (units sent in msg * agreed sale unit price) != %v (signed in cheque)", price, units, swap.local.SellAt, amount)
+	}
+	if err != nil {
+		log.Trace(fmt.Sprintf("<%v> invalid promise (amount: %v, channel: %v): %v", swap.proto, amount, swap.In, err))
+		return err
+	}
+
+	// credit remote peer with units
+	swap.Add(-units)
+	log.Trace(fmt.Sprintf("<%v> received promise (amount: %v, channel: %v): %v", swap.proto, amount, swap.In, promise))
+
+	return nil
+}
+*/

swarm/swap/swap_test.go

@@ -0,0 +1,290 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package swap
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"math/big"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/p2p/protocols"
+	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethereum/go-ethereum/swarm/state"
+	colorable "github.com/mattn/go-colorable"
+)
+
+var (
+	p2pPort       = 30100
+	ipcpath       = ".swarm.ipc"
+	datadirPrefix = ".data_"
+	stackW        = &sync.WaitGroup{}
+	loglevel      = flag.Int("loglevel", 2, "verbosity of logs")
+)
+
+var testSpec = &protocols.Spec{
+	Name:       "swapTestSpec",
+	Version:    1,
+	MaxMsgSize: 10 * 1024 * 1024,
+	Messages: []interface{}{
+		testExceedsPayAtMsg{},
+		testExceedsDropAtMsg{},
+	},
+}
+
+type dummyRW struct{}
+
+func (d *dummyRW) WriteMsg(msg p2p.Msg) error {
+	return nil
+}
+
+func (d *dummyRW) ReadMsg() (p2p.Msg, error) {
+	return p2p.Msg{
+		Code:       0,
+		Size:       0,
+		Payload:    nil,
+		ReceivedAt: time.Now(),
+	}, nil
+}
+
+type testExceedsPayAtMsg struct{}
+type testExceedsDropAtMsg struct{}
+
+func (tmsg *testExceedsPayAtMsg) GetMsgPrice() *big.Int {
+	diff := &big.Int{}
+	return diff.Sub(payAt, big.NewInt(1))
+}
+
+func (tmsg *testExceedsDropAtMsg) GetMsgPrice() *big.Int {
+	diff := &big.Int{}
+	return diff.Sub(dropAt, big.NewInt(1))
+}
+
+func init() {
+	flag.Parse()
+
+	log.PrintOrigins(true)
+	log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(colorable.NewColorableStderr(), log.TerminalFormat(true))))
+}
+
+func TestLimits(t *testing.T) {
+	if dropAt.Cmp(payAt) > -1 {
+		t.Fatal(fmt.Sprintf("dropAt limit is not lower than payAt limit, dropAt: %s, payAt: %s", dropAt.String(), payAt.String()))
+	}
+}
+
+//unit test for exceeds pay limit
+func TestExceedsPayAt(t *testing.T) {
+	swap, testDir := createTestSwap(t)
+	defer os.RemoveAll(testDir)
+
+	testPeer := newDummyPeer()
+	sp := NewSwapPeer(testPeer, swap)
+	sp.handlerFunc = dummyMsgHandler
+
+	ctx := context.Background()
+	sp.Send(ctx, &testExceedsPayAtMsg{})
+
+	cheques := sp.swapAccount.chequeManager.openDebitCheques[sp.ID()]
+	if cheques == nil {
+		t.Fatal("Expected cheques for this peer to be present, but are nil")
+	}
+	if len(cheques) == 0 {
+		t.Fatal("Expected a cheque to have arrived, but len is zero")
+	}
+	if cheques[0].serial != 1 {
+		t.Fatal(fmt.Sprintf("Expected the serial to be one (first message but is: %d", cheques[0].serial))
+	}
+	absPayAt := big.NewInt(0)
+	if cheques[0].amount.Cmp(absPayAt.Abs(payAt)) != 0 {
+		t.Fatal(fmt.Sprintf("Expected the serial to be one (first message but is: %d", cheques[0].serial))
+	}
+}
+
+//unit test for exceeds drop limit
+func TestExceedsDropAt(t *testing.T) {
+	swap, testDir := createTestSwap(t)
+	defer os.RemoveAll(testDir)
+
+	testPeer := newDummyPeer()
+	sp := NewSwapPeer(testPeer, swap)
+	sp.handlerFunc = dummyMsgHandler
+
+	ctx := context.Background()
+	err := sp.Send(ctx, &testExceedsDropAtMsg{})
+	if err != ErrInsufficientFunds {
+		t.Fatal("Expected test to fail with insufficient funds, but it didn't")
+	}
+}
+
+func createTestSwap(t *testing.T) (*Swap, string) {
+	dir, err := ioutil.TempDir("", "swap_test_store")
+	if err != nil {
+		t.Fatal(err)
+	}
+	stateStore, err2 := state.NewDBStore(dir)
+	if err2 != nil {
+		t.Fatal(err2)
+	}
+	swap, err3 := NewSwap(NewDefaultSwapParams().Params, stateStore)
+	if err3 != nil {
+		t.Fatal(err3)
+	}
+	return swap, dir
+}
+
+func runProtocol(peer *protocols.Peer, swap *Swap) {
+	sp := NewSwapPeer(peer, swap)
+	sp.Peer.Run(dummyMsgHandler)
+}
+
+func dummyMsgHandler(ctx context.Context, msg interface{}) error {
+	return nil
+}
+
+func TestSwapRPC(t *testing.T) {
+	swap, testDir := createTestSwap(t)
+	defer os.RemoveAll(testDir)
+
+	// create the two nodes
+	stack_one, err := newServiceNode(p2pPort, 0, 0)
+	if err != nil {
+		t.Fatal("Create servicenode #1 fail", "err", err)
+	}
+
+	instance := NewSwapProtocol(swap)
+	// wrapper function for servicenode to start the service
+	swapsvc := func(ctx *node.ServiceContext) (node.Service, error) {
+		return &API{
+			SwapProtocol: instance,
+		}, nil
+	}
+
+	// register adds the service to the services the servicenode starts when started
+	err = stack_one.Register(swapsvc)
+	if err != nil {
+		t.Fatal("Register service in servicenode #1 fail", "err", err)
+	}
+	// start the nodes
+	err = stack_one.Start()
+	if err != nil {
+		t.Fatal("servicenode #1 start failed", "err", err)
+	}
+
+	// connect to the servicenode RPCs
+	rpcclient_one, err := rpc.Dial(filepath.Join(stack_one.DataDir(), ipcpath))
+	if err != nil {
+		t.Fatal("connect to servicenode #1 IPC fail", "err", err)
+	}
+	defer os.RemoveAll(stack_one.DataDir())
+
+	var balance *big.Int
+	err = rpcclient_one.Call(&balance, "swap_balance")
+	if err != nil {
+		t.Fatal("servicenode #1 RPC failed", "err", err)
+	}
+	log.Debug("servicenode #1 balance", "balance-1", balance)
+
+	if balance.Cmp(big.NewInt(0)) != 0 {
+		t.Fatal("Expected balance to be 0 but it is not")
+	}
+
+	dummyPeer1 := newDummyPeer()
+	dummyPeer2 := newDummyPeer()
+	id1 := dummyPeer1.ID()
+	id2 := dummyPeer2.ID()
+
+	fake1 := int64(234)
+	fake2 := int64(-100)
+	fakeBalance1 := big.NewInt(fake1)
+	fakeBalance2 := big.NewInt(fake2)
+
+	swap.peers[id1] = NewSwapPeer(dummyPeer1, swap)
+	swap.peers[id2] = NewSwapPeer(dummyPeer2, swap)
+	swap.peers[id1].balance = fakeBalance1
+	swap.peers[id2].balance = fakeBalance2
+
+	err = rpcclient_one.Call(&balance, "swap_balanceWithPeer", id1)
+	if err != nil {
+		t.Fatal("servicenode #1 RPC failed", "err", err)
+	}
+	log.Debug("balance1", "balance-1", balance)
+	if balance.Cmp(fakeBalance1) != 0 {
+		t.Fatal(fmt.Sprintf("Expected balance %s to be equal to fake balance %s, but it is not", balance.String(), fakeBalance1.String()))
+	}
+
+	err = rpcclient_one.Call(&balance, "swap_balanceWithPeer", id2)
+	if err != nil {
+		t.Fatal("servicenode #1 RPC failed", "err", err)
+	}
+	log.Debug("balance2", "balance-2", balance)
+	if balance.Cmp(fakeBalance2) != 0 {
+		t.Fatal(fmt.Sprintf("Expected balance %s to be equal to fake balance %s, but it is not", balance.String(), fakeBalance2.String()))
+	}
+
+	err = rpcclient_one.Call(&balance, "swap_balance")
+	if err != nil {
+		t.Fatal("servicenode #1 RPC failed", "err", err)
+	}
+	log.Debug("balance", "balance", balance)
+
+	fakeSum := big.NewInt(fake1 + fake2)
+	if balance.Cmp(fakeSum) != 0 {
+		t.Fatal(fmt.Sprintf("Expected balance %s to be equal to sum %s, but it is not", balance.String(), fakeSum.String()))
+	}
+}
+
+func newDummyPeer() *protocols.Peer {
+	id := adapters.RandomNodeConfig().ID
+	return protocols.NewPeer(p2p.NewPeer(id, "testPeer", nil), &dummyRW{}, testSpec)
+}
+
+func newServiceNode(port int, httpport int, wsport int, modules ...string) (*node.Node, error) {
+	cfg := &node.DefaultConfig
+	cfg.P2P.ListenAddr = fmt.Sprintf(":%d", port)
+	cfg.P2P.EnableMsgEvents = true
+	cfg.P2P.NoDiscovery = true
+	cfg.IPCPath = ipcpath
+	cfg.DataDir = fmt.Sprintf("%s%d", datadirPrefix, port)
+	if httpport > 0 {
+		cfg.HTTPHost = node.DefaultHTTPHost
+		cfg.HTTPPort = httpport
+	}
+	if wsport > 0 {
+		cfg.WSHost = node.DefaultWSHost
+		cfg.WSPort = wsport
+		cfg.WSOrigins = []string{"*"}
+		for i := 0; i < len(modules); i++ {
+			cfg.WSModules = append(cfg.WSModules, modules[i])
+		}
+	}
+	stack, err := node.New(cfg)
+	if err != nil {
+		return nil, fmt.Errorf("ServiceNode create fail: %v", err)
+	}
+	return stack, nil
+}

swarm/swarm.go

@@ -51,6 +51,7 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/storage"
 	"github.com/ethereum/go-ethereum/swarm/storage/mock"
 	"github.com/ethereum/go-ethereum/swarm/storage/mru"
+	"github.com/ethereum/go-ethereum/swarm/swap"
 	"github.com/ethereum/go-ethereum/swarm/tracing"
 )
 
@@ -78,6 +79,7 @@ type Swarm struct {
 	lstore      *storage.LocalStore // local store, needs to store for releasing resources after node stopped
 	sfs         *fuse.SwarmFS       // need this to cleanup all the active mounts on node exit
 	ps          *pss.Pss
+	swap        *swap.Swap
 
 	tracerClose io.Closer
 }
@@ -178,7 +180,12 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e
 	)
 	delivery := stream.NewDelivery(to, db)
 
-	self.streamer = stream.NewRegistry(addr, delivery, db, stateStore, &stream.RegistryOptions{
+	self.swap, err = swap.NewSwap(swap.NewDefaultSwapParams().Params, stateStore)
+	if err != nil {
+		return nil, err
+	}
+
+	self.streamer = stream.NewRegistry(addr, delivery, db, stateStore, self.swap, &stream.RegistryOptions{
 		SkipCheck:       config.DeliverySkipCheck,
 		DoSync:          config.SyncEnabled,
 		DoRetrieve:      true,
@@ -363,6 +370,15 @@ func (self *Swarm) Start(srv *p2p.Server) error {
 	}
 	log.Info(fmt.Sprintf("Swarm network started on bzz address: %x", self.bzz.Hive.Overlay.BaseAddr()))
 
+	/*
+		err = self.swap.Start(srv)
+		if err != nil {
+			log.Error("swap failed", "err", err)
+			return err
+		}
+		log.Debug("Swap accounting initialized")
+	*/
+
 	if self.ps != nil {
 		self.ps.Start(srv)
 		log.Info("Pss started")