From 339e3c89a762e0dfec2653dca6cdbbe5f96b4a26 Mon Sep 17 00:00:00 2001 From: nolash Date: Tue, 6 Jun 2017 22:20:21 +0200 Subject: [PATCH] swarm, swarm/network, swarm/pss, les: fixed swarm bin + minor adj removed all Pss in symbol names whereever possible (and sensible) discovery switch in hive now prevents add of suggested peers this is probably not a permanent solution bzz and pss now started from swarm.go, Outputs from bzz/pss Protocols() and APIs() are appended to the Swarm methods --- les/handler.go | 2 +- les/serverpool.go | 2 +- swarm/network/hive.go | 28 ++--- swarm/pss/client/client.go | 95 +++++++-------- swarm/pss/client/client_test.go | 67 +++++------ swarm/pss/common.go | 32 ++--- swarm/pss/pss.go | 73 ++++++------ swarm/pss/pss_test.go | 81 +++++++------ swarm/pss/pssapi.go | 46 ++++---- swarm/pss/simulations/service_test.go | 8 +- swarm/pss/types.go | 30 ++--- swarm/swarm.go | 164 ++++++++++++-------------- 12 files changed, 292 insertions(+), 336 deletions(-) diff --git a/les/handler.go b/les/handler.go index 64023af0f5..aaf4e1d7a3 100644 --- a/les/handler.go +++ b/les/handler.go @@ -259,7 +259,7 @@ func (pm *ProtocolManager) removePeer(id string) { func (pm *ProtocolManager) Start(srvr *p2p.Server) { var topicDisc *discv5.Network if srvr != nil { - topicDisc = srvr.DiscV5 + topicDisc = srvr.DiscV5() } lesTopic := discv5.Topic("LES@" + common.Bytes2Hex(pm.blockchain.Genesis().Hash().Bytes()[0:8])) if pm.lightSync { diff --git a/les/serverpool.go b/les/serverpool.go index 64fe991c63..f4c3101cdf 100644 --- a/les/serverpool.go +++ b/les/serverpool.go @@ -143,7 +143,7 @@ func newServerPool(db ethdb.Database, dbPrefix []byte, server *p2p.Server, topic pool.discSetPeriod = make(chan time.Duration, 1) pool.discNodes = make(chan *discv5.Node, 100) pool.discLookups = make(chan bool, 100) - go pool.server.DiscV5.SearchTopic(topic, pool.discSetPeriod, pool.discNodes, pool.discLookups) + go pool.server.DiscV5().SearchTopic(topic, pool.discSetPeriod, pool.discNodes, pool.discLookups) } go pool.eventLoop() diff --git a/swarm/network/hive.go b/swarm/network/hive.go index da33020523..e402930569 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -127,23 +127,23 @@ func (self *Hive) Start(server *p2p.Server) error { log.Debug("hive delegate to overlay driver: suggest addr to connect to") // log.Trace("hive delegate to overlay driver: suggest addr to connect to") addr, order, want := self.SuggestPeer() - - if addr != nil { - log.Info(fmt.Sprintf("========> connect to bee %v", addr)) - under, err := discover.ParseNode(string(addr.(Addr).Under())) - if err == nil { - server.AddPeer(under) + if self.Discovery { + if addr != nil { + log.Info(fmt.Sprintf("========> connect to bee %v", addr)) + under, err := discover.ParseNode(string(addr.(Addr).Under())) + if err == nil { + server.AddPeer(under) + } else { + log.Error(fmt.Sprintf("===X====> connect to bee %v failed: invalid node URL: %v", addr, err)) + } } else { - log.Error(fmt.Sprintf("===X====> connect to bee %v failed: invalid node URL: %v", addr, err)) + log.Trace("cannot suggest peers") } - } else { - log.Trace("cannot suggest peers") - } - want = want && self.Discovery - if want { - log.Debug(fmt.Sprintf("========> request peers nearest %v", addr)) - RequestOrder(self.Overlay, uint8(order), self.PeersBroadcastSetSize, self.MaxPeersPerRequest) + if want { + log.Debug(fmt.Sprintf("========> request peers nearest %v", addr)) + RequestOrder(self.Overlay, uint8(order), self.PeersBroadcastSetSize, self.MaxPeersPerRequest) + } } log.Info(fmt.Sprintf("%v", self)) diff --git a/swarm/pss/client/client.go b/swarm/pss/client/client.go index f97a8b211d..a9bef6d3db 100644 --- a/swarm/pss/client/client.go +++ b/swarm/pss/client/client.go @@ -4,7 +4,6 @@ import ( "context" "fmt" "sync" - "sync/atomic" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/event" @@ -29,50 +28,49 @@ const ( // RemotePort: port of node running websockets proxy to pss (0 = go-ethereum node default) // Secure: whether or not to use secure connection // SelfHost: local if host to connect from -type PssClientConfig struct { +type ClientConfig struct { SelfHost string RemoteHost string RemotePort int Secure bool } -func NewPssClientConfig() *PssClientConfig { - return &PssClientConfig{ +func NewClientConfig() *ClientConfig { + return &ClientConfig{ SelfHost: "localhost", RemoteHost: "localhost", RemotePort: 8546, } } -type PssClient struct { +type Client struct { localuri string remoteuri string ctx context.Context cancel func() subscription *rpc.ClientSubscription topicsC chan []byte - msgC chan pss.PssAPIMsg + msgC chan pss.APIMsg quitC chan struct{} - quitting uint32 ws *rpc.Client lock sync.Mutex - peerPool map[pss.PssTopic]map[pot.Address]*pssRPCRW - protos map[pss.PssTopic]*p2p.Protocol + peerPool map[pss.Topic]map[pot.Address]*pssRPCRW + protos map[pss.Topic]*p2p.Protocol } type pssRPCRW struct { - *PssClient - topic *pss.PssTopic + *Client + topic *pss.Topic msgC chan []byte addr pot.Address } -func (self *PssClient) newpssRPCRW(addr pot.Address, topic *pss.PssTopic) *pssRPCRW { +func (self *Client) newpssRPCRW(addr pot.Address, topic *pss.Topic) *pssRPCRW { return &pssRPCRW{ - PssClient: self, - topic: topic, - msgC: make(chan []byte), - addr: addr, + Client: self, + topic: topic, + msgC: make(chan []byte), + addr: addr, } } @@ -91,7 +89,7 @@ func (rw *pssRPCRW) WriteMsg(msg p2p.Msg) error { log.Trace("got writemsg pssclient", "msg", msg) rlpdata := make([]byte, msg.Size) msg.Payload.Read(rlpdata) - pmsg, err := rlp.EncodeToBytes(pss.PssProtocolMsg{ + pmsg, err := rlp.EncodeToBytes(pss.ProtocolMsg{ Code: msg.Code, Size: msg.Size, Payload: rlpdata, @@ -99,13 +97,13 @@ func (rw *pssRPCRW) WriteMsg(msg p2p.Msg) error { if err != nil { return err } - return rw.PssClient.ws.CallContext(rw.PssClient.ctx, nil, "pss_sendPss", rw.topic, pss.PssAPIMsg{ + return rw.Client.ws.CallContext(rw.Client.ctx, nil, "pss_send", rw.topic, pss.APIMsg{ Addr: rw.addr.Bytes(), Msg: pmsg, }) } -func NewPssClient(ctx context.Context, cancel func(), config *PssClientConfig) *PssClient { +func NewClient(ctx context.Context, cancel func(), config *ClientConfig) *Client { prefix := "ws" if ctx == nil { @@ -115,27 +113,15 @@ func NewPssClient(ctx context.Context, cancel func(), config *PssClientConfig) * cancel = func() { return } } - pssc := &PssClient{ - msgC: make(chan pss.PssAPIMsg), + pssc := &Client{ + msgC: make(chan pss.APIMsg), quitC: make(chan struct{}), - peerPool: make(map[pss.PssTopic]map[pot.Address]*pssRPCRW), - protos: make(map[pss.PssTopic]*p2p.Protocol), + peerPool: make(map[pss.Topic]map[pot.Address]*pssRPCRW), + protos: make(map[pss.Topic]*p2p.Protocol), ctx: ctx, cancel: cancel, } - if config.RemoteHost == "" { - config.RemoteHost = "localhost" - } - - if config.RemotePort == 0 { - config.RemotePort = defaultWSHost - } - - if config.SelfHost == "" { - config.SelfHost = "localhost" - } - if config.Secure { prefix = "wss" } @@ -146,23 +132,22 @@ func NewPssClient(ctx context.Context, cancel func(), config *PssClientConfig) * return pssc } -func NewPssClientWithRPC(ctx context.Context, client *rpc.Client) *PssClient { - return &PssClient{ - msgC: make(chan pss.PssAPIMsg), +func NewClientWithRPC(ctx context.Context, client *rpc.Client) *Client { + return &Client{ + msgC: make(chan pss.APIMsg), quitC: make(chan struct{}), - peerPool: make(map[pss.PssTopic]map[pot.Address]*pssRPCRW), - protos: make(map[pss.PssTopic]*p2p.Protocol), + peerPool: make(map[pss.Topic]map[pot.Address]*pssRPCRW), + protos: make(map[pss.Topic]*p2p.Protocol), ws: client, ctx: ctx, } } -func (self *PssClient) shutdown() { - atomic.StoreUint32(&self.quitting, 1) +func (self *Client) shutdown() { self.cancel() } -func (self *PssClient) Start() error { +func (self *Client) Start() error { if self.ws != nil { return nil } @@ -177,11 +162,11 @@ func (self *PssClient) Start() error { return nil } -func (self *PssClient) RunProtocol(proto *p2p.Protocol) error { +func (self *Client) RunProtocol(proto *p2p.Protocol) error { topic := pss.NewTopic(proto.Name, int(proto.Version)) - msgC := make(chan pss.PssAPIMsg) + msgC := make(chan pss.APIMsg) self.peerPool[topic] = make(map[pot.Address]*pssRPCRW) - sub, err := self.ws.Subscribe(self.ctx, "pss", msgC, "receivePss", topic) + sub, err := self.ws.Subscribe(self.ctx, "pss", msgC, "receive", topic) if err != nil { return fmt.Errorf("pss event subscription failed: %v", err) } @@ -215,12 +200,12 @@ func (self *PssClient) RunProtocol(proto *p2p.Protocol) error { return nil } -func (self *PssClient) Stop() error { +func (self *Client) Stop() error { self.cancel() return nil } -func (self *PssClient) AddPssPeer(addr pot.Address, spec *protocols.Spec) { +func (self *Client) AddPssPeer(addr pot.Address, spec *protocols.Spec) { topic := pss.NewTopic(spec.Name, int(spec.Version)) if self.peerPool[topic][addr] == nil { self.peerPool[topic][addr] = self.newpssRPCRW(addr, &topic) @@ -230,31 +215,31 @@ func (self *PssClient) AddPssPeer(addr pot.Address, spec *protocols.Spec) { } } -func (self *PssClient) RemovePssPeer(addr pot.Address, spec *protocols.Spec) { +func (self *Client) RemovePssPeer(addr pot.Address, spec *protocols.Spec) { topic := pss.NewTopic(spec.Name, int(spec.Version)) delete(self.peerPool[topic], addr) } -func (self *PssClient) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription { +func (self *Client) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription { log.Error("PSS client handles events internally, use the read functions instead") return nil } -func (self *PssClient) PeerCount() int { +func (self *Client) PeerCount() int { return len(self.peerPool) } -func (self *PssClient) NodeInfo() *p2p.NodeInfo { +func (self *Client) NodeInfo() *p2p.NodeInfo { return nil } -func (self *PssClient) PeersInfo() []*p2p.PeerInfo { +func (self *Client) PeersInfo() []*p2p.PeerInfo { return nil } -func (self *PssClient) AddPeer(node *discover.Node) { +func (self *Client) AddPeer(node *discover.Node) { log.Error("Cannot add peer in PSS with discover.Node, need swarm overlay address") } -func (self *PssClient) RemovePeer(node *discover.Node) { +func (self *Client) RemovePeer(node *discover.Node) { log.Error("Cannot remove peer in PSS with discover.Node, need swarm overlay address") } diff --git a/swarm/pss/client/client_test.go b/swarm/pss/client/client_test.go index 99341e7651..115316501a 100644 --- a/swarm/pss/client/client_test.go +++ b/swarm/pss/client/client_test.go @@ -26,8 +26,8 @@ func init() { func TestRunProtocol(t *testing.T) { quitC := make(chan struct{}) ps := pss.NewTestPss(nil) - ping := &pss.PssPing{ - QuitC: make(chan struct{}), + ping := &pss.Ping{ + C: make(chan struct{}), } proto := newProtocol(ping) _, err := baseTester(t, proto, ps, nil, nil, quitC) @@ -42,8 +42,8 @@ func TestIncoming(t *testing.T) { ps := pss.NewTestPss(nil) ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) var addr []byte - ping := &pss.PssPing{ - QuitC: make(chan struct{}), + ping := &pss.Ping{ + C: make(chan struct{}), } proto := newProtocol(ping) client, err := baseTester(t, proto, ps, ctx, cancel, quitC) @@ -53,37 +53,28 @@ func TestIncoming(t *testing.T) { client.ws.Call(&addr, "psstest_baseAddr") - code, _ := pss.PssPingProtocol.GetCode(&pss.PssPingMsg{}) - rlpbundle, err := pss.NewProtocolMsg(code, &pss.PssPingMsg{ + code, _ := pss.PingProtocol.GetCode(&pss.PingMsg{}) + rlpbundle, err := pss.NewProtocolMsg(code, &pss.PingMsg{ Created: time.Now(), }) if err != nil { - t.Fatalf("couldn't make pssmsg") + t.Fatalf("couldn't make pssmsg: %v", err) } - pssenv := pss.PssEnvelope{ - From: addr, - Topic: pss.NewTopic(proto.Name, int(proto.Version)), - TTL: pss.DefaultTTL, - Payload: rlpbundle, - } + pssenv := pss.NewEnvelope(addr, pss.NewTopic(proto.Name, int(proto.Version)), rlpbundle) pssmsg := pss.PssMsg{ To: addr, - Payload: &pssenv, + Payload: pssenv, } ps.Process(&pssmsg) - go func() { - <-ping.QuitC - client.cancel() - }() - select { case <-client.ctx.Done(): t.Fatalf("outgoing timed out or canceled") - default: + case <-ping.C: } + quitC <- struct{}{} } @@ -94,8 +85,8 @@ func TestOutgoing(t *testing.T) { var addr []byte var potaddr pot.Address - ping := &pss.PssPing{ - QuitC: make(chan struct{}), + ping := &pss.Ping{ + C: make(chan struct{}), } proto := newProtocol(ping) client, err := baseTester(t, proto, ps, ctx, cancel, quitC) @@ -106,25 +97,25 @@ func TestOutgoing(t *testing.T) { client.ws.Call(&addr, "psstest_baseAddr") copy(potaddr[:], addr) - msg := &pss.PssPingMsg{ + msg := &pss.PingMsg{ Created: time.Now(), } - topic := pss.NewTopic(pss.PssPingProtocol.Name, int(pss.PssPingProtocol.Version)) - client.AddPssPeer(potaddr, pss.PssPingProtocol) + topic := pss.NewTopic(pss.PingProtocol.Name, int(pss.PingProtocol.Version)) + client.AddPssPeer(potaddr, pss.PingProtocol) nid, _ := discover.HexID("0x00") p := p2p.NewPeer(nid, fmt.Sprintf("%v", potaddr), []p2p.Cap{}) - pp := protocols.NewPeer(p, client.peerPool[topic][potaddr], pss.PssPingProtocol) + pp := protocols.NewPeer(p, client.peerPool[topic][potaddr], pss.PingProtocol) pp.Send(msg) select { case <-client.ctx.Done(): t.Fatalf("outgoing timed out or canceled") - default: + case <-ping.C: } quitC <- struct{}{} } -func baseTester(t *testing.T, proto *p2p.Protocol, ps *pss.Pss, ctx context.Context, cancel func(), quitC chan struct{}) (*PssClient, error) { +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) @@ -143,30 +134,30 @@ func baseTester(t *testing.T, proto *p2p.Protocol, ps *pss.Pss, ctx context.Cont return client, nil } -func newProtocol(ping *pss.PssPing) *p2p.Protocol { +func newProtocol(ping *pss.Ping) *p2p.Protocol { return &p2p.Protocol{ - Name: pss.PssPingProtocol.Name, - Version: pss.PssPingProtocol.Version, + Name: pss.PingProtocol.Name, + Version: pss.PingProtocol.Version, Length: 1, Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error { - pp := protocols.NewPeer(p, rw, pss.PssPingProtocol) - pp.Run(ping.PssPingHandler) + pp := protocols.NewPeer(p, rw, pss.PingProtocol) + pp.Run(ping.PingHandler) return nil }, } } -func newClient(t *testing.T, ctx context.Context, cancel func(), quitC chan struct{}) *PssClient { +func newClient(t *testing.T, ctx context.Context, cancel func(), quitC chan struct{}) *Client { - conf := NewPssClientConfig() + conf := NewClientConfig() - pssclient := NewPssClient(ctx, cancel, conf) + pssclient := NewClient(ctx, cancel, conf) ps := pss.NewTestPss(nil) srv := rpc.NewServer() - srv.RegisterName("pss", pss.NewPssAPI(ps)) - srv.RegisterName("psstest", pss.NewPssAPITest(ps)) + srv.RegisterName("pss", pss.NewAPI(ps)) + srv.RegisterName("psstest", pss.NewAPITest(ps)) ws := srv.WebsocketHandler([]string{"*"}) uri := fmt.Sprintf("%s:%d", "localhost", 8546) diff --git a/swarm/pss/common.go b/swarm/pss/common.go index e8af1b944f..d6274bcd63 100644 --- a/swarm/pss/common.go +++ b/swarm/pss/common.go @@ -13,33 +13,33 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" ) -type PssPingMsg struct { +type PingMsg struct { Created time.Time } -type PssPing struct { - QuitC chan struct{} +type Ping struct { + C chan struct{} } -func (self *PssPing) PssPingHandler(msg interface{}) error { +func (self *Ping) PingHandler(msg interface{}) error { log.Warn("got ping", "msg", msg) - self.QuitC <- struct{}{} + self.C <- struct{}{} return nil } -var PssPingProtocol = &protocols.Spec{ +var PingProtocol = &protocols.Spec{ Name: "psstest", Version: 1, MaxMsgSize: 10 * 1024 * 1024, Messages: []interface{}{ - PssPingMsg{}, + PingMsg{}, }, } -var PssPingTopic = NewTopic(PssPingProtocol.Name, int(PssPingProtocol.Version)) +var PingTopic = NewTopic(PingProtocol.Name, int(PingProtocol.Version)) -func NewPssPingMsg(to []byte, spec *protocols.Spec, topic PssTopic, senderaddr []byte) PssMsg { - data := PssPingMsg{ +func NewPingMsg(to []byte, spec *protocols.Spec, topic Topic, senderaddr []byte) PssMsg { + data := PingMsg{ Created: time.Now(), } code, found := spec.GetCode(&data) @@ -54,19 +54,19 @@ func NewPssPingMsg(to []byte, spec *protocols.Spec, topic PssTopic, senderaddr [ pssmsg := PssMsg{ To: to, - Payload: NewPssEnvelope(senderaddr, topic, rlpbundle), + Payload: NewEnvelope(senderaddr, topic, rlpbundle), } return pssmsg } -func NewPssPingProtocol(handler func(interface{}) error) *p2p.Protocol { +func NewPingProtocol(handler func(interface{}) error) *p2p.Protocol { return &p2p.Protocol{ - Name: PssPingProtocol.Name, - Version: PssPingProtocol.Version, - Length: uint64(PssPingProtocol.MaxMsgSize), + Name: PingProtocol.Name, + Version: PingProtocol.Version, + Length: uint64(PingProtocol.MaxMsgSize), Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error { - pp := protocols.NewPeer(p, rw, PssPingProtocol) + pp := protocols.NewPeer(p, rw, PingProtocol) log.Trace(fmt.Sprintf("running pss vprotocol on peer %v", p)) err := pp.Run(handler) return err diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 0c1954dcd9..58e6df735c 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -2,14 +2,12 @@ package pss import ( "bytes" - //"encoding/binary" "errors" "fmt" "sync" "time" "github.com/ethereum/go-ethereum/common" - //"github.com/ethereum/go-ethereum/crypto/sha3" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/discover" @@ -54,6 +52,8 @@ type pssCacheEntry struct { type pssDigest [digestLength]byte +// implements node.Service +// // pss provides sending messages to nodes without having to be directly connected to them. // // The messages are wrapped in a PssMsg structure and routed using the swarm kademlia routing. @@ -67,15 +67,15 @@ type pssDigest [digestLength]byte // - a dispatcher lookup, mapping protocols to topics // - a message cache to spot messages that previously have been forwarded type Pss struct { - network.Overlay // we can get the overlayaddress from this - peerPool map[pot.Address]map[PssTopic]p2p.MsgReadWriter // keep track of all virtual p2p.Peers we are currently speaking to - fwdPool map[pot.Address]*protocols.Peer // keep track of all peers sitting on the pssmsg routing layer - handlers map[PssTopic]map[*PssHandler]bool // topic and version based pss payload handlers - fwdcache map[pssDigest]pssCacheEntry // checksum of unique fields from pssmsg mapped to expiry, cache to determine whether to drop msg - cachettl time.Duration // how long to keep messages in fwdcache + network.Overlay // we can get the overlayaddress from this + peerPool map[pot.Address]map[Topic]p2p.MsgReadWriter // keep track of all virtual p2p.Peers we are currently speaking to + fwdPool map[pot.Address]*protocols.Peer // keep track of all peers sitting on the pssmsg routing layer + handlers map[Topic]map[*Handler]bool // topic and version based pss payload handlers + fwdcache map[pssDigest]pssCacheEntry // checksum of unique fields from pssmsg mapped to expiry, cache to determine whether to drop msg + cachettl time.Duration // how long to keep messages in fwdcache lock sync.Mutex dpa *storage.DPA - debug bool + debug bool } func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) { @@ -94,18 +94,16 @@ func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) { } // Creates a new Pss instance. A node should only need one of these -// -// TODO: error check overlay integrity func NewPss(k network.Overlay, dpa *storage.DPA, params *PssParams) *Pss { return &Pss{ Overlay: k, - peerPool: make(map[pot.Address]map[PssTopic]p2p.MsgReadWriter, PssPeerCapacity), + peerPool: make(map[pot.Address]map[Topic]p2p.MsgReadWriter, PssPeerCapacity), fwdPool: make(map[pot.Address]*protocols.Peer), - handlers: make(map[PssTopic]map[*PssHandler]bool), + handlers: make(map[Topic]map[*Handler]bool), fwdcache: make(map[pssDigest]pssCacheEntry), cachettl: params.Cachettl, dpa: dpa, - debug: params.Debug, + debug: params.Debug, } } @@ -145,7 +143,7 @@ func (self *Pss) APIs() []rpc.API { rpc.API{ Namespace: "pss", Version: "0.1", - Service: NewPssAPI(self), + Service: NewAPI(self), Public: true, }, } @@ -153,32 +151,32 @@ func (self *Pss) APIs() []rpc.API { apis = append(apis, rpc.API{ Namespace: "pss", Version: "0.1", - Service: NewPssAPITest(self), + Service: NewAPITest(self), Public: true, }) } return apis } -// Takes the generated PssTopic of a protocol/chatroom etc, and links a handler function to it +// Takes the generated Topic of a protocol/chatroom etc, and links a handler function to it // This allows the implementer to retrieve the right handler functions (invoke the right protocol) // for an incoming message by inspecting the topic on it. // a topic allows for multiple handlers // returns a deregister function which needs to be called to deregister the handler // (similar to event.Subscription.Unsubscribe()) -func (self *Pss) Register(topic *PssTopic, handler PssHandler) func() { +func (self *Pss) Register(topic *Topic, handler Handler) func() { self.lock.Lock() defer self.lock.Unlock() handlers := self.handlers[*topic] if handlers == nil { - handlers = make(map[*PssHandler]bool) + handlers = make(map[*Handler]bool) self.handlers[*topic] = handlers } handlers[&handler] = true return func() { self.deregister(topic, &handler) } } -func (self *Pss) deregister(topic *PssTopic, h *PssHandler) { +func (self *Pss) deregister(topic *Topic, h *Handler) { self.lock.Lock() defer self.lock.Unlock() handlers := self.handlers[*topic] @@ -243,13 +241,12 @@ func (self *Pss) checkFwdCache(addr []byte, digest pssDigest) bool { return false } -func (self *Pss) getHandlers(topic PssTopic) map[*PssHandler]bool { +func (self *Pss) getHandlers(topic Topic) map[*Handler]bool { self.lock.Lock() defer self.lock.Unlock() return self.handlers[topic] } -// func (self *Pss) handlePssMsg(msg interface{}) error { pssmsg := msg.(*PssMsg) @@ -280,12 +277,12 @@ func (self *Pss) Process(pssmsg *PssMsg) error { return nil } -// Sends a message using The message could be anything at all, and will be handled by whichever handler function is mapped to PssTopic using *Pss.Register() +// Sends a message using The message could be anything at all, and will be handled by whichever handler function is mapped to Topic using *Pss.Register() // // The to address is a swarm overlay address -func (self *Pss) Send(to []byte, topic PssTopic, msg []byte) error { +func (self *Pss) SendRaw(to []byte, topic Topic, msg []byte) error { sender := self.Overlay.BaseAddr() - pssenv := NewPssEnvelope(sender, topic, msg) + pssenv := NewEnvelope(sender, topic, msg) pssmsg := &PssMsg{ To: to, Payload: pssenv, @@ -354,7 +351,7 @@ func (self *Pss) Forward(msg *PssMsg) error { // Links a pss peer address and topic to a dedicated p2p.MsgReadWriter in the pss peerpool, and runs the specificed protocol on this p2p.MsgReadWriter and the specified peer // // The effect is that now we have a "virtual" protocol running on an artificial p2p.Peer, which can be looked up and piped to through Pss using swarm overlay address and topic -func (self *Pss) AddPeer(p *p2p.Peer, addr pot.Address, run func(*p2p.Peer, p2p.MsgReadWriter) error, topic PssTopic, rw p2p.MsgReadWriter) error { +func (self *Pss) AddPeer(p *p2p.Peer, addr pot.Address, run func(*p2p.Peer, p2p.MsgReadWriter) error, topic Topic, rw p2p.MsgReadWriter) error { self.lock.Lock() defer self.lock.Unlock() self.addPeerTopic(addr, topic, rw) @@ -366,15 +363,15 @@ func (self *Pss) AddPeer(p *p2p.Peer, addr pot.Address, run func(*p2p.Peer, p2p. return nil } -func (self *Pss) addPeerTopic(id pot.Address, topic PssTopic, rw p2p.MsgReadWriter) error { +func (self *Pss) addPeerTopic(id pot.Address, topic Topic, rw p2p.MsgReadWriter) error { if self.peerPool[id] == nil { - self.peerPool[id] = make(map[PssTopic]p2p.MsgReadWriter, PssPeerTopicDefaultCapacity) + self.peerPool[id] = make(map[Topic]p2p.MsgReadWriter, PssPeerTopicDefaultCapacity) } self.peerPool[id][topic] = rw return nil } -func (self *Pss) removePeerTopic(rw p2p.MsgReadWriter, topic PssTopic) { +func (self *Pss) removePeerTopic(rw p2p.MsgReadWriter, topic Topic) { prw, ok := rw.(*PssReadWriter) if !ok { return @@ -389,7 +386,10 @@ func (self *Pss) isSelfRecipient(msg *PssMsg) bool { return bytes.Equal(msg.To, self.Overlay.BaseAddr()) } -func (self *Pss) isActive(id pot.Address, topic PssTopic) bool { +func (self *Pss) isActive(id pot.Address, topic Topic) bool { + if self.peerPool[id] == nil { + return false + } return self.peerPool[id][topic] != nil } @@ -405,7 +405,7 @@ type PssReadWriter struct { LastActive time.Time rw chan p2p.Msg spec *protocols.Spec - topic *PssTopic + topic *Topic } // Implements p2p.MsgReader @@ -420,7 +420,7 @@ func (prw PssReadWriter) WriteMsg(msg p2p.Msg) error { log.Warn("got writemsg pssclient", "msg", msg) rlpdata := make([]byte, msg.Size) msg.Payload.Read(rlpdata) - pmsg, err := rlp.EncodeToBytes(PssProtocolMsg{ + pmsg, err := rlp.EncodeToBytes(ProtocolMsg{ Code: msg.Code, Size: msg.Size, Payload: rlpdata, @@ -428,7 +428,7 @@ func (prw PssReadWriter) WriteMsg(msg p2p.Msg) error { if err != nil { return err } - return prw.Pss.Send(prw.To.Bytes(), *prw.topic, pmsg) + return prw.SendRaw(prw.To.Bytes(), *prw.topic, pmsg) } // Injects a p2p.Msg into the MsgReadWriter, so that it appears on the associated p2p.MsgReader @@ -442,13 +442,12 @@ func (prw PssReadWriter) injectMsg(msg p2p.Msg) error { type PssProtocol struct { *Pss proto *p2p.Protocol - topic *PssTopic + topic *Topic spec *protocols.Spec } // Constructor -//func RegisterPssProtocol(pss *Pss, topic *PssTopic, spec *protocols.Spec, targetprotocol *p2p.Protocol) *PssProtocol { -func RegisterPssProtocol(ps *Pss, topic *PssTopic, spec *protocols.Spec, targetprotocol *p2p.Protocol) error { +func RegisterPssProtocol(ps *Pss, topic *Topic, spec *protocols.Spec, targetprotocol *p2p.Protocol) error { pp := &PssProtocol{ Pss: ps, proto: targetprotocol, @@ -461,7 +460,7 @@ func RegisterPssProtocol(ps *Pss, topic *PssTopic, spec *protocols.Spec, targetp func (self *PssProtocol) handle(msg []byte, p *p2p.Peer, senderAddr []byte) error { hashoaddr := pot.NewHashAddressFromBytes(senderAddr).Address - if !self.Pss.isActive(hashoaddr, *self.topic) { + if !self.isActive(hashoaddr, *self.topic) { rw := &PssReadWriter{ Pss: self.Pss, To: hashoaddr, diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index d289ec7fed..875771b463 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -6,7 +6,6 @@ import ( "encoding/hex" "fmt" "io/ioutil" - //"math/rand" "os" "sync" "testing" @@ -41,12 +40,10 @@ func init() { log.Root().SetHandler(h) } -func TestPssCache(t *testing.T) { +func TestCache(t *testing.T) { var err error to, _ := hex.DecodeString("08090a0b0c0d0e0f1011121314150001020304050607161718191a1b1c1d1e1f") oaddr, _ := hex.DecodeString("000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f") - //uaddr, _ := hex.DecodeString("101112131415161718191a1b1c1d1e1f000102030405060708090a0b0c0d0e0f") - //proofbytes := []byte{241, 172, 117, 105, 88, 154, 82, 33, 176, 188, 91, 244, 245, 85, 86, 16, 120, 232, 70, 45, 182, 188, 99, 103, 157, 3, 202, 121, 252, 21, 129, 22} proofbytes, _ := hex.DecodeString("822fff7527f7ae630c1224921e50a7ca1b27324f00f3966623bd503780c7ab33") ps := NewTestPss(oaddr) pp := NewPssParams(false) @@ -54,20 +51,20 @@ func TestPssCache(t *testing.T) { datatwo := []byte("bar") fwdaddr := network.RandomAddr() msg := &PssMsg{ - Payload: &PssEnvelope{ + Payload: &Envelope{ TTL: 0, From: oaddr, - Topic: PssPingTopic, + Topic: PingTopic, Payload: data, }, To: to, } msgtwo := &PssMsg{ - Payload: &PssEnvelope{ + Payload: &Envelope{ TTL: 0, From: oaddr, - Topic: PssPingTopic, + Topic: PingTopic, Payload: datatwo, }, To: to, @@ -133,7 +130,7 @@ func TestPssCache(t *testing.T) { } } -func TestPssRegisterHandler(t *testing.T) { +func TestRegisterHandler(t *testing.T) { var err error addr := network.RandomAddr() ps := NewTestPss(addr.OAddr) @@ -151,13 +148,13 @@ func TestPssRegisterHandler(t *testing.T) { return nil } deregister := ps.Register(&topic, checkMsg) - pssmsg := &PssMsg{Payload: NewPssEnvelope(from.OAddr, topic, payload)} + pssmsg := &PssMsg{Payload: NewEnvelope(from.OAddr, topic, payload)} err = ps.Process(pssmsg) if err != nil { t.Fatal(err) } var i int - err = ps.Process(&PssMsg{Payload: NewPssEnvelope(from.OAddr, wrongtopic, payload)}) + err = ps.Process(&PssMsg{Payload: NewEnvelope(from.OAddr, wrongtopic, payload)}) expErr := "" if err == nil || err.Error() == expErr { t.Fatalf("unhandled topic expected '%v', got '%v'", expErr, err) @@ -172,25 +169,25 @@ func TestPssRegisterHandler(t *testing.T) { } deregister() deregister2() - err = ps.Process(&PssMsg{Payload: NewPssEnvelope(from.OAddr, topic, payload)}) + err = ps.Process(&PssMsg{Payload: NewEnvelope(from.OAddr, topic, payload)}) expErr = "" if err == nil || err.Error() == expErr { t.Fatalf("reregister handler expected %v, got %v", expErr, err) } } -func TestPssSimpleLinear(t *testing.T) { +func TestSimpleLinear(t *testing.T) { var err error nodeconfig := adapters.RandomNodeConfig() addr := network.NewAddrFromNodeID(nodeconfig.ID) _ = p2ptest.NewTestPeerPool() ps := NewTestPss(addr.Over()) - ping := &PssPing{ - QuitC: make(chan struct{}), + ping := &Ping{ + C: make(chan struct{}), } - err = RegisterPssProtocol(ps, &PssPingTopic, PssPingProtocol, NewPssPingProtocol(ping.PssPingHandler)) + err = RegisterPssProtocol(ps, &PingTopic, PingProtocol, NewPingProtocol(ping.PingHandler)) if err != nil { t.Fatalf("Failed to register virtual protocol in pss: %v", err) @@ -198,7 +195,7 @@ func TestPssSimpleLinear(t *testing.T) { run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { id := p.ID() pp := protocols.NewPeer(p, rw, pssSpec) - bp := &testPssOverlayConn{ + bp := &testOverlayConn{ Peer: pp, addr: network.ToOverlayAddr(id[:]), } @@ -212,7 +209,7 @@ func TestPssSimpleLinear(t *testing.T) { pt := p2ptest.NewProtocolTester(t, nodeconfig.ID, 2, run) - msg := NewPssPingMsg(network.ToOverlayAddr(pt.IDs[0].Bytes()), PssPingProtocol, PssPingTopic, []byte{1, 2, 3}) + msg := NewPingMsg(network.ToOverlayAddr(pt.IDs[0].Bytes()), PingProtocol, PingTopic, []byte{1, 2, 3}) exchange := p2ptest.Exchange{ Expects: []p2ptest.Expect{ @@ -237,27 +234,27 @@ func TestPssSimpleLinear(t *testing.T) { } } -func TestPssFullRandom50n(t *testing.T) { +func TestFullRandom50n(t *testing.T) { adapter := adapters.NewSimAdapter(services) - testPssFullRandom(t, adapter, 50, 50, 50) + testFullRandom(t, adapter, 50, 50, 50) } -func TestPssFullRandom25n(t *testing.T) { +func TestFullRandom25n(t *testing.T) { adapter := adapters.NewSimAdapter(services) - testPssFullRandom(t, adapter, 25, 25, 25) + testFullRandom(t, adapter, 25, 25, 25) } -func TestPssFullRandom10n(t *testing.T) { +func TestFullRandom10n(t *testing.T) { adapter := adapters.NewSimAdapter(services) - testPssFullRandom(t, adapter, 10, 10, 10) + testFullRandom(t, adapter, 10, 10, 10) } -func TestPssFullRandom5n(t *testing.T) { +func TestFullRandom5n(t *testing.T) { adapter := adapters.NewSimAdapter(services) - testPssFullRandom(t, adapter, 5, 5, 5) + testFullRandom(t, adapter, 5, 5, 5) } -func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int, fullnodecount int, msgcount int) { +func testFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int, fullnodecount int, msgcount int) { var i int var msgfromids []discover.NodeID var msgreceived []discover.NodeID @@ -393,10 +390,10 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int if err != nil { return fmt.Errorf("error getting node client: %s", err) } - msg := PssPingMsg{Created: time.Now()} - code, _ := PssPingProtocol.GetCode(&PssPingMsg{}) + msg := PingMsg{Created: time.Now()} + code, _ := PingProtocol.GetCode(&PingMsg{}) pmsg, _ := NewProtocolMsg(code, msg) - client.CallContext(ctx, &rpcerr, "pss_sendPss", PssPingTopic, PssAPIMsg{ + client.CallContext(ctx, &rpcerr, "pss_send", PingTopic, APIMsg{ Addr: fullpeers[msgtoids[ii]], Msg: pmsg, }) @@ -467,8 +464,8 @@ func triggerChecks(ctx context.Context, wg *sync.WaitGroup, trigger *chan discov return fmt.Errorf("error getting peer events for node %v: %s", id, err) } - msgevents := make(chan PssAPIMsg) - msgsub, err := client.Subscribe(context.Background(), "pss", msgevents, "receivePss", PssPingTopic) + msgevents := make(chan APIMsg) + msgsub, err := client.Subscribe(context.Background(), "pss", msgevents, "receive", PingTopic) if err != nil { return fmt.Errorf("error getting msg events for node %v: %s", id, err) } @@ -543,10 +540,10 @@ func newServices() adapters.Services { pssp := NewPssParams(true) ps := NewPss(kademlia(ctx.Config.ID), dpa, pssp) - ping := &PssPing{ - QuitC: make(chan struct{}), + ping := &Ping{ + C: make(chan struct{}), } - err = RegisterPssProtocol(ps, &PssPingTopic, PssPingProtocol, NewPssPingProtocol(ping.PssPingHandler)) + err = RegisterPssProtocol(ps, &PingTopic, PingProtocol, NewPingProtocol(ping.PingHandler)) if err != nil { log.Error("Couldnt register pss protocol", "err", err) os.Exit(1) @@ -557,10 +554,12 @@ func newServices() adapters.Services { //"bzz": func(id discover.NodeID, snapshot []byte) node.Service { "bzz": func(ctx *adapters.ServiceContext) (node.Service, error) { addr := network.NewAddrFromNodeID(ctx.Config.ID) + hp := network.NewHiveParams() + hp.Discovery = true config := &network.BzzConfig{ OverlayAddr: addr.Over(), UnderlayAddr: addr.Under(), - HiveParams: network.NewHiveParams(), + HiveParams: hp, } return network.NewBzz(config, kademlia(ctx.Config.ID), stateStore), nil }, @@ -573,22 +572,22 @@ type connmap struct { lock sync.Mutex } -type testPssOverlayConn struct { +type testOverlayConn struct { *protocols.Peer addr []byte } -func (self *testPssOverlayConn) Address() []byte { +func (self *testOverlayConn) Address() []byte { return self.addr } -func (self *testPssOverlayConn) Off() network.OverlayAddr { +func (self *testOverlayConn) Off() network.OverlayAddr { return self } -func (self *testPssOverlayConn) Drop(err error) { +func (self *testOverlayConn) Drop(err error) { } -func (self *testPssOverlayConn) Update(o network.OverlayAddr) network.OverlayAddr { +func (self *testOverlayConn) Update(o network.OverlayAddr) network.OverlayAddr { return self } diff --git a/swarm/pss/pssapi.go b/swarm/pss/pssapi.go index 80c5aad453..e5221c7494 100644 --- a/swarm/pss/pssapi.go +++ b/swarm/pss/pssapi.go @@ -11,18 +11,18 @@ import ( "github.com/ethereum/go-ethereum/swarm/network" ) -// PssAPI is the RPC API module for Pss -type PssAPI struct { +// API is the RPC API module for Pss +type API struct { *Pss } -// NewPssAPI constructs a PssAPI instance -func NewPssAPI(ps *Pss) *PssAPI { - return &PssAPI{Pss: ps} +// NewAPI constructs a PssAPI instance +func NewAPI(ps *Pss) *API { + return &API{Pss: ps} } // NewMsg API endpoint creates an RPC subscription -func (pssapi *PssAPI) ReceivePss(ctx context.Context, topic PssTopic) (*rpc.Subscription, error) { +func (pssapi *API) Receive(ctx context.Context, topic Topic) (*rpc.Subscription, error) { notifier, supported := rpc.NotifierFromContext(ctx) if !supported { return nil, fmt.Errorf("Subscribe not supported") @@ -30,7 +30,7 @@ func (pssapi *PssAPI) ReceivePss(ctx context.Context, topic PssTopic) (*rpc.Subs psssub := notifier.CreateSubscription() handler := func(msg []byte, p *p2p.Peer, from []byte) error { - apimsg := &PssAPIMsg{ + apimsg := &APIMsg{ Msg: msg, Addr: from, } @@ -39,7 +39,7 @@ func (pssapi *PssAPI) ReceivePss(ctx context.Context, topic PssTopic) (*rpc.Subs } return nil } - deregf := pssapi.Pss.Register(&topic, handler) + deregf := pssapi.Register(&topic, handler) go func() { defer deregf() @@ -55,39 +55,35 @@ func (pssapi *PssAPI) ReceivePss(ctx context.Context, topic PssTopic) (*rpc.Subs } // SendRaw sends the message (serialized into byte slice) to a peer with topic -func (pssapi *PssAPI) SendPss(topic PssTopic, msg PssAPIMsg) error { - if pssapi.Pss.debug && bytes.Equal(msg.Addr, pssapi.Pss.BaseAddr()) { +func (pssapi *API) Send(topic Topic, msg APIMsg) error { + if pssapi.debug && bytes.Equal(msg.Addr, pssapi.BaseAddr()) { log.Warn("Pss debug enabled; send to self shortcircuit", "apimsg", msg, "topic", topic) - env := NewPssEnvelope(msg.Addr, topic, msg.Msg) - return pssapi.Pss.Process(&PssMsg{ - To: pssapi.Pss.BaseAddr(), + env := NewEnvelope(msg.Addr, topic, msg.Msg) + return pssapi.Process(&PssMsg{ + To: pssapi.BaseAddr(), Payload: env, }) } - return pssapi.Pss.Send(msg.Addr, topic, msg.Msg) - /*if err != nil { - return fmt.Errorf("send error: %v", err) - } - return fmt.Errorf("ok sent")*/ + return pssapi.SendRaw(msg.Addr, topic, msg.Msg) } // PssAPITest are temporary API calls for development use only // These symbols should not be included in production environment -type PssAPITest struct { +type APITest struct { *Pss } -// NewPssAPI constructs a PssAPI instance -func NewPssAPITest(ps *Pss) *PssAPITest { - return &PssAPITest{Pss: ps} +// NewAPI constructs a API instance +func NewAPITest(ps *Pss) *APITest { + return &APITest{Pss: ps} } // temporary for access to overlay while faking kademlia healthy routines -func (pssapitest *PssAPITest) GetForwarder(addr []byte) (fwd struct { +func (pssapitest *APITest) GetForwarder(addr []byte) (fwd struct { Addr []byte Count int }) { - pssapitest.Pss.Overlay.EachConn(addr, 255, func(op network.OverlayConn, po int, isproxbin bool) bool { + pssapitest.Overlay.EachConn(addr, 255, func(op network.OverlayConn, po int, isproxbin bool) bool { if bytes.Equal(fwd.Addr, []byte{}) { fwd.Addr = op.Address() } @@ -98,6 +94,6 @@ func (pssapitest *PssAPITest) GetForwarder(addr []byte) (fwd struct { } // BaseAddr gets our own overlayaddress -func (pssapitest *PssAPITest) BaseAddr() ([]byte, error) { +func (pssapitest *APITest) BaseAddr() ([]byte, error) { return pssapitest.Pss.BaseAddr(), nil } diff --git a/swarm/pss/simulations/service_test.go b/swarm/pss/simulations/service_test.go index 28ab54c229..8fb1d10269 100644 --- a/swarm/pss/simulations/service_test.go +++ b/swarm/pss/simulations/service_test.go @@ -31,7 +31,7 @@ func init() { // TestPssProtocol starts a pss network along with two test nodes which run // protocols via the pss network, connects those two test nodes and then // waits for them to handshake -func TestPssProtocol(t *testing.T) { +func TestProtocol(t *testing.T) { // define the services w := &testWrapper{} services := adapters.Services{ @@ -141,7 +141,7 @@ func (t *testWrapper) newTestService(ctx *adapters.ServiceContext) (node.Service } return &testService{ id: ctx.Config.ID, - pss: client.NewPssClientWithRPC(context.Background(), rpcClient), + pss: client.NewClientWithRPC(context.Background(), rpcClient), handshakes: make(chan *testHandshake), }, nil } @@ -154,7 +154,7 @@ type testHandshake struct { // so that clients can wait for handshakes to complete type testService struct { id discover.NodeID - pss *client.PssClient + pss *client.Client handshakes chan *testHandshake } @@ -199,7 +199,7 @@ func (t *testService) run(_ *p2p.Peer, rw p2p.MsgReadWriter) error { } type TestAPI struct { - pss *client.PssClient + pss *client.Client handshakes chan *testHandshake } diff --git a/swarm/pss/types.go b/swarm/pss/types.go index 8743842d7d..92b1f520df 100644 --- a/swarm/pss/types.go +++ b/swarm/pss/types.go @@ -21,21 +21,21 @@ const ( // Defines params for Pss type PssParams struct { Cachettl time.Duration - Debug bool + Debug bool } // Initializes default params for Pss func NewPssParams(debug bool) *PssParams { return &PssParams{ Cachettl: defaultDigestCacheTTL, - Debug: debug, + Debug: debug, } } // Encapsulates the message transported over pss. type PssMsg struct { To []byte - Payload *PssEnvelope + Payload *Envelope } func (msg *PssMsg) Serialize() []byte { @@ -51,23 +51,23 @@ func (self *PssMsg) String() string { // Topic defines the context of a message being transported over pss // It is used by pss to determine what action is to be taken on an incoming message // Typically, one can map protocol handlers for the message payloads by mapping topic to them; see *Pss.Register() -type PssTopic [TopicLength]byte +type Topic [TopicLength]byte -func (self *PssTopic) String() string { +func (self *Topic) String() string { return fmt.Sprintf("%x", self) } // Pre-Whisper placeholder, payload of PssMsg -type PssEnvelope struct { - Topic PssTopic +type Envelope struct { + Topic Topic TTL uint16 Payload []byte From []byte } // creates Pss envelope from sender address, topic and raw payload -func NewPssEnvelope(addr []byte, topic PssTopic, payload []byte) *PssEnvelope { - return &PssEnvelope{ +func NewEnvelope(addr []byte, topic Topic, payload []byte) *Envelope { + return &Envelope{ From: addr, Topic: topic, TTL: DefaultTTL, @@ -76,7 +76,7 @@ func NewPssEnvelope(addr []byte, topic PssTopic, payload []byte) *PssEnvelope { } // encapsulates a protocol msg as PssEnvelope data -type PssProtocolMsg struct { +type ProtocolMsg struct { Code uint64 Size uint32 Payload []byte @@ -85,7 +85,7 @@ type PssProtocolMsg struct { // PssAPIMsg is the type for messages, it extends the rlp encoded protocol Msg // with the Sender's overlay address -type PssAPIMsg struct { +type APIMsg struct { Msg []byte Addr []byte } @@ -100,7 +100,7 @@ func NewProtocolMsg(code uint64, msg interface{}) ([]byte, error) { // previous attempts corrupted nested structs in the payload iself upon deserializing // therefore we use two separate []byte fields instead of peerAddr // TODO verify that nested structs cannot be used in rlp - smsg := &PssProtocolMsg{ + smsg := &ProtocolMsg{ Code: code, Size: uint32(len(rlpdata)), Payload: rlpdata, @@ -110,12 +110,12 @@ func NewProtocolMsg(code uint64, msg interface{}) ([]byte, error) { } // Message handler func for a topic -type PssHandler func(msg []byte, p *p2p.Peer, from []byte) error +type Handler func(msg []byte, p *p2p.Peer, from []byte) error // constructs a new PssTopic from a given name and version. // // Analogous to the name and version members of p2p.Protocol -func NewTopic(s string, v int) (topic PssTopic) { +func NewTopic(s string, v int) (topic Topic) { h := sha3.NewKeccak256() h.Write([]byte(s)) buf := make([]byte, TopicLength/8) @@ -126,7 +126,7 @@ func NewTopic(s string, v int) (topic PssTopic) { } func ToP2pMsg(msg []byte) (p2p.Msg, error) { - payload := &PssProtocolMsg{} + payload := &ProtocolMsg{} if err := rlp.DecodeBytes(msg, payload); err != nil { return p2p.Msg{}, fmt.Errorf("pss protocol handler unable to decode payload as p2p message: %v", err) } diff --git a/swarm/swarm.go b/swarm/swarm.go index 9d5d5a0b21..09260c634f 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -21,7 +21,6 @@ import ( "context" "crypto/ecdsa" "fmt" - "time" "github.com/ethereum/go-ethereum/accounts/abi/bind" "github.com/ethereum/go-ethereum/common" @@ -37,26 +36,27 @@ import ( 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" ) // 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) + //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 - pssEnabled bool - pss *network.Pss 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 { @@ -89,7 +89,6 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api. backend: backend, privateKey: config.Swap.PrivateKey(), corsString: cors, - pssEnabled: pssEnabled, } log.Debug(fmt.Sprintf("Setting up Swarm service components")) @@ -102,18 +101,26 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api. log.Debug("Set up local db access (iterator/counter)") kp := network.NewKadParams() - to := network.NewKademlia( common.FromHex(config.BzzKey), kp, ) - // set up the kademlia hive - self.hive = network.NewHive( - config.HiveParams, // configuration parameters - to, - ) - log.Debug(fmt.Sprintf("Set up swarm network with Kademlia hive")) + bzzconfig := &network.BzzConfig{ + UnderlayAddr: common.FromHex(config.PublicKey), + OverlayAddr: common.FromHex(config.BzzKey), + 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") @@ -126,6 +133,12 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api. self.dpa = storage.NewDPA(dpaChunkStore, self.config.ChunkerParams) log.Debug(fmt.Sprintf("-> Content Store API")) + // 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) @@ -176,25 +189,16 @@ func (self *Swarm) Start(net *p2p.Server) error { log.Warn(fmt.Sprintf("Starting Swarm service")) - self.hive.Start( - net, - func() <-chan time.Time { - return time.NewTicker(time.Second).C - }, - nil, - ) + // 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: %v", self.bzz.Hive.Overlay.BaseAddr())) - log.Info(fmt.Sprintf("Swarm network started on bzz address: %v", self.hive.BaseAddr())) - - if self.pssEnabled { - pssparams := network.NewPssParams() - self.pss = network.NewPss(self.hive.Overlay, pssparams) - - // for testing purposes, shold be removed in production environment!! - pingtopic, _ := network.MakeTopic("pss", 1) - self.pss.Register(pingtopic, self.pss.GetPingHandler()) - - log.Debug("Pss started: %v", self.pss) + if self.pss != nil { + self.pss.Start(net) } self.dpa.Start() @@ -222,7 +226,11 @@ func (self *Swarm) Start(net *p2p.Server) error { // stops all component services. func (self *Swarm) Stop() error { self.dpa.Stop() - self.hive.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() @@ -236,53 +244,30 @@ func (self *Swarm) Stop() error { } // implements the node.Service interface -func (self *Swarm) Protocols() []p2p.Protocol { - ct := network.BzzCodeMap() - if self.pssEnabled { - ct.Register(1, &network.PssMsg{}) +func (self *Swarm) Protocols() (protos []p2p.Protocol) { + + for _, p := range self.bzz.Protocols() { + protos = append(protos, p) } - ct.Register(2, network.DiscoveryMsgs...) - // srv := func(p network.Peer) error { - // if self.pssEnabled { - // p.Register(&network.PssMsg{}, func(msg interface{}) error { - // pssmsg := msg.(*network.PssMsg) + if self.pss != nil { + 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, + // ) // - // if self.pss.IsSelfRecipient(pssmsg) { - // log.Trace("pss for us, yay! ... let's process!") - // env := pssmsg.Payload - // umsg := env.Payload - // f := self.pss.GetHandler(env.Topic) - // if f == nil { - // return fmt.Errorf("No registered handler for topic '%s'", env.Topic) - // } - // nid := discover.MustBytesID(env.SenderUAddr) - // p := p2p.NewPeer(nid.NodeID, fmt.Sprintf("%x", common.ByteLabel(nid.Bytes())), []p2p.Cap{}) - // return f(umsg, p, env.SenderOAddr) - // } else { - // log.Trace("pss was for someone else :'( ... forwarding") - // return self.pss.Forward(pssmsg) - // } - // return nil - // }) - // } - // self.hive.Add(p) - // p.DisconnectHook(func(err error) { - // self.hive.Remove(p) - // }) - // return nil - // } - - proto := network.Bzz( - self.hive.Overlay.GetAddr().Over(), - self.hive.Overlay.GetAddr().Under(), - ct, - nil, - nil, - nil, - ) - - return []p2p.Protocol{*proto} + return } // implements node.Service @@ -301,7 +286,7 @@ func (self *Swarm) APIs() []rpc.API { { Namespace: "bzz", Version: "0.1", - Service: api.NewControl(self.api, self.hive), + Service: api.NewControl(self.api, self.bzz.Hive), Public: false, }, { @@ -333,13 +318,14 @@ func (self *Swarm) APIs() []rpc.API { // {Namespace, Version, api.NewAdmin(self), false}, } - if self.pssEnabled { - apis = append(apis, rpc.API{ - Namespace: "eth", - Version: "0.1/pss", - Service: network.NewPssApi(self.pss), - Public: true, - }) + 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