diff --git a/swarm/swap/api.go b/swarm/swap/api.go new file mode 100644 index 0000000000..2de11d9efa --- /dev/null +++ b/swarm/swap/api.go @@ -0,0 +1,62 @@ +// 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" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/hexutil" +) + +// 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 { + *Swap +} + +type SwapMetrics struct { +} + +type Cheque struct { +} + +func NewAPI(swap *Swap) *API { + return &API{Swap: swap} +} + +func (swapapi *API) Balance(ctx context.Context) (balance int, err error) { + balance = 0 + err = nil + return +} + +func (swapapi *API) GetSwapMetrics() (*SwapMetrics, error) { + return nil, nil +} + +func (swapapi *API) IssueCheque(recipient *common.Address) (*Cheque, error) { + return nil, nil +} + +func (swapapi *API) RedeemCheque(cheque *Cheque) { +} diff --git a/swarm/swap/protocol.go b/swarm/swap/protocol.go new file mode 100644 index 0000000000..63acbbe929 --- /dev/null +++ b/swarm/swap/protocol.go @@ -0,0 +1,177 @@ +// 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 . + +// +build !nopssprotocol + +package swap + +import ( + "sync" + "time" + + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/protocols" + "github.com/ethereum/go-ethereum/rlp" +) + +const ( + IsActiveProtocol = true +) + +// Convenience wrapper for devp2p protocol messages for transport over pss +type ProtocolMsg struct { + Code uint64 + Size uint32 + Payload []byte + ReceivedAt time.Time +} + +// Creates a ProtocolMsg +func NewProtocolMsg(code uint64, msg interface{}) ([]byte, error) { + + rlpdata, err := rlp.EncodeToBytes(msg) + if err != nil { + return nil, err + } + + // TODO verify that nested structs cannot be used in rlp + smsg := &ProtocolMsg{ + Code: code, + Size: uint32(len(rlpdata)), + Payload: rlpdata, + } + + return rlp.EncodeToBytes(smsg) +} + +// Protocol options to be passed to a new Protocol instance +// +// The parameters specify which encryption schemes to allow +type ProtocolParams struct { +} + +// Convenience object for emulation devp2p over pss +type Protocol struct { + proto *p2p.Protocol + spec *protocols.Spec + RWPoolMu sync.Mutex +} + +/* +// Activates devp2p emulation over a specific pss topic +// +// One or both encryption schemes must be specified. If +// only one is specified, the protocol will not be valid +// for the other, and will make the message handler +// return errors +func RegisterProtocol(ps *Pss, topic *Topic, spec *protocols.Spec, targetprotocol *p2p.Protocol, options *ProtocolParams) (*Protocol, error) { + if !options.Asymmetric && !options.Symmetric { + return nil, fmt.Errorf("specify at least one of asymmetric or symmetric messaging mode") + } + pp := &Protocol{ + Pss: ps, + proto: targetprotocol, + topic: topic, + spec: spec, + pubKeyRWPool: make(map[string]p2p.MsgReadWriter), + symKeyRWPool: make(map[string]p2p.MsgReadWriter), + Asymmetric: options.Asymmetric, + Symmetric: options.Symmetric, + } + return pp, nil +} + +*/ +// Generic handler for incoming messages over devp2p emulation +// +// To be passed to pss.Register() +// +// Will run the protocol on a new incoming peer, provided that +// the encryption key of the message has a match in the internal +// pss keypool +// +// Fails if protocol is not valid for the message encryption scheme, +// if adding a new peer fails, or if the message is not a serialized +// p2p.Msg (which it always will be if it is sent from this object). +func (p *Protocol) Handle(msg []byte, peer *p2p.Peer, asymmetric bool, keyid string) error { + return nil +} + +/* +// Runs an emulated pss Protocol on the specified peer, +// linked to a specific topic +// `key` and `asymmetric` specifies what encryption key +// to link the peer to. +// The key must exist in the pss store prior to adding the peer. +func (p *Protocol) AddPeer(peer *p2p.Peer, topic Topic, asymmetric bool, key string) (p2p.MsgReadWriter, error) { + rw := &PssReadWriter{ + Pss: p.Pss, + rw: make(chan p2p.Msg), + spec: p.spec, + topic: p.topic, + key: key, + } + if asymmetric { + rw.sendFunc = p.Pss.SendAsym + } else { + rw.sendFunc = p.Pss.SendSym + } + if asymmetric { + p.Pss.pubKeyPoolMu.Lock() + if _, ok := p.Pss.pubKeyPool[key]; !ok { + return nil, fmt.Errorf("asym key does not exist: %s", key) + } + p.Pss.pubKeyPoolMu.Unlock() + p.RWPoolMu.Lock() + p.pubKeyRWPool[key] = rw + p.RWPoolMu.Unlock() + } else { + p.Pss.symKeyPoolMu.Lock() + if _, ok := p.Pss.symKeyPool[key]; !ok { + return nil, fmt.Errorf("symkey does not exist: %s", key) + } + p.Pss.symKeyPoolMu.Unlock() + p.RWPoolMu.Lock() + p.symKeyRWPool[key] = rw + p.RWPoolMu.Unlock() + } + go func() { + err := p.proto.Run(peer, rw) + log.Warn(fmt.Sprintf("pss vprotocol quit on %v topic %v: %v", peer, topic, err)) + }() + return rw, nil +} + +func (p *Protocol) RemovePeer(asymmetric bool, key string) { + log.Debug("closing pss peer", "asym", asymmetric, "key", key) + p.RWPoolMu.Lock() + defer p.RWPoolMu.Unlock() + if asymmetric { + rw := p.pubKeyRWPool[key].(*PssReadWriter) + rw.closed = true + delete(p.pubKeyRWPool, key) + } else { + rw := p.symKeyRWPool[key].(*PssReadWriter) + rw.closed = true + delete(p.symKeyRWPool, key) + } +} + +// Uniform translation of protocol specifiers to topic +func ProtocolTopic(spec *protocols.Spec) Topic { + return BytesToTopic([]byte(fmt.Sprintf("%s:%d", spec.Name, spec.Version))) +} +*/ diff --git a/swarm/swap/swap.go b/swarm/swap/swap.go new file mode 100644 index 0000000000..7b58331cbf --- /dev/null +++ b/swarm/swap/swap.go @@ -0,0 +1,433 @@ +// 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" + "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" + "github.com/ethereum/go-ethereum/p2p/protocols" + "github.com/ethereum/go-ethereum/rpc" + "github.com/ethereum/go-ethereum/swarm/log" + 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 = 100 // threshold that triggers payment {request} (units) + dropAt = 10000 // threshold that triggers disconnect (units) +) + +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 { + lock sync.Mutex // mutex for balance access + balance int // units of chunk/retrieval request + local *Params // local peer's swap parameters + remote *Profile // remote peer's swap profile +} + +// 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 uint // threshold that triggers payment request + DropAt uint // 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) (swap *Swap, err error) { + + swap = &Swap{ + local: local, + } + + //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: uint(payAt), + DropAt: uint(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), + } +} + +///////////////////////////////////////////////////////////////////// +// SECTION: node.Service interface +///////////////////////////////////////////////////////////////////// + +func (p *Swap) Start(srv *p2p.Server) error { + log.Debug("Started swap") + return nil +} + +func (p *Swap) Stop() error { + log.Info("swap shutting down") + return nil +} + +var swapSpec = &protocols.Spec{ + Name: swapProtocolName, + Version: swapVersion, + MaxMsgSize: defaultMaxMsgSize, + Messages: []interface{}{ + SwapMsg{}, + }, +} + +func (p *Swap) Protocols() []p2p.Protocol { + return []p2p.Protocol{ + { + Name: swapSpec.Name, + Version: swapSpec.Version, + Length: swapSpec.Length(), + Run: p.Run, + }, + } +} + +func (p *Swap) Run(peer *p2p.Peer, rw p2p.MsgReadWriter) error { + pp := protocols.NewPeer(peer, rw, swapSpec) + /* + p.fwdPoolMu.Lock() + p.fwdPool[peer.Info().ID] = pp + p.fwdPoolMu.Unlock() + */ + return pp.Run(p.handleSwapMsg) +} + +func (p *Swap) APIs() []rpc.API { + apis := []rpc.API{ + { + Namespace: "swap", + Version: "1.0", + Service: NewAPI(p), + Public: true, + }, + } + return apis +} + +// Filters incoming messages for processing or forwarding. +// Check if address partially matches +// If yes, it CAN be for us, and we process it +// Only passes error to pss protocol handler if payload is not valid pssmsg +func (p *Swap) handleSwapMsg(ctx context.Context, msg interface{}) error { + return nil +} + +// 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..289ba028be --- /dev/null +++ b/swarm/swap/swap_test.go @@ -0,0 +1,275 @@ +// 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 ( + "flag" + "fmt" + "os" + "path/filepath" + "sync" + "testing" + + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/node" + "github.com/ethereum/go-ethereum/rpc" + 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") +) + +func init() { + flag.Parse() + + log.PrintOrigins(true) + log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(colorable.NewColorableStderr(), log.TerminalFormat(true)))) +} + +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 +} + +func TestBasicSwap(t *testing.T) { + + // create the two nodes + stack_one, err := newServiceNode(p2pPort, 0, 0) + if err != nil { + log.Crit("Create servicenode #1 fail", "err", err) + } + stack_two, err := newServiceNode(p2pPort+1, 0, 0) + if err != nil { + log.Crit("Create servicenode #2 fail", "err", err) + } + + instance, err := NewSwap(NewDefaultSwapParams().Params) + if err != nil { + t.Fatal("Couldn't create Swap instance") + } + // wrapper function for servicenode to start the service + swapsvc := func(ctx *node.ServiceContext) (node.Service, error) { + return &API{ + Swap: instance, + }, nil + } + + // register adds the service to the services the servicenode starts when started + err = stack_one.Register(swapsvc) + if err != nil { + log.Crit("Register service in servicenode #1 fail", "err", err) + } + err = stack_two.Register(swapsvc) + if err != nil { + log.Crit("Register service in servicenode #2 fail", "err", err) + } + + // start the nodes + err = stack_one.Start() + if err != nil { + log.Crit("servicenode #1 start failed", "err", err) + } + err = stack_two.Start() + if err != nil { + log.Crit("servicenode #2 start failed", "err", err) + } + + // connect to the servicenode RPCs + rpcclient_one, err := rpc.Dial(filepath.Join(stack_one.DataDir(), ipcpath)) + if err != nil { + log.Crit("connect to servicenode #1 IPC fail", "err", err) + } + defer os.RemoveAll(stack_one.DataDir()) + + rpcclient_two, err := rpc.Dial(filepath.Join(stack_two.DataDir(), ipcpath)) + if err != nil { + log.Crit("connect to servicenode #2 IPC fail", "err", err) + } + defer os.RemoveAll(stack_two.DataDir()) + + // display that the initial pong counts are 0 + var balance int + err = rpcclient_one.Call(&balance, "swap_balance") + if err != nil { + log.Crit("servicenode #1 pongcount RPC failed", "err", err) + } + log.Info("servicenode #1 before ping", "balance-1", balance) + + err = rpcclient_two.Call(&balance, "swap_balance") + if err != nil { + log.Crit("servicenode #2 pongcount RPC failed", "err", err) + } + log.Info("servicenode #2 before ping", "balance-2", balance) + + /* + // get the server instances + srv_one := stack_one.Server() + srv_two := stack_two.Server() + + // subscribe to peerevents + eventOneC := make(chan *p2p.PeerEvent) + sub_one := srv_one.SubscribeEvents(eventOneC) + + eventTwoC := make(chan *p2p.PeerEvent) + sub_two := srv_two.SubscribeEvents(eventTwoC) + + // connect the nodes + p2pnode_two := srv_two.Self() + srv_one.AddPeer(p2pnode_two) + + // fork and do the pinging + stackW.Add(2) + pingmax_one := 4 + pingmax_two := 2 + + go func() { + + // when we get the add event, we know we are connected + ev := <-eventOneC + if ev.Type != "add" { + log.Error("server #1 expected peer add", "eventtype", ev.Type) + stackW.Done() + return + } + log.Debug("server #1 connected", "peer", ev.Peer) + + // send the pings + for i := 0; i < pingmax_one; i++ { + err := rpcclient_one.Call(nil, "foo_ping", ev.Peer) + if err != nil { + log.Error("server #1 RPC ping fail", "err", err) + stackW.Done() + break + } + } + + // wait for all msgrecv events + // pings we receive, and pongs we expect from pings we sent + for i := 0; i < pingmax_two+pingmax_one; { + ev := <-eventOneC + log.Warn("msg", "type", ev.Type, "i", i) + if ev.Type == "msgrecv" { + i++ + } + } + + stackW.Done() + }() + + // mirrors the previous go func + go func() { + ev := <-eventTwoC + if ev.Type != "add" { + log.Error("expected peer add", "eventtype", ev.Type) + stackW.Done() + return + } + log.Debug("server #2 connected", "peer", ev.Peer) + for i := 0; i < pingmax_two; i++ { + err := rpcclient_two.Call(nil, "foo_ping", ev.Peer) + if err != nil { + log.Error("server #2 RPC ping fail", "err", err) + stackW.Done() + break + } + } + + for i := 0; i < pingmax_one+pingmax_two; { + ev := <-eventTwoC + if ev.Type == "msgrecv" { + log.Warn("msg", "type", ev.Type, "i", i) + i++ + } + } + + stackW.Done() + }() + + // wait for the two ping pong exchanges to finish + stackW.Wait() + + // tell the API to shut down + // this will disconnect the peers and close the channels connecting API and protocol + err = rpcclient_one.Call(nil, "foo_quit", srv_two.Self().ID) + if err != nil { + log.Error("server #1 RPC quit fail", "err", err) + } + err = rpcclient_two.Call(nil, "foo_quit", srv_one.Self().ID) + if err != nil { + log.Error("server #2 RPC quit fail", "err", err) + } + + // disconnect will generate drop events + for { + ev := <-eventOneC + if ev.Type == "drop" { + break + } + } + for { + ev := <-eventTwoC + if ev.Type == "drop" { + break + } + } + + // proudly inspect the results + err = rpcclient_one.Call(&count, "foo_pongCount") + if err != nil { + log.Crit("servicenode #1 pongcount RPC failed", "err", err) + } + log.Info("servicenode #1 after ping", "pongcount", count) + + err = rpcclient_two.Call(&count, "foo_pongCount") + if err != nil { + log.Crit("servicenode #2 pongcount RPC failed", "err", err) + } + log.Info("servicenode #2 after ping", "pongcount", count) + + // bring down the servicenodes + sub_one.Unsubscribe() + sub_two.Unsubscribe() + stack_one.Stop() + stack_two.Stop() + */ +} diff --git a/swarm/swarm.go b/swarm/swarm.go index aea0989a1e..e2585c6de3 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -51,6 +51,8 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" "github.com/ethereum/go-ethereum/swarm/storage/feed" "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 +80,7 @@ type Swarm struct { netStore *storage.NetStore sfs *fuse.SwarmFS // need this to cleanup all the active mounts on node exit ps *pss.Pss + swap *swap.Swap tracerClose io.Closer } @@ -206,6 +209,10 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e self.bzz = network.NewBzz(bzzconfig, to, stateStore, stream.Spec, self.streamer.Run) + self.swap, err = swap.NewSwap(config.Swap.Params) + if err != nil { + return nil, err + } // Pss = postal service over swarm (devp2p over bzz) self.ps, err = pss.NewPss(to, config.Pss) if err != nil { @@ -361,6 +368,13 @@ func (self *Swarm) Start(srv *p2p.Server) error { } log.Info("Swarm network started", "bzzaddr", fmt.Sprintf("%x", self.bzz.Hive.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) }