forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-27 15:16:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

swarm, cmd/swarm: Removed websocket opinionation in client constructor

Browse source 
PR cleanup
This commit is contained in:
lash 2017-06-20 21:30:46 +02:00 committed by Lewis Marshall
parent 65023d7d4b
commit 10adb6e917
8 changed files with 152 additions and 681 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
cmd/swarm/main.go
View file

@ -118,18 +118,9 @@ var (
Usage: "force mime type",
}
PssEnabledFlag = cli.BoolFlag{
Name: "pss",
Name: "pss",
Usage: "Enable pss (message passing over swarm)",
}
PssHostFlag = cli.StringFlag{
Name: "psshost",
Usage: fmt.Sprintf("Websockets host or ip for pss (default '%v')", node.DefaultWSHost),
Value: node.DefaultWSHost,
}
PssPortFlag = cli.IntFlag{
Name: "pssport",
Usage: fmt.Sprintf("Websockets port for pss (default %d)", node.DefaultWSPort),
}
CorsStringFlag = cli.StringFlag{
Name: "corsdomain",
Usage: "Domain on which to send Access-Control-Allow-Origin header (multiple domains can be supplied separated by a ',')",
@ -312,14 +303,6 @@ func version(ctx *cli.Context) error {
func bzzd(ctx *cli.Context) error {
cfg := defaultNodeConfig
if ctx.GlobalIsSet(PssEnabledFlag.Name) {
cfg.WSHost = ctx.GlobalString(PssHostFlag.Name)
cfg.WSModules = []string{"eth","pss"}
cfg.WSOrigins = []string{"*"}
if ctx.GlobalIsSet(PssPortFlag.Name) {
cfg.WSPort = ctx.GlobalInt(PssPortFlag.Name)
}
}
utils.SetNodeConfig(ctx, &cfg)
stack, err := node.New(&cfg)
if err != nil {

19
swarm/network/protocol.go
View file

@ -292,25 +292,6 @@ func (self *bzzHandshake) Perform(p *p2p.Peer, rw p2p.MsgReadWriter) (err error)
return fmt.Errorf("version mismatch %d (!= %d)", rhs.Version, self.Version)
}
self.peerAddr = rhs.Addr
//log.Debug("L&R handshake", "local", fmt.Sprintf("%x", self.Addr.Over()), "remote", fmt.Sprintf("%x", rhs.Addr.Over()))
// log.Debug("RHS handshake", "addr", rhs.Addr, "peeraddr", rhs.peerAddr)
// log.Debug("LHS handshake", "addr", self.Addr, "peeraddr", self.peerAddr)
//
// log.Warn(
// strings.Join(
// []string{
// "handshake ok:",
// fmt.Sprintf("p2p.Peer.ID(): ...... %v", p.ID()),
// fmt.Sprintf("peerAddr.UAddr: ...... %x", self.peerAddr.UAddr[:32]),
// fmt.Sprintf("peerAddr.OAddr: ...... %x", self.peerAddr.OAddr[:32]),
// fmt.Sprintf("selfAddr.UAddr: ...... %x", self.Addr.UAddr[:32]),
// fmt.Sprintf("selfAddr.OAddr: ...... %x", self.Addr.OAddr[:32]),
// fmt.Sprintf("ToOverlayAddr(p2p.Peer.ID): .... %x", ToOverlayAddr(peerid[:])),
// fmt.Sprintf("ToOverlayAddr(peerAddr.UAddr): .. %x", ToOverlayAddr(self.peerAddr.UAddr)),
// }, "\n",
// ),
// )
return nil
}

8
swarm/network/simulations/discovery/discovery_test.go
View file

@ -3,8 +3,8 @@ package discovery_test
import (
"context"
"encoding/json"
"fmt"
"flag"
"fmt"
"io/ioutil"
"os"
"testing"
@ -30,8 +30,8 @@ var services = adapters.Services{
var (
snapshotFile = flag.String("snapshot", "", "create snapshot")
nodeCount = flag.Int("nodes", 10, "number of nodes to create (default 10)")
verbose = flag.Bool("verbose", false, "output extra logs")
nodeCount = flag.Int("nodes", 10, "number of nodes to create (default 10)")
verbose = flag.Bool("verbose", false, "output extra logs")
)
func init() {
@ -156,7 +156,7 @@ func testDiscoverySimulation(t *testing.T, adapter adapters.NodeAdapter) {
t.Fatalf("corrupt json snapshot: %v", err)
}
log.Info("writing snapshot", "file", *snapshotFile)
err = ioutil.WriteFile(*snapshotFile, jsonsnapshot, os.ModePerm)
err = ioutil.WriteFile(*snapshotFile, jsonsnapshot, 0755)
if err != nil {
t.Fatal(err)
}

226
swarm/pss/client/client.go
View file

@ -1,82 +1,3 @@
// simple abstraction for implementing pss functionality
//
// the pss client library aims to simplify usage of the p2p.protocols package over pss
//
// IO is performed using the ordinary p2p.MsgReadWriter interface, which transparently communicates with a pss node via RPC using websockets as transport layer, using methods in the PssAPI class in the swarm/pss package
//
//
// Minimal-ish usage example (requires a running pss node with websocket RPC):
//
//
// import (
// "context"
// "fmt"
// "os"
// pss "github.com/ethereum/go-ethereum/swarm/pss/client"
// "github.com/ethereum/go-ethereum/p2p/protocols"
// "github.com/ethereum/go-ethereum/p2p"
// "github.com/ethereum/go-ethereum/pot"
// "github.com/ethereum/go-ethereum/log"
// )
//
// type FooMsg struct {
// Bar int
// }
//
//
// func fooHandler (msg interface{}) error {
// foomsg, ok := msg.(*FooMsg)
// if ok {
// log.Debug("Yay, just got a message", "msg", foomsg)
// }
// return fmt.Errorf("Unknown message")
// }
//
// spec := &protocols.Spec{
// Name: "foo",
// Version: 1,
// MaxMsgSize: 1024,
// Messages: []interface{}{
// FooMsg{},
// },
// }
//
// proto := &p2p.Protocol{
// Name: spec.Name,
// Version: spec.Version,
// Length: uint64(len(spec.Messages)),
// Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
// pp := protocols.NewPeer(p, rw, spec)
// return pp.Run(fooHandler)
// },
// }
//
// func implementation() {
// cfg := pss.NewClientConfig()
// psc := pss.NewClient(context.Background(), nil, cfg)
// err := psc.Start()
// if err != nil {
// log.Crit("can't start pss client")
// os.Exit(1)
// }
//
// log.Debug("connected to pss node", "bzz addr", psc.BaseAddr)
//
// err = psc.RunProtocol(proto)
// if err != nil {
// log.Crit("can't start protocol on pss websocket")
// os.Exit(1)
// }
//
// addr := pot.RandomAddress() // should be a real address, of course
// psc.AddPssPeer(addr, spec)
//
// // use the protocol for something
//
// psc.Stop()
// }
//
// BUG(test): TestIncoming test times out due to deadlock issues in the swarm hive
package client
import (
@ -86,7 +7,6 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/protocols"
@ -99,45 +19,29 @@ import (
const (
inboxCapacity = 3000
outboxCapacity = 100
defaultWSHost = 8546
addrLen = common.HashLength
)
// RemoteHost: hostname of node running websockets proxy to pss (default localhost)
// RemotePort: port of node running websockets proxy to pss (0 = go-ethereum node default)
// SelfHost: local if host to connect from
// Secure: whether or not to use secure connection (not currently in use)
type ClientConfig struct {
RemoteHost string
RemotePort int
SelfHost string
Secure bool
}
// Generates a pss client configuration with default values
func NewClientConfig() *ClientConfig {
return &ClientConfig{
SelfHost: node.DefaultWSHost,
RemoteHost: node.DefaultWSHost,
RemotePort: node.DefaultWSPort,
}
}
// After a successful connection with Client.Start, BaseAddr contains the swarm overlay address of the pss node
type Client struct {
BaseAddr []byte
localuri string
remoteuri string
ctx context.Context
cancel func()
subscription *rpc.ClientSubscription
topicsC chan []byte
msgC chan pss.APIMsg
quitC chan struct{}
ws *rpc.Client
lock sync.Mutex
peerPool map[pss.Topic]map[pot.Address]*pssRPCRW
protos map[pss.Topic]*p2p.Protocol
BaseAddr []byte
// peers
peerPool map[pss.Topic]map[pot.Address]*pssRPCRW
protos map[pss.Topic]*p2p.Protocol
// rpc connections
rpc *rpc.Client
sub *rpc.ClientSubscription
// channels
topicsC chan []byte
msgC chan pss.APIMsg
quitC chan struct{}
ctx context.Context
cancel func()
lock sync.Mutex
}
// implements p2p.MsgReadWriter
@ -181,26 +85,46 @@ func (rw *pssRPCRW) WriteMsg(msg p2p.Msg) error {
return err
}
return rw.Client.ws.CallContext(rw.Client.ctx, nil, "pss_send", rw.topic, pss.APIMsg{
return rw.Client.rpc.CallContext(rw.Client.ctx, nil, "pss_send", rw.topic, pss.APIMsg{
Addr: rw.addr.Bytes(),
Msg: pmsg,
})
}
// Constructor for production-environment clients
// Performs sanity checks on configuration paramters and gets everything ready to connect to pss node
func NewClient(ctx context.Context, cancel func(), config *ClientConfig) (*Client, error) {
prefix := "ws"
func NewClient(ctx context.Context, cancel func(), rpcurl string) (*Client, error) {
rpcclient, err := rpc.Dial(rpcurl)
if err != nil {
return nil, err
}
client, err := NewClientWithRPC(ctx, cancel, rpcclient)
if err != nil {
return nil, err
}
return client, nil
}
// Constructor for test implementations
// The 'rpcclient' parameter allows passing a in-memory rpc client to act as the remote websocket RPC.
func NewClientWithRPC(ctx context.Context, cancel func(), rpcclient *rpc.Client) (*Client, error) {
client := newClient(ctx, cancel)
client.rpc = rpcclient
err := client.rpc.CallContext(client.ctx, &client.BaseAddr, "pss_baseAddr")
if err != nil {
return nil, fmt.Errorf("cannot get pss node baseaddress: %v", err)
}
return client, nil
}
func newClient(ctx context.Context, cancel func()) (client *Client) {
if ctx == nil {
ctx = context.Background()
}
if cancel == nil {
cancel = func() { return }
cancel = func() {}
}
pssc := &Client{
client = &Client{
msgC: make(chan pss.APIMsg),
quitC: make(chan struct{}),
peerPool: make(map[pss.Topic]map[pot.Address]*pssRPCRW),
@ -208,64 +132,13 @@ func NewClient(ctx context.Context, cancel func(), config *ClientConfig) (*Clien
ctx: ctx,
cancel: cancel,
}
if config.Secure {
prefix = "wss"
}
pssc.remoteuri = fmt.Sprintf("%s://%s:%d", prefix, config.RemoteHost, config.RemotePort)
pssc.localuri = fmt.Sprintf("%s://%s", prefix, config.SelfHost)
return pssc, nil
}
// Constructor for test implementations
// The 'rpcclient' parameter allows passing a in-memory rpc client to act as the remote websocket RPC.
func NewClientWithRPC(ctx context.Context, rpcclient *rpc.Client) (*Client, error) {
var oaddr []byte
err := rpcclient.CallContext(ctx, &oaddr, "pss_baseAddr")
if err != nil {
return nil, fmt.Errorf("cannot get pss node baseaddress: %v", err)
}
return &Client{
msgC: make(chan pss.APIMsg),
quitC: make(chan struct{}),
peerPool: make(map[pss.Topic]map[pot.Address]*pssRPCRW),
protos: make(map[pss.Topic]*p2p.Protocol),
ws: rpcclient,
ctx: ctx,
BaseAddr: oaddr,
}, nil
return
}
func (self *Client) shutdown() {
self.cancel()
}
// Connects to the websockets RPC
// Retrieves the swarm overlay address from the pss node
func (self *Client) Start() error {
if self.ws != nil {
return nil
}
log.Debug("Dialing ws", "src", self.localuri, "dst", self.remoteuri)
ws, err := rpc.DialWebsocket(self.ctx, self.remoteuri, self.localuri)
if err != nil {
return fmt.Errorf("Couldnt dial pss websocket: %v", err)
}
var oaddr []byte
err = ws.CallContext(self.ctx, &oaddr, "pss_baseAddr")
if err != nil {
return err
}
self.ws = ws
self.BaseAddr = oaddr
return nil
}
// Mounts a new devp2p protcool on the pss connection
// the protocol is aliased as a "pss topic"
// uses normal devp2p Send and incoming message handler routines from the p2p/protocols package
@ -275,12 +148,11 @@ func (self *Client) RunProtocol(proto *p2p.Protocol) error {
topic := pss.NewTopic(proto.Name, int(proto.Version))
msgC := make(chan pss.APIMsg)
self.peerPool[topic] = make(map[pot.Address]*pssRPCRW)
sub, err := self.ws.Subscribe(self.ctx, "pss", msgC, "receive", topic)
sub, err := self.rpc.Subscribe(self.ctx, "pss", msgC, "receive", topic)
if err != nil {
return fmt.Errorf("pss event subscription failed: %v", err)
}
self.subscription = sub
self.sub = sub
// dispatch incoming messages
go func() {

26
swarm/pss/client/client_test.go
View file

@ -51,7 +51,7 @@ func TestIncoming(t *testing.T) {
t.Fatalf(err.Error())
}
client.ws.Call(&addr, "psstest_baseAddr")
client.rpc.Call(&addr, "psstest_baseAddr")
code, _ := pss.PingProtocol.GetCode(&pss.PingMsg{})
rlpbundle, err := pss.NewProtocolMsg(code, &pss.PingMsg{
@ -94,7 +94,7 @@ func TestOutgoing(t *testing.T) {
t.Fatalf(err.Error())
}
client.ws.Call(&addr, "psstest_baseAddr")
client.rpc.Call(&addr, "psstest_baseAddr")
copy(potaddr[:], addr)
msg := &pss.PingMsg{
@ -118,12 +118,7 @@ func TestOutgoing(t *testing.T) {
func baseTester(t *testing.T, proto *p2p.Protocol, ps *pss.Pss, ctx context.Context, cancel func(), quitC chan struct{}) (*Client, error) {
var err error
client := newClient(t, ctx, cancel, quitC)
err = client.Start()
if err != nil {
return nil, err
}
client := newTestclient(t, ctx, cancel, quitC)
err = client.RunProtocol(proto)
@ -148,14 +143,7 @@ func newProtocol(ping *pss.Ping) *p2p.Protocol {
}
}
func newClient(t *testing.T, ctx context.Context, cancel func(), quitC chan struct{}) *Client {
conf := NewClientConfig()
pssclient, err := NewClient(ctx, cancel, conf)
if err != nil {
t.Fatalf(err.Error())
}
func newTestclient(t *testing.T, ctx context.Context, cancel func(), quitC chan struct{}) *Client {
ps := pss.NewTestPss(nil)
srv := rpc.NewServer()
@ -177,5 +165,11 @@ func newClient(t *testing.T, ctx context.Context, cancel func(), quitC chan stru
<-quitC
sock.Close()
}()
pssclient, err := NewClient(ctx, cancel, "ws://localhost:8546")
if err != nil {
t.Fatalf(err.Error())
}
return pssclient
}

80
swarm/pss/client/doc.go Normal file
View file

@ -0,0 +1,80 @@
// simple abstraction for implementing pss functionality
//
// the pss client library aims to simplify usage of the p2p.protocols package over pss
//
// IO is performed using the ordinary p2p.MsgReadWriter interface, which transparently communicates with a pss node via RPC using websockets as transport layer, using methods in the PssAPI class in the swarm/pss package
//
//
// Minimal-ish usage example (requires a running pss node with websocket RPC):
//
//
// import (
// "context"
// "fmt"
// "os"
// pss "github.com/ethereum/go-ethereum/swarm/pss/client"
// "github.com/ethereum/go-ethereum/p2p/protocols"
// "github.com/ethereum/go-ethereum/p2p"
// "github.com/ethereum/go-ethereum/pot"
// "github.com/ethereum/go-ethereum/log"
// )
//
// type FooMsg struct {
// Bar int
// }
//
//
// func fooHandler (msg interface{}) error {
// foomsg, ok := msg.(*FooMsg)
// if ok {
// log.Debug("Yay, just got a message", "msg", foomsg)
// }
// return fmt.Errorf("Unknown message")
// }
//
// spec := &protocols.Spec{
// Name: "foo",
// Version: 1,
// MaxMsgSize: 1024,
// Messages: []interface{}{
// FooMsg{},
// },
// }
//
// proto := &p2p.Protocol{
// Name: spec.Name,
// Version: spec.Version,
// Length: uint64(len(spec.Messages)),
// Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
// pp := protocols.NewPeer(p, rw, spec)
// return pp.Run(fooHandler)
// },
// }
//
// func implementation() {
// cfg := pss.NewClientConfig()
// psc := pss.NewClient(context.Background(), nil, cfg)
// err := psc.Start()
// if err != nil {
// log.Crit("can't start pss client")
// os.Exit(1)
// }
//
// log.Debug("connected to pss node", "bzz addr", psc.BaseAddr)
//
// err = psc.RunProtocol(proto)
// if err != nil {
// log.Crit("can't start protocol on pss websocket")
// os.Exit(1)
// }
//
// addr := pot.RandomAddress() // should be a real address, of course
// psc.AddPssPeer(addr, spec)
//
// // use the protocol for something
//
// psc.Stop()
// }
//
// BUG(test): TestIncoming test times out due to deadlock issues in the swarm hive
package client

40
swarm/swarm.go
View file

@ -1,5 +1,3 @@
// +build !psschat
// Copyright 2016 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
@ -35,6 +33,7 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/api"
httpapi "github.com/ethereum/go-ethereum/swarm/api/http"
@ -47,13 +46,12 @@ import (
// the swarm stack
type Swarm struct {
config *api.Config // swarm configuration
api *api.Api // high level api layer (fs/manifest)
dns api.Resolver // DNS registrar
storage storage.ChunkStore // internal access to storage, common interface to cloud storage backends
dpa *storage.DPA // distributed preimage archive, the local API to the storage with document level storage/retrieval support
cloud storage.CloudStore // procurement, cloud storage backend (can multi-cloud)
//hive *network.Hive // the logistic manager
config *api.Config // swarm configuration
api *api.Api // high level api layer (fs/manifest)
dns api.Resolver // DNS registrar
storage storage.ChunkStore // internal access to storage, common interface to cloud storage backends
dpa *storage.DPA // distributed preimage archive, the local API to the storage with document level storage/retrieval support
cloud storage.CloudStore // procurement, cloud storage backend (can multi-cloud)
bzz *network.Bzz // hive and bzz protocol
backend chequebook.Backend // simple blockchain Backend
privateKey *ecdsa.PrivateKey
@ -122,14 +120,6 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.
}
self.bzz = network.NewBzz(bzzconfig, to, nil)
// set up the kademlia hive
// self.hive = network.NewHive(
// config.HiveParams, // configuration parameters
// self.Kademlia,
// stateStore,
// )
// log.Debug(fmt.Sprintf("Set up swarm network with Kademlia hive"))
//
// setup cloud storage internal access layer
self.storage = storage.NewNetStore(hash, self.lstore, nil, config.StoreParams)
log.Debug("-> swarm net store shared access layer to Swarm Chunk Store")
@ -213,7 +203,6 @@ func (self *Swarm) Start(net *p2p.Server) error {
log.Warn(fmt.Sprintf("Starting Swarm service"))
// self.hive.Start(net)
err := self.bzz.Start(net)
if err != nil {
log.Error("bzz failed", "err", err)
@ -251,7 +240,6 @@ func (self *Swarm) Start(net *p2p.Server) error {
// stops all component services.
func (self *Swarm) Stop() error {
self.dpa.Stop()
//self.hive.Stop()
self.bzz.Stop()
//if self.pss != nil {
// self.pss.Stop()
@ -281,18 +269,6 @@ func (self *Swarm) Protocols() (protos []p2p.Protocol) {
protos = append(protos, p)
}
}
// ct := network.BzzCodeMap()
// ct.Register(2, network.DiscoveryMsgs...)
// proto := network.Bzz(
// self.hive.Overlay.GetAddr().Over(),
// self.hive.Overlay.GetAddr().Under(),
// ct,
// nil,
// nil,
// nil,
// )
//
return
}
@ -380,7 +356,7 @@ func NewLocalSwarm(datadir, port string) (self *Swarm, err error) {
return
}
config, err := api.NewConfig(datadir, common.Address{}, prvKey, network.NetworkID)
config, err := api.NewConfig(datadir, common.Address{}, prvKey, network.NetworkId)
if err != nil {
return
}

415
swarm/swarm_psschat.go
View file

@ -1,415 +0,0 @@
// +build psschat
// Copyright 2016 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 swarm
import (
"bytes"
"context"
"crypto/ecdsa"
"fmt"
"io/ioutil"
"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/ens"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/ethclient"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/api"
httpapi "github.com/ethereum/go-ethereum/swarm/api/http"
"github.com/ethereum/go-ethereum/swarm/fuse"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/pss"
"github.com/ethereum/go-ethereum/swarm/storage"
//psschat "github.com/ethereum/go-ethereum/swarm/pss/protocols/chat"
)
// the swarm stack
type Swarm struct {
config *api.Config // swarm configuration
api *api.Api // high level api layer (fs/manifest)
dns api.Resolver // DNS registrar
storage storage.ChunkStore // internal access to storage, common interface to cloud storage backends
dpa *storage.DPA // distributed preimage archive, the local API to the storage with document level storage/retrieval support
cloud storage.CloudStore // procurement, cloud storage backend (can multi-cloud)
//hive *network.Hive // the logistic manager
bzz *network.Bzz // hive and bzz protocol
backend chequebook.Backend // simple blockchain Backend
privateKey *ecdsa.PrivateKey
corsString string
swapEnabled bool
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
pss *pss.Pss
}
type SwarmAPI struct {
Api *api.Api
Backend chequebook.Backend
PrvKey *ecdsa.PrivateKey
}
func (self *Swarm) API() *SwarmAPI {
return &SwarmAPI{
Api: self.api,
Backend: self.backend,
PrvKey: self.privateKey,
}
}
// creates a new swarm service instance
// implements node.Service
func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.Config, swapEnabled, syncEnabled bool, cors string, pssEnabled bool) (self *Swarm, err error) {
if bytes.Equal(common.FromHex(config.PublicKey), storage.ZeroKey) {
return nil, fmt.Errorf("empty public key")
}
if bytes.Equal(common.FromHex(config.BzzKey), storage.ZeroKey) {
return nil, fmt.Errorf("empty bzz key")
}
self = &Swarm{
config: config,
swapEnabled: swapEnabled,
backend: backend,
privateKey: config.Swap.PrivateKey(),
corsString: cors,
}
log.Debug(fmt.Sprintf("Setting up Swarm service components"))
hash := storage.MakeHashFunc(config.ChunkerParams.Hash)
self.lstore, err = storage.NewLocalStore(hash, config.StoreParams)
if err != nil {
return
}
log.Debug("Set up local db access (iterator/counter)")
kp := network.NewKadParams()
to := network.NewKademlia(
common.FromHex(config.BzzKey),
kp,
)
config.HiveParams.Discovery = true
nodeid := discover.PubkeyID(crypto.ToECDSAPub(common.FromHex(config.PublicKey)))
addr := network.NewAddrFromNodeID(nodeid)
bzzconfig := &network.BzzConfig{
OverlayAddr: common.FromHex(config.BzzKey),
UnderlayAddr: addr.UAddr,
HiveParams: config.HiveParams,
}
self.bzz = network.NewBzz(bzzconfig, to, nil)
// set up the kademlia hive
// self.hive = network.NewHive(
// config.HiveParams, // configuration parameters
// self.Kademlia,
// stateStore,
// )
// log.Debug(fmt.Sprintf("Set up swarm network with Kademlia hive"))
//
// setup cloud storage internal access layer
self.storage = storage.NewNetStore(hash, self.lstore, nil, config.StoreParams)
log.Debug("-> swarm net store shared access layer to Swarm Chunk Store")
// set up DPA, the cloud storage local access layer
dpaChunkStore := storage.NewDpaChunkStore(self.lstore, self.storage)
log.Debug(fmt.Sprintf("-> Local Access to Swarm"))
// Swarm Hash Merklised Chunking for Arbitrary-length Document/File storage
self.dpa = storage.NewDPA(dpaChunkStore, self.config.ChunkerParams)
log.Debug(fmt.Sprintf("-> Content Store API"))
// netstore is broken, temp workaround to fiddle with pss
cachedir, err := ioutil.TempDir("", "bzz-tmp")
if err != nil {
return nil, err
}
self.dpa, err = storage.NewLocalDPA(cachedir)
if err != nil {
return nil, err
}
// Pss = postal service over swarm (devp2p over bzz)
if pssEnabled {
pssparams := pss.NewPssParams(false)
self.pss = pss.NewPss(to, self.dpa, pssparams)
}
// set up high level api
transactOpts := bind.NewKeyedTransactor(self.privateKey)
if backend == (*ethclient.Client)(nil) {
log.Warn("No ENS, please specify non-empty --ethapi to use domain name resolution")
} else {
self.dns, err = ens.NewENS(transactOpts, config.EnsRoot, self.backend)
if err != nil {
return nil, err
}
}
log.Debug(fmt.Sprintf("-> Swarm Domain Name Registrar @ address %v", config.EnsRoot.Hex()))
self.api = api.NewApi(self.dpa, self.dns)
// Manifests for Smart Hosting
log.Debug(fmt.Sprintf("-> Web3 virtual server API"))
self.sfs = fuse.NewSwarmFS(self.api)
log.Debug("-> Initializing Fuse file system")
return self, nil
}
/*
Start is called when the stack is started
* starts the network kademlia hive peer management
* (starts netStore level 0 api)
* starts DPA level 1 api (chunking -> store/retrieve requests)
* (starts level 2 api)
* starts http proxy server
* registers url scheme handlers for bzz, etc
* TODO: start subservices like sword, swear, swarmdns
*/
// implements the node.Service interface
func (self *Swarm) Start(net *p2p.Server) error {
// update uaddr to correct enode
newaddr := self.bzz.UpdateLocalAddr([]byte(net.Self().String()))
log.Warn("Updated bzz local addr", "oaddr", fmt.Sprintf("%x", newaddr.OAddr), "uaddr", fmt.Sprintf("%x", newaddr.UAddr))
// set chequebook
if self.swapEnabled {
ctx := context.Background() // The initial setup has no deadline.
err := self.SetChequebook(ctx)
if err != nil {
return fmt.Errorf("Unable to set chequebook for SWAP: %v", err)
}
log.Debug(fmt.Sprintf("-> cheque book for SWAP: %v", self.config.Swap.Chequebook()))
} else {
log.Debug(fmt.Sprintf("SWAP disabled: no cheque book set"))
}
log.Warn(fmt.Sprintf("Starting Swarm service"))
// self.hive.Start(net)
err := self.bzz.Start(net)
if err != nil {
log.Error("bzz failed", "err", err)
return err
}
log.Info(fmt.Sprintf("Swarm network started on bzz address: %x", self.bzz.Hive.Overlay.BaseAddr()))
if self.pss != nil {
self.pss.Start(net)
topic := pss.NewTopic("pssChat", 1)
self.pss.Register(&topic, func(msg []byte, p *p2p.Peer, from []byte) error {
log.Trace("placeholder handler for node got psschat proto incoming", "msg", msg, "from", from)
return nil
})
log.Info("Pss started ... with 'pssChat' :)")
}
self.dpa.Start()
log.Debug(fmt.Sprintf("Swarm DPA started"))
// start swarm http proxy server
if self.config.Port != "" {
addr := ":" + self.config.Port
go httpapi.StartHttpServer(self.api, &httpapi.ServerConfig{
Addr: addr,
CorsString: self.corsString,
})
}
log.Debug(fmt.Sprintf("Swarm http proxy started on port: %v", self.config.Port))
if self.corsString != "" {
log.Debug(fmt.Sprintf("Swarm http proxy started with corsdomain: %v", self.corsString))
}
return nil
}
// implements the node.Service interface
// stops all component services.
func (self *Swarm) Stop() error {
self.dpa.Stop()
//self.hive.Stop()
self.bzz.Stop()
//if self.pss != nil {
// self.pss.Stop()
//}
if ch := self.config.Swap.Chequebook(); ch != nil {
ch.Stop()
ch.Save()
}
if self.lstore != nil {
self.lstore.DbStore.Close()
}
self.sfs.Stop()
return self.config.Save()
}
// implements the node.Service interface
func (self *Swarm) Protocols() (protos []p2p.Protocol) {
for _, p := range self.bzz.Protocols() {
protos = append(protos, p)
}
if self.pss != nil {
log.Warn("adding pss protos")
for _, p := range self.pss.Protocols() {
protos = append(protos, p)
}
}
// ct := network.BzzCodeMap()
// ct.Register(2, network.DiscoveryMsgs...)
// proto := network.Bzz(
// self.hive.Overlay.GetAddr().Over(),
// self.hive.Overlay.GetAddr().Under(),
// ct,
// nil,
// nil,
// nil,
// )
//
return
}
// implements node.Service
// Apis returns the RPC Api descriptors the Swarm implementation offers
func (self *Swarm) APIs() []rpc.API {
apis := []rpc.API{
// public APIs
{
Namespace: "bzz",
Version: "0.1",
Service: &Info{self.config, chequebook.ContractParams},
Public: true,
},
// admin APIs
{
Namespace: "bzz",
Version: "0.1",
Service: api.NewControl(self.api, self.bzz.Hive),
Public: false,
},
{
Namespace: "chequebook",
Version: chequebook.Version,
Service: chequebook.NewApi(self.config.Swap.Chequebook),
Public: false,
},
{
Namespace: "swarmfs",
Version: fuse.Swarmfs_Version,
Service: self.sfs,
Public: false,
},
// storage APIs
// DEPRECATED: Use the HTTP API instead
{
Namespace: "bzz",
Version: "0.1",
Service: api.NewStorage(self.api),
Public: true,
},
{
Namespace: "bzz",
Version: "0.1",
Service: api.NewFileSystem(self.api),
Public: false,
},
// {Namespace, Version, api.NewAdmin(self), false},
}
for _, api := range self.bzz.APIs() {
apis = append(apis, api)
}
if self.pss != nil {
for _, api := range self.pss.APIs() {
apis = append(apis, api)
}
}
return apis
}
func (self *Swarm) Api() *api.Api {
return self.api
}
// SetChequebook ensures that the local checquebook is set up on chain.
func (self *Swarm) SetChequebook(ctx context.Context) error {
err := self.config.Swap.SetChequebook(ctx, self.backend, self.config.Path)
if err != nil {
return err
}
log.Info(fmt.Sprintf("new chequebook set (%v): saving config file, resetting all connections in the hive", self.config.Swap.Contract.Hex()))
self.config.Save()
return nil
}
// Local swarm without netStore
func NewLocalSwarm(datadir, port string) (self *Swarm, err error) {
prvKey, err := crypto.GenerateKey()
if err != nil {
return
}
config, err := api.NewConfig(datadir, common.Address{}, prvKey, network.NetworkId)
if err != nil {
return
}
config.Port = port
dpa, err := storage.NewLocalDPA(datadir)
if err != nil {
return
}
self = &Swarm{
api: api.NewApi(dpa, nil),
config: config,
}
return
}
// serialisable info about swarm
type Info struct {
*api.Config
*chequebook.Params
}
func (self *Info) Info() *Info {
return self
}