diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go
index 491dc9fd58..0559a942ea 100644
--- a/swarm/network/stream/common_test.go
+++ b/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()
diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go
index 36040339d3..4ed7a25c69 100644
--- a/swarm/network/stream/delivery.go
+++ b/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)
diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go
index 972cc859a3..7b2c1f9dcf 100644
--- a/swarm/network/stream/delivery_test.go
+++ b/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,
diff --git a/swarm/network/stream/intervals_test.go b/swarm/network/stream/intervals_test.go
index f4294134b9..02626ab65b 100644
--- a/swarm/network/stream/intervals_test.go
+++ b/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)
diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go
index 80b9ab711a..ce861b67d4 100644
--- a/swarm/network/stream/peer.go
+++ b/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),
diff --git a/swarm/network/stream/snapshot_retrieval_test.go b/swarm/network/stream/snapshot_retrieval_test.go
index 4ff947b215..32c1f63c23 100644
--- a/swarm/network/stream/snapshot_retrieval_test.go
+++ b/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,
})
diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go
index 6acab50af4..1b264011e7 100644
--- a/swarm/network/stream/snapshot_sync_test.go
+++ b/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())
diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go
index cd0580a0c0..9d559eeefd 100644
--- a/swarm/network/stream/stream.go
+++ b/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
diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go
index f72aa34441..e9ee4df458 100644
--- a/swarm/network/stream/syncer_test.go
+++ b/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,
})
diff --git a/swarm/network_test.go b/swarm/network_test.go
index 176c635d82..1da677584e 100644
--- a/swarm/network_test.go
+++ b/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")
- longrunning = flag.Bool("longrunning", false, "do run long-running tests")
- waitKademlia = flag.Bool("waitkademlia", false, "wait for healthy kademlia before checking files availability")
+ loglevel = flag.Int("loglevel", 2, "verbosity of logs")
+ longrunning = flag.Bool("longrunning", false, "do run long-running tests")
+ 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) {
diff --git a/swarm/swap/api.go b/swarm/swap/api.go
new file mode 100644
index 0000000000..960530e870
--- /dev/null
+++ b/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 .
+
+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
+}
diff --git a/swarm/swap/chequemanager.go b/swarm/swap/chequemanager.go
new file mode 100644
index 0000000000..a9e4494b0a
--- /dev/null
+++ b/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 .
+
+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
+}
diff --git a/swarm/swap/protocol.go b/swarm/swap/protocol.go
new file mode 100644
index 0000000000..9863ce20f4
--- /dev/null
+++ b/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 .
+
+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
+}
diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go
new file mode 100644
index 0000000000..3334ab9905
--- /dev/null
+++ b/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 .
+
+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
+}
+*/
diff --git a/swarm/swap/swap_test.go b/swarm/swap/swap_test.go
new file mode 100644
index 0000000000..99408d9092
--- /dev/null
+++ b/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 .
+
+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
+}
diff --git a/swarm/swarm.go b/swarm/swarm.go
index a895bdfa55..30db6e1528 100644
--- a/swarm/swarm.go
+++ b/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")