From 7341530f6fbaafb844d7ba0e90ccc1703c39266f Mon Sep 17 00:00:00 2001 From: nolash Date: Thu, 18 May 2017 18:15:50 +0200 Subject: [PATCH] swarm/pss, p2p/simulations/adapters: added pss routing peerpool --- p2p/simulations/adapters/inproc.go | 62 ++++++------ swarm/pss/client.go | 126 ++++++++++++------------ swarm/pss/pss.go | 78 +++++++-------- swarm/pss/pss_test.go | 149 ++++++++++++++++++++++------- swarm/pss/pssapi.go | 8 +- 5 files changed, 248 insertions(+), 175 deletions(-) diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go index 37438ce276..cae306712a 100644 --- a/p2p/simulations/adapters/inproc.go +++ b/p2p/simulations/adapters/inproc.go @@ -35,8 +35,8 @@ import ( // SimAdapter is a NodeAdapter which creates in-memory nodes and connects them // using an in-memory p2p.MsgReadWriter pipe type SimAdapter struct { - mtx sync.RWMutex - nodes map[discover.NodeID]*SimNode + mtx sync.RWMutex + nodes map[discover.NodeID]*SimNode } // NewSimAdapter creates a SimAdapter which is capable of running in-memory @@ -44,7 +44,7 @@ type SimAdapter struct { // node is passed to the NewNode function in the NodeConfig) func NewSimAdapter(services map[string]ServiceFunc) *SimAdapter { return &SimAdapter{ - nodes: make(map[discover.NodeID]*SimNode), + nodes: make(map[discover.NodeID]*SimNode), } } @@ -56,7 +56,7 @@ func (s *SimAdapter) Name() string { // NewNode returns a new SimNode using the given config func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) { var nodeprotos []p2p.Protocol - + s.mtx.Lock() defer s.mtx.Unlock() @@ -67,25 +67,31 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) { } // check the service is valid and initialize it -/* - serviceFunc, exists := s.services[config.Service] - if !exists { - return nil, fmt.Errorf("unknown node service %q", config.Service) - } + /* + serviceFunc, exists := s.services[config.Service] + if !exists { + return nil, fmt.Errorf("unknown node service %q", config.Service) + } - node := &SimNode{ - Id: id, - config: config, - adapter: s, - serviceFunc: serviceFunc, -*/ + node := &SimNode{ + Id: id, + config: config, + adapter: s, + serviceFunc: serviceFunc, + */ //serviceFunc, exists := s.services[config.Service] - + //if !exists { // return nil, fmt.Errorf("unknown node service %q", config.Service) //} //service := serviceFunc(id) - + + for _, service := range serviceFuncs[config.Service](id, nil) { + for _, proto := range service.Protocols() { + nodeprotos = append(nodeprotos, proto) + } + } + _, err := node.New(&node.Config{ P2P: p2p.Config{ PrivateKey: config.PrivateKey, @@ -100,18 +106,12 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) { return nil, err } - for _, service := range serviceFuncs[config.Service](id, nil) { - for _, proto := range service.Protocols() { - nodeprotos = append(nodeprotos, proto) - } - } - simnode := &SimNode{ - Id: id, + Id: id, serviceFunc: serviceFuncs[config.Service], adapter: s, config: config, - running: []node.Service{}, + running: []node.Service{}, } s.nodes[id.NodeID] = simnode return simnode, nil @@ -150,11 +150,11 @@ type SimNode struct { Id *NodeId config *NodeConfig adapter *SimAdapter - serviceFunc ServiceFunc + serviceFunc ServiceFunc node *node.Node client *rpc.Client rpcMux *rpcMux - running []node.Service + running []node.Service } // Addr returns the node's discovery address @@ -210,11 +210,11 @@ func (self *SimNode) Start(snapshot []byte) error { if self.node != nil { return errors.New("node already started") } - + services := []node.ServiceConstructor{} - + sf := self.serviceFunc(self.Id, snapshot) - + for i, _ := range sf { service := sf[i] sc := func(ctx *node.ServiceContext) (node.Service, error) { @@ -238,7 +238,7 @@ func (self *SimNode) Start(snapshot []byte) error { if err != nil { return err } - + for _, service := range services { log.Debug(fmt.Sprintf("service %v", service)) if err := node.Register(service); err != nil { diff --git a/swarm/pss/client.go b/swarm/pss/client.go index d71d7928a9..bcf2c5695b 100644 --- a/swarm/pss/client.go +++ b/swarm/pss/client.go @@ -1,11 +1,11 @@ -package pss +package pss import ( "context" "fmt" "sync" "sync/atomic" - + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/log" @@ -18,60 +18,58 @@ import ( ) const ( - inboxCapacity = 3000 + inboxCapacity = 3000 outboxCapacity = 100 - addrLen = common.HashLength + addrLen = common.HashLength ) -// implements p2p.Server -// implements net.Conn type PssClient struct { - localuri string - remoteuri string - ctx context.Context - cancel func() + localuri string + remoteuri string + ctx context.Context + cancel func() subscription *rpc.ClientSubscription - topicsC chan []byte - msgC chan PssAPIMsg - quitC chan struct{} - quitting uint32 - ws *rpc.Client - lock sync.Mutex - peerPool map[PssTopic]map[pot.Address]*pssRPCRW - protos []*p2p.Protocol + topicsC chan []byte + msgC chan PssAPIMsg + quitC chan struct{} + quitting uint32 + ws *rpc.Client + lock sync.Mutex + peerPool map[PssTopic]map[pot.Address]*pssRPCRW + protos []*p2p.Protocol } type pssRPCRW struct { *PssClient topic *PssTopic - spec *protocols.Spec - msgC chan []byte - addr pot.Address + spec *protocols.Spec + msgC chan []byte + addr pot.Address } func (self *PssClient) newpssRPCRW(addr pot.Address, spec *protocols.Spec, topic *PssTopic) *pssRPCRW { - return &pssRPCRW { + return &pssRPCRW{ PssClient: self, - topic: topic, - spec: spec, - msgC: make(chan []byte), - addr: addr, + topic: topic, + spec: spec, + msgC: make(chan []byte), + addr: addr, } } func (rw *pssRPCRW) ReadMsg() (p2p.Msg, error) { - msg := <- rw.msgC + msg := <-rw.msgC log.Warn("pssrpcrw read", "msg", msg) pmsg, err := ToP2pMsg(msg) if err != nil { return p2p.Msg{}, err } - + return pmsg, nil } func (rw *pssRPCRW) WriteMsg(msg p2p.Msg) error { - + ifc, found := rw.spec.NewMsg(msg.Code) if !found { return fmt.Errorf("could not find interface for msg #%d", msg.Code) @@ -81,12 +79,12 @@ func (rw *pssRPCRW) WriteMsg(msg p2p.Msg) error { if err != nil { return fmt.Errorf("Could not render protocolmessage", "error", err) } - + return rw.PssClient.ws.CallContext(rw.PssClient.ctx, nil, "pss_sendRaw", rw.topic, PssAPIMsg{ Addr: rw.addr.Bytes(), - Msg: pmsg, + Msg: pmsg, }) - + } // remotehost: hostname of node running websockets proxy to pss (default localhost) @@ -96,38 +94,38 @@ func (rw *pssRPCRW) WriteMsg(msg p2p.Msg) error { func NewPssClient(ctx context.Context, cancel func(), remotehost string, remoteport int, secure bool, originhost string) *PssClient { prefix := "ws" - + if ctx == nil { ctx = context.Background() - cancel = func() {return} + cancel = func() { return } } pssc := &PssClient{ - msgC: make(chan PssAPIMsg), - quitC: make(chan struct{}), + msgC: make(chan PssAPIMsg), + quitC: make(chan struct{}), peerPool: make(map[PssTopic]map[pot.Address]*pssRPCRW), - ctx: ctx, - cancel: cancel, + ctx: ctx, + cancel: cancel, } if remotehost == "" { remotehost = "localhost" } - + if remoteport == 0 { remoteport = node.DefaultWSPort } - + if originhost == "" { originhost = "localhost" } - + if secure { prefix = "wss" } - + pssc.remoteuri = fmt.Sprintf("%s://%s:%d", prefix, remotehost, remoteport) pssc.localuri = fmt.Sprintf("%s://%s", prefix, originhost) - + return pssc } @@ -142,9 +140,9 @@ func (self *PssClient) Start() error { if err != nil { return fmt.Errorf("Couldnt dial pss websocket: %v", err) } - + self.ws = ws - + return nil } @@ -156,32 +154,32 @@ func (self *PssClient) RunProtocol(proto *p2p.Protocol, spec *protocols.Spec) er if err != nil { return fmt.Errorf("pss event subscription failed: %v", err) } - + self.subscription = sub - + // dispatch incoming messages go func() { for { select { - case msg := <- msgC: - var addr pot.Address - copy(addr[:], msg.Addr) - if self.peerPool[topic][addr] == nil { - self.peerPool[topic][addr] = self.newpssRPCRW(addr, spec, &topic) - nid, _ := discover.HexID("0x00") - p := p2p.NewPeer(nid, fmt.Sprintf("%v", addr), []p2p.Cap{}) - go proto.Run(p, self.peerPool[topic][addr]) - } - go func() { - self.peerPool[topic][addr].msgC <- msg.Msg - }() - case <-self.quitC: - self.shutdown() - return + case msg := <-msgC: + var addr pot.Address + copy(addr[:], msg.Addr) + if self.peerPool[topic][addr] == nil { + self.peerPool[topic][addr] = self.newpssRPCRW(addr, spec, &topic) + nid, _ := discover.HexID("0x00") + p := p2p.NewPeer(nid, fmt.Sprintf("%v", addr), []p2p.Cap{}) + go proto.Run(p, self.peerPool[topic][addr]) + } + go func() { + self.peerPool[topic][addr].msgC <- msg.Msg + }() + case <-self.quitC: + self.shutdown() + return } } }() - + self.protos = append(self.protos, proto) return nil } @@ -199,7 +197,7 @@ func (self *PssClient) AddPssPeer(addr pot.Address, spec *protocols.Spec) { } func (self *PssClient) RemovePssPeer(addr pot.Address, spec *protocols.Spec) { - topic := NewTopic(spec.Name, int(spec.Version)) + topic := NewTopic(spec.Name, int(spec.Version)) delete(self.peerPool[topic], addr) } diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 36873fcee0..318ae33404 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -13,8 +13,8 @@ import ( "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/discover" - "github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/p2p/protocols" + "github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/pot" "github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/rpc" @@ -76,23 +76,22 @@ func (self *PssTopic) String() string { // Pre-Whisper placeholder, payload of PssMsg type PssEnvelope struct { - Topic PssTopic - TTL uint16 - Payload []byte - From []byte + Topic PssTopic + 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{ - From: addr, - Topic: topic, - TTL: DefaultTTL, - Payload: payload, + From: addr, + Topic: topic, + TTL: DefaultTTL, + Payload: payload, } } - func (msg *PssMsg) serialize() []byte { rlpdata, _ := rlp.EncodeToBytes(msg) /*buf := bytes.NewBuffer(nil) @@ -103,7 +102,6 @@ func (msg *PssMsg) serialize() []byte { return rlpdata } - var pssSpec = &protocols.Spec{ Name: "pss", Version: 1, @@ -144,13 +142,15 @@ type pssHandler func(msg []byte, p *p2p.Peer, from []byte) error // - 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 + 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 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 lock sync.Mutex - dpa *storage.DPA + dpa *storage.DPA } func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) { @@ -173,12 +173,13 @@ func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) { // TODO: error check overlay integrity func NewPss(k network.Overlay, dpa *storage.DPA, params *PssParams) *Pss { return &Pss{ - Overlay: k, + Overlay: k, peerPool: make(map[pot.Address]map[PssTopic]p2p.MsgReadWriter, PssPeerCapacity), + fwdPool: make(map[pot.Address]*protocols.Peer), handlers: make(map[PssTopic]map[*pssHandler]bool), fwdcache: make(map[pssDigest]pssCacheEntry), cachettl: params.Cachettl, - dpa: dpa, + dpa: dpa, } } @@ -203,12 +204,15 @@ func (self *Pss) Protocols() []p2p.Protocol { func (self *Pss) Run(p *p2p.Peer, rw p2p.MsgReadWriter) error { pp := protocols.NewPeer(p, rw, pssSpec) + id := p.ID() + h := pot.NewHashAddressFromBytes(network.ToOverlayAddr(id[:])) + self.fwdPool[h.Address] = pp return pp.Run(self.handlePssMsg) } func (self *Pss) APIs() []rpc.API { return []rpc.API{ - rpc.API { + rpc.API{ Namespace: "pss", Version: "0.1", Service: NewPssAPI(self), @@ -251,7 +255,6 @@ func (self *Pss) deregister(topic *PssTopic, h *pssHandler) { // currently not in use as forwarder address is not known in the handler function hooked to the pss dispatcher. // it is included as a courtesy to custom transport layers that may want to implement this func (self *Pss) AddToCache(addr []byte, msg *PssMsg) error { - //digest := self.hashMsg(msg) digest, err := self.storeMsg(msg) if err != nil { return err @@ -376,17 +379,16 @@ func (self *Pss) Forward(msg *PssMsg) error { // send with kademlia // find the closest peer to the recipient and attempt to send self.Overlay.EachConn(msg.To, 256, func(op network.OverlayConn, po int, isproxbin bool) bool { - p, ok := op.(senderPeer) - if !ok { - return true - } + //p, ok := op.(senderPeer) + h := pot.NewHashAddressFromBytes(op.Address()) + pp := self.fwdPool[h.Address] addr := self.Overlay.BaseAddr() - sendMsg := fmt.Sprintf("%x: msg to %x via %x", common.ByteLabel(addr), common.ByteLabel(msg.To), common.ByteLabel(p.Address())) - if self.checkFwdCache(p.Address(), digest) { + sendMsg := fmt.Sprintf("%x: msg to %x via %x", common.ByteLabel(addr), common.ByteLabel(msg.To), common.ByteLabel(op.Address())) + if self.checkFwdCache(op.Address(), digest) { log.Info(fmt.Sprintf("%v: peer already forwarded to", sendMsg)) return true } - err := p.Send(msg) + err := pp.Send(msg) if err != nil { log.Warn(fmt.Sprintf("%v: failed forwarding: %v", sendMsg, err)) return true @@ -394,10 +396,10 @@ func (self *Pss) Forward(msg *PssMsg) error { log.Trace(fmt.Sprintf("%v: successfully forwarded", sendMsg)) sent++ // if equality holds, p is always the first peer given in the iterator - if bytes.Equal(msg.To, p.Address()) || !isproxbin { + if bytes.Equal(msg.To, op.Address()) || !isproxbin { return false } - log.Trace(fmt.Sprintf("%x is in proxbin, keep forwarding", common.ByteLabel(p.Address()))) + log.Trace(fmt.Sprintf("%x is in proxbin, keep forwarding", common.ByteLabel(op.Address()))) return true }) @@ -459,7 +461,7 @@ type PssReadWriter struct { To pot.Address LastActive time.Time rw chan p2p.Msg - spec *protocols.Spec + spec *protocols.Spec topic *PssTopic } @@ -497,21 +499,20 @@ func (prw PssReadWriter) injectMsg(msg p2p.Msg) error { type PssProtocol struct { *Pss proto *p2p.Protocol - topic *PssTopic - spec *protocols.Spec + topic *PssTopic + spec *protocols.Spec } // Constructor //func RegisterPssProtocol(pss *Pss, topic *PssTopic, spec *protocols.Spec, targetprotocol *p2p.Protocol) *PssProtocol { func RegisterPssProtocol(pss *Pss, topic *PssTopic, spec *protocols.Spec, targetprotocol *p2p.Protocol) error { pp := &PssProtocol{ - Pss: pss, + Pss: pss, proto: targetprotocol, - topic: topic, - spec: spec, + topic: topic, + spec: spec, } pss.Register(topic, pp.handle) - //return pp return nil } @@ -522,7 +523,7 @@ func (self *PssProtocol) handle(msg []byte, p *p2p.Peer, senderAddr []byte) erro Pss: self.Pss, To: hashoaddr, rw: make(chan p2p.Msg), - spec: self.spec, + spec: self.spec, topic: self.topic, } self.Pss.AddPeer(p, hashoaddr, self.proto.Run, *self.topic, rw) @@ -554,8 +555,8 @@ func newProtocolMsg(code uint64, msg interface{}) ([]byte, error) { // therefore we use two separate []byte fields instead of peerAddr // TODO verify that nested structs cannot be used in rlp smsg := &PssProtocolMsg{ - Code: code, - Size: uint32(len(rlpdata)), + Code: code, + Size: uint32(len(rlpdata)), Payload: rlpdata, } @@ -568,20 +569,19 @@ func newProtocolMsg(code uint64, msg interface{}) ([]byte, error) { func NewTopic(s string, v int) (topic PssTopic) { h := sha3.NewKeccak256() h.Write([]byte(s)) - buf := make([]byte, TopicLength / 8) + buf := make([]byte, TopicLength/8) binary.PutUvarint(buf, uint64(v)) h.Write(buf) copy(topic[:], h.Sum(buf)[:]) return topic } - func ToP2pMsg(msg []byte) (p2p.Msg, error) { payload := &PssProtocolMsg{} 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) } - + return p2p.Msg{ Code: payload.Code, Size: uint32(len(payload.Payload)), diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index d9849d042a..8e09ccc805 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -6,6 +6,7 @@ import ( "encoding/hex" "fmt" "io/ioutil" + "math/rand" "os" "testing" "time" @@ -17,6 +18,7 @@ import ( "github.com/ethereum/go-ethereum/p2p/simulations" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" + "github.com/ethereum/go-ethereum/pot" "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/storage" ) @@ -190,10 +192,13 @@ func TestPssSimpleLinear(t *testing.T) { } run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { id := p.ID() + pp := protocols.NewPeer(p, rw, pssSpec) bp := &testPssPeer{ - Peer: protocols.NewPeer(p, rw, pssSpec), + Peer: pp, addr: network.ToOverlayAddr(id[:]), } + h := pot.NewHashAddressFromBytes(bp.addr) + ps.fwdPool[h.Address] = pp ps.Overlay.On(bp) defer ps.Overlay.Off(bp) log.Debug(fmt.Sprintf("%v", ps.Overlay)) @@ -244,6 +249,8 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int trigger := make(chan *adapters.NodeId) ids := make([]*adapters.NodeId, nodeCount) + fullids := ids[0:fullnodecount] + fullpeers := [][]byte{} for i := 0; i < nodeCount; i++ { nodeconfig := adapters.RandomNodeConfig() @@ -261,6 +268,9 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int t.Fatal("error triggering checks for node %s: %s", node.ID().Label(), err) } ids[i] = node.ID() + if i < fullnodecount { + fullpeers = append(fullpeers, network.ToOverlayAddr(node.ID().Bytes())) + } } // run a simulation which connects the 10 nodes in a ring and waits @@ -295,14 +305,21 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int return false, fmt.Errorf("error getting node client: %s", err) } - log.Debug("in check", "node", id) - - if lastid != nil { - //msg := pssPingMsg{Created: time.Now(),} - client.CallContext(context.Background(), nil, "pss_sendRaw", pssPingTopic, PssAPIMsg{ - Addr: lastid.Bytes(), - Msg: []byte{1, 2, 3}, - }) + for _, fid := range fullids { + if fid == id { + fpeeridx := rand.Int() % (fullnodecount - 1) + log.Debug(fmt.Sprintf("fpeeridx %d, fpeer len %d", fpeeridx, len(fullpeers))) + if bytes.Equal(fullpeers[fpeeridx], network.ToOverlayAddr(fid.Bytes())) { + fpeeridx++ + } + msg := pssPingMsg{Created: time.Now()} + code, _ := pssPingProtocol.GetCode(&pssPingMsg{}) + pmsg, _ := newProtocolMsg(code, msg) + client.CallContext(context.Background(), nil, "pss_sendRaw", pssPingTopic, PssAPIMsg{ + Addr: fullpeers[fpeeridx], + Msg: pmsg, + }) + } } lastid = id @@ -325,6 +342,36 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int t.Fatalf("simulation failed: %s", result.Error) } + trigger = make(chan *adapters.NodeId) + + action = func(ctx context.Context) error { + return nil + } + check = func(ctx context.Context, id *adapters.NodeId) (bool, error) { + select { + case <-ctx.Done(): + return false, ctx.Err() + default: + } + + return true, nil + } + timeout = 5 * time.Second + ctx, cancel = context.WithTimeout(context.Background(), timeout) + defer cancel() + + result = simulations.NewSimulation(net).Run(ctx, &simulations.Step{ + Action: action, + Trigger: trigger, + Expect: &simulations.Expectation{ + Nodes: fullids, + Check: check, + }, + }) + if result.Error != nil { + t.Fatalf("simulation failed: %s", result.Error) + } + t.Log("Simulation Passed:") t.Logf("Duration: %s", result.FinishedAt.Sub(result.StartedAt)) @@ -334,40 +381,57 @@ func testPssFullRandom(t *testing.T, adapter adapters.NodeAdapter, nodecount int // triggerChecks triggers a simulation step check whenever a peer is added or // removed from the given node func triggerChecks(trigger chan *adapters.NodeId, net *simulations.Network, id *adapters.NodeId) error { + + gotpeer := make(map[*adapters.NodeId]bool) + node := net.GetNode(id) if node == nil { return fmt.Errorf("unknown node: %s", id) } + client, err := node.Client() + if err != nil { + return err + } + + peerevents := make(chan *p2p.PeerEvent) + peersub, err := client.Subscribe(context.Background(), "admin", peerevents, "peerEvents") + if err != nil { + 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, "newMsg", pssPingTopic) + if err != nil { + return fmt.Errorf("error getting peer events for node %v: %s", id, err) + } + go func() { - time.Sleep(time.Second) - trigger <- id - }() - /* - client, err := node.Client() - if err != nil { - return err - } - events := make(chan PssAPIMsg) - sub, err := client.Subscribe(context.Background(), "pss", events, "newMsg", topic) - if err != nil { - return fmt.Errorf("error getting peer events for node %v: %s", id, err) - } - go func() { - defer sub.Unsubscribe() - for { - select { - case msg := <-events: - log.Warn("pss rpc got msg", "msg", msg) + defer msgsub.Unsubscribe() + defer peersub.Unsubscribe() + for { + select { + case event := <-peerevents: + nid := adapters.NewNodeId(event.Peer[:]) + if event.Type == "add" && !gotpeer[nid] { trigger <- id - case err := <-sub.Err(): - if err != nil { - log.Error(fmt.Sprintf("error getting peer events for node %v", id), "err", err) - } - return + gotpeer[nid] = true } + case <-msgevents: + trigger <- id + case err := <-peersub.Err(): + if err != nil { + log.Error(fmt.Sprintf("error getting peer events for node %v", id), "err", err) + } + return + + case err := <-msgsub.Err(): + if err != nil { + log.Error(fmt.Sprintf("error getting msg for node %v", id), "err", err) + } + return } - }() - */ + } + }() return nil } @@ -404,9 +468,20 @@ func newServices() adapters.Services { log.Error("local dpa creation failed", "error", err) return nil } - pssp := NewPssParams() - return []node.Service{network.NewBzz(config, kademlia, adapters.NewSimStateStore()), NewPss(kademlia, dpa, pssp)} + pssp := NewPssParams() + ps := NewPss(kademlia, dpa, pssp) + + ping := &pssPing{ + quitC: make(chan struct{}), + } + err = RegisterPssProtocol(ps, &pssPingTopic, pssPingProtocol, newPssPingProtocol(ping.pssPingHandler)) + if err != nil { + log.Error("Couldnt register pss protocol", "err", err) + os.Exit(1) + } + + return []node.Service{network.NewBzz(config, kademlia, adapters.NewSimStateStore()), ps} }, } } diff --git a/swarm/pss/pssapi.go b/swarm/pss/pssapi.go index d0d9abd25a..f0d4b6bf44 100644 --- a/swarm/pss/pssapi.go +++ b/swarm/pss/pssapi.go @@ -36,11 +36,11 @@ func (pssapi *PssAPI) NewMsg(ctx context.Context, topic PssTopic) (*rpc.Subscrip psssub := notifier.CreateSubscription() handler := func(msg []byte, p *p2p.Peer, from []byte) error { apimsg := &PssAPIMsg{ - Msg: msg, + Msg: msg, Addr: from, } if err := notifier.Notify(psssub.ID, apimsg); err != nil { - log.Warn(fmt.Sprintf("notification on pss sub topic %v rpc (sub %v) msg %v failed!", topic, psssub.ID, msg)) + log.Warn(fmt.Sprintf("notification on pss sub topic %v rpc (sub %v) msg %v failed!", topic, psssub.ID, msg)) } return nil } @@ -60,7 +60,7 @@ func (pssapi *PssAPI) NewMsg(ctx context.Context, topic PssTopic) (*rpc.Subscrip return psssub, nil } -// SendRaw sends the message (serialised into byte slice) to a peer with topic +// SendRaw sends the message (serialized into byte slice) to a peer with topic func (pssapi *PssAPI) SendRaw(topic PssTopic, msg PssAPIMsg) error { err := pssapi.Pss.Send(msg.Addr, topic, msg.Msg) if err != nil { @@ -70,7 +70,7 @@ func (pssapi *PssAPI) SendRaw(topic PssTopic, msg PssAPIMsg) error { } // BaseAddr gets our own overlayaddress -func (pssapi *PssAPI) BaseAddr() ([]byte, error) { +func (pssapi *PssAPI) BaseAddr() ([]byte, error) { log.Warn("inside baseaddr") return pssapi.Pss.Overlay.BaseAddr(), nil }