pss: Refactor. Step 1. Refactor PssMsg (ethersphere#1734)

* Add ut testing lib

* pss: refactor msgParams to message.Flags and RLP serialization

* pss: move Topic to message package

* pss: refactor PssMsg to message package

* pss/message: Flags.EncodeRLP cosmetic change

* pss: remove ut and rewrite tests

Author: jpeletier

pss/ARCHITECTURE.md

@@ -2,7 +2,7 @@
 
 Pss provides devp2p functionality for swarm nodes without the need for a direct tcp connection between them.
 
-Messages are encapsulated in a devp2p message structure `PssMsg`. These capsules are forwarded from node to node using ordinary tcp devp2p until they reach their destination: The node or nodes who can successfully decrypt the message.
+Messages are encapsulated in a devp2p message structure `message.Message`. These capsules are forwarded from node to node using ordinary tcp devp2p until they reach their destination: The node or nodes who can successfully decrypt the message.
 
 | Layer     | Contents        |
 |-----------|-----------------|

pss/api.go

@@ -25,6 +25,7 @@ import (
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/ethersphere/swarm/log"
+	"github.com/ethersphere/swarm/pss/message"
 )
 
 // Wrapper for receiving pss messages when using the pss API
@@ -50,7 +51,7 @@ func NewAPI(ps *Pss) *API {
 //
 // All incoming messages to the node matching this topic will be encapsulated in the APIMsg
 // struct and sent to the subscriber
-func (pssapi *API) Receive(ctx context.Context, topic Topic, raw bool, prox bool) (*rpc.Subscription, error) {
+func (pssapi *API) Receive(ctx context.Context, topic message.Topic, raw bool, prox bool) (*rpc.Subscription, error) {
 	notifier, supported := rpc.NotifierFromContext(ctx)
 	if !supported {
 		return nil, fmt.Errorf("Subscribe not supported")
@@ -90,7 +91,7 @@ func (pssapi *API) Receive(ctx context.Context, topic Topic, raw bool, prox bool
 	return psssub, nil
 }
 
-func (pssapi *API) GetAddress(topic Topic, asymmetric bool, key string) (PssAddress, error) {
+func (pssapi *API) GetAddress(topic message.Topic, asymmetric bool, key string) (PssAddress, error) {
 	var addr PssAddress
 	if asymmetric {
 		peer, ok := pssapi.Pss.pubKeyPool[key][topic]
@@ -122,7 +123,7 @@ func (pssapi *API) GetPublicKey() (keybytes hexutil.Bytes) {
 }
 
 // Set Public key to associate with a particular Pss peer
-func (pssapi *API) SetPeerPublicKey(pubkey hexutil.Bytes, topic Topic, addr PssAddress) error {
+func (pssapi *API) SetPeerPublicKey(pubkey hexutil.Bytes, topic message.Topic, addr PssAddress) error {
 	pk, err := pssapi.Pss.Crypto.UnmarshalPublicKey(pubkey)
 	if err != nil {
 		return fmt.Errorf("Cannot unmarshal pubkey: %x", pubkey)
@@ -139,50 +140,50 @@ func (pssapi *API) GetSymmetricKey(symkeyid string) (hexutil.Bytes, error) {
 	return hexutil.Bytes(symkey), err
 }
 
-func (pssapi *API) GetSymmetricAddressHint(topic Topic, symkeyid string) (PssAddress, error) {
+func (pssapi *API) GetSymmetricAddressHint(topic message.Topic, symkeyid string) (PssAddress, error) {
 	return pssapi.Pss.symKeyPool[symkeyid][topic].address, nil
 }
 
-func (pssapi *API) GetAsymmetricAddressHint(topic Topic, pubkeyid string) (PssAddress, error) {
+func (pssapi *API) GetAsymmetricAddressHint(topic message.Topic, pubkeyid string) (PssAddress, error) {
 	return pssapi.Pss.pubKeyPool[pubkeyid][topic].address, nil
 }
 
-func (pssapi *API) StringToTopic(topicstring string) (Topic, error) {
-	topicbytes := BytesToTopic([]byte(topicstring))
+func (pssapi *API) StringToTopic(topicstring string) (message.Topic, error) {
+	topicbytes := message.NewTopic([]byte(topicstring))
 	if topicbytes == rawTopic {
 		return rawTopic, errors.New("Topic string hashes to 0x00000000 and cannot be used")
 	}
 	return topicbytes, nil
 }
 
-func (pssapi *API) SendAsym(pubkeyhex string, topic Topic, msg hexutil.Bytes) error {
+func (pssapi *API) SendAsym(pubkeyhex string, topic message.Topic, msg hexutil.Bytes) error {
 	if err := validateMsg(msg); err != nil {
 		return err
 	}
 	return pssapi.Pss.SendAsym(pubkeyhex, topic, msg[:])
 }
 
-func (pssapi *API) SendSym(symkeyhex string, topic Topic, msg hexutil.Bytes) error {
+func (pssapi *API) SendSym(symkeyhex string, topic message.Topic, msg hexutil.Bytes) error {
 	if err := validateMsg(msg); err != nil {
 		return err
 	}
 	return pssapi.Pss.SendSym(symkeyhex, topic, msg[:])
 }
 
-func (pssapi *API) SendRaw(addr hexutil.Bytes, topic Topic, msg hexutil.Bytes) error {
+func (pssapi *API) SendRaw(addr hexutil.Bytes, topic message.Topic, msg hexutil.Bytes) error {
 	if err := validateMsg(msg); err != nil {
 		return err
 	}
 	return pssapi.Pss.SendRaw(PssAddress(addr), topic, msg[:])
 }
 
-func (pssapi *API) GetPeerTopics(pubkeyhex string) ([]Topic, error) {
+func (pssapi *API) GetPeerTopics(pubkeyhex string) ([]message.Topic, error) {
 	topics, _, err := pssapi.Pss.GetPublickeyPeers(pubkeyhex)
 	return topics, err
 
 }
 
-func (pssapi *API) GetPeerAddress(pubkeyhex string, topic Topic) (PssAddress, error) {
+func (pssapi *API) GetPeerAddress(pubkeyhex string, topic message.Topic) (PssAddress, error) {
 	return pssapi.Pss.getPeerAddress(pubkeyhex, topic)
 }
 

pss/client/client.go

@@ -33,6 +33,7 @@ import (
 	"github.com/ethersphere/swarm/log"
 	"github.com/ethersphere/swarm/p2p/protocols"
 	"github.com/ethersphere/swarm/pss"
+	"github.com/ethersphere/swarm/pss/message"
 )
 
 const (
@@ -46,8 +47,8 @@ type Client struct {
 	BaseAddrHex string
 
 	// peers
-	peerPool map[pss.Topic]map[string]*pssRPCRW
-	protos   map[pss.Topic]*p2p.Protocol
+	peerPool map[message.Topic]map[string]*pssRPCRW
+	protos   map[message.Topic]*p2p.Protocol
 
 	// rpc connections
 	rpc  *rpc.Client
@@ -71,7 +72,7 @@ type pssRPCRW struct {
 	closed   bool
 }
 
-func (c *Client) newpssRPCRW(pubkeyid string, addr pss.PssAddress, topicobj pss.Topic) (*pssRPCRW, error) {
+func (c *Client) newpssRPCRW(pubkeyid string, addr pss.PssAddress, topicobj message.Topic) (*pssRPCRW, error) {
 	topic := topicobj.String()
 	err := c.rpc.Call(nil, "pss_setPeerPublicKey", pubkeyid, topic, hexutil.Encode(addr[:]))
 	if err != nil {
@@ -218,8 +219,8 @@ func NewClientWithRPC(rpcclient *rpc.Client) (*Client, error) {
 func newClient() (client *Client) {
 	client = &Client{
 		quitC:    make(chan struct{}),
-		peerPool: make(map[pss.Topic]map[string]*pssRPCRW),
-		protos:   make(map[pss.Topic]*p2p.Protocol),
+		peerPool: make(map[message.Topic]map[string]*pssRPCRW),
+		protos:   make(map[message.Topic]*p2p.Protocol),
 	}
 	return
 }
@@ -232,7 +233,7 @@ func newClient() (client *Client) {
 // when an incoming message is received from a peer that is not yet known to the client,
 // this peer object is instantiated, and the protocol is run on it.
 func (c *Client) RunProtocol(ctx context.Context, proto *p2p.Protocol) error {
-	topicobj := pss.BytesToTopic([]byte(fmt.Sprintf("%s:%d", proto.Name, proto.Version)))
+	topicobj := message.NewTopic([]byte(fmt.Sprintf("%s:%d", proto.Name, proto.Version)))
 	topichex := topicobj.String()
 	msgC := make(chan pss.APIMsg)
 	c.peerPool[topicobj] = make(map[string]*pssRPCRW)

pss/doc.go

@@ -16,7 +16,7 @@
 
 // Pss provides devp2p functionality for swarm nodes without the need for a direct tcp connection between them.
 //
-// Messages are encapsulated in a devp2p message structure `PssMsg`. These capsules are forwarded from node to node using ordinary tcp devp2p until it reaches its destination: The node or nodes who can successfully decrypt the message.
+// Messages are encapsulated in a devp2p message structure `message.Message`. These capsules are forwarded from node to node using ordinary tcp devp2p until it reaches its destination: The node or nodes who can successfully decrypt the message.
 //
 // Routing of messages is done using swarm's own kademlia routing. Optionally routing can be turned off, forcing the message to be sent to all peers, similar to the behavior of the whisper protocol.
 //
@@ -50,11 +50,11 @@
 //
 // Under the hood, pss implements its own MsgReadWriter, which bridges MsgReadWriter.WriteMsg with Pss.SendRaw, and deftly adds an InjectMsg method which pipes incoming messages to appear on the MsgReadWriter.ReadMsg channel.
 //
-// An incoming connection is nothing more than an actual PssMsg appearing with a certain Topic. If a Handler har been registered to that Topic, the message will be passed to it. This constitutes a "new" connection if:
+// An incoming connection is nothing more than an actual message.Message appearing with a certain Topic. If a Handler har been registered to that Topic, the message will be passed to it. This constitutes a "new" connection if:
 //
 // - The pss node never called AddPeer with this combination of remote peer address and topic, and
 //
-// - The pss node never received a PssMsg from this remote peer with this specific Topic before.
+// - The pss node never received a message.Message from this remote peer with this specific Topic before.
 //
 // If it is a "new" connection, the protocol will be "run" on the remote peer, in the same manner as if it was pre-emptively added.
 //

pss/forwarding_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/ethersphere/swarm/network"
 	"github.com/ethersphere/swarm/p2p/protocols"
 	"github.com/ethersphere/swarm/pot"
+	"github.com/ethersphere/swarm/pss/message"
 )
 
 type testCase struct {
@@ -247,7 +248,7 @@ func testForwardMsg(t *testing.T, ps *Pss, c *testCase) {
 	resultMap := make(map[pot.Address]int)
 
 	defer func() { sendFunc = sendMsg }()
-	sendFunc = func(_ *Pss, sp *network.Peer, _ *PssMsg) bool {
+	sendFunc = func(_ *Pss, sp *network.Peer, _ *message.Message) bool {
 		if tries < nFails {
 			tries++
 			return false
@@ -346,8 +347,8 @@ func newTestDiscoveryPeer(addr pot.Address, kad *network.Kademlia) *network.Peer
 	return network.NewPeer(bp, kad)
 }
 
-func newTestMsg(addr []byte) *PssMsg {
-	msg := newPssMsg(&msgParams{})
+func newTestMsg(addr []byte) *message.Message {
+	msg := message.New(message.Flags{})
 	msg.To = addr[:]
 	msg.Expire = uint32(time.Now().Add(time.Second * 60).Unix())
 	msg.Topic = [4]byte{}

pss/handshake.go

@@ -31,6 +31,7 @@ import (
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/ethersphere/swarm/log"
+	"github.com/ethersphere/swarm/pss/message"
 )
 
 const (
@@ -54,7 +55,7 @@ type handshakeMsg struct {
 	Limit   uint16
 	Keys    [][]byte
 	Request uint8
-	Topic   Topic
+	Topic   message.Topic
 }
 
 // internal representation of an individual symmetric key
@@ -112,8 +113,8 @@ type HandshakeController struct {
 	symKeySendLimit      uint16
 	symKeyCapacity       uint8
 	symKeyIndex          map[string]*handshakeKey
-	handshakes           map[string]map[Topic]*handshake
-	deregisterFuncs      map[Topic]func()
+	handshakes           map[string]map[message.Topic]*handshake
+	deregisterFuncs      map[message.Topic]func()
 }
 
 // Attach HandshakeController to pss node
@@ -128,8 +129,8 @@ func SetHandshakeController(pss *Pss, params *HandshakeParams) error {
 		symKeySendLimit:      params.SymKeySendLimit,
 		symKeyCapacity:       params.SymKeyCapacity,
 		symKeyIndex:          make(map[string]*handshakeKey),
-		handshakes:           make(map[string]map[Topic]*handshake),
-		deregisterFuncs:      make(map[Topic]func()),
+		handshakes:           make(map[string]map[message.Topic]*handshake),
+		deregisterFuncs:      make(map[message.Topic]func()),
 	}
 	api := &HandshakeAPI{
 		namespace: "pss",
@@ -147,7 +148,7 @@ func SetHandshakeController(pss *Pss, params *HandshakeParams) error {
 
 // Return all unexpired symmetric keys from store by
 // peer (public key), topic and specified direction
-func (ctl *HandshakeController) validKeys(pubkeyid string, topic *Topic, in bool) (validkeys []*string) {
+func (ctl *HandshakeController) validKeys(pubkeyid string, topic *message.Topic, in bool) (validkeys []*string) {
 	ctl.lock.Lock()
 	defer ctl.lock.Unlock()
 	now := time.Now()
@@ -177,11 +178,11 @@ func (ctl *HandshakeController) validKeys(pubkeyid string, topic *Topic, in bool
 
 // Add all given symmetric keys with validity limits to store by
 // peer (public key), topic and specified direction
-func (ctl *HandshakeController) updateKeys(pubkeyid string, topic *Topic, in bool, symkeyids []string, limit uint16) {
+func (ctl *HandshakeController) updateKeys(pubkeyid string, topic *message.Topic, in bool, symkeyids []string, limit uint16) {
 	ctl.lock.Lock()
 	defer ctl.lock.Unlock()
 	if _, ok := ctl.handshakes[pubkeyid]; !ok {
-		ctl.handshakes[pubkeyid] = make(map[Topic]*handshake)
+		ctl.handshakes[pubkeyid] = make(map[message.Topic]*handshake)
 
 	}
 	if ctl.handshakes[pubkeyid][*topic] == nil {
@@ -214,14 +215,14 @@ func (ctl *HandshakeController) updateKeys(pubkeyid string, topic *Topic, in boo
 	}
 }
 
-func (ctl *HandshakeController) releaseKey(symkeyid string, topic *Topic) bool {
+func (ctl *HandshakeController) releaseKey(symkeyid string, topic *message.Topic) bool {
 	ctl.lock.Lock()
 	defer ctl.lock.Unlock()
 	return ctl.releaseKeyNoLock(symkeyid, topic)
 }
 
 // Expire a symmetric key, making it eligible for garbage collection
-func (ctl *HandshakeController) releaseKeyNoLock(symkeyid string, topic *Topic) bool {
+func (ctl *HandshakeController) releaseKeyNoLock(symkeyid string, topic *message.Topic) bool {
 	if ctl.symKeyIndex[symkeyid] == nil {
 		log.Debug("no symkey", "symkeyid", symkeyid)
 		return false
@@ -236,7 +237,7 @@ func (ctl *HandshakeController) releaseKeyNoLock(symkeyid string, topic *Topic)
 // Expired means:
 // - expiry timestamp is set, and grace period is exceeded
 // - message validity limit is reached
-func (ctl *HandshakeController) cleanHandshake(pubkeyid string, topic *Topic, in bool, out bool) int {
+func (ctl *HandshakeController) cleanHandshake(pubkeyid string, topic *message.Topic, in bool, out bool) int {
 	ctl.lock.Lock()
 	defer ctl.lock.Unlock()
 	var deletecount int
@@ -290,7 +291,7 @@ func (ctl *HandshakeController) getSymKey(symkeyid string) *handshakeKey {
 	return ctl.symKeyIndex[symkeyid]
 }
 
-// Passed as a PssMsg handler for the topic handshake is activated on
+// Passed as a message.Message handler for the topic handshake is activated on
 // Handles incoming key exchange messages and
 // counts message usage by symmetric key (expiry limit control)
 // Only returns error if key handler fails
@@ -378,7 +379,7 @@ func (ctl *HandshakeController) handleKeys(pubkeyid string, keymsg *handshakeMsg
 // If number of valid outgoing keys is less than the ideal/max
 // amount, a request is sent for the amount of keys to make up
 // the difference
-func (ctl *HandshakeController) sendKey(pubkeyid string, topic *Topic, keycount uint8) ([]string, error) {
+func (ctl *HandshakeController) sendKey(pubkeyid string, topic *message.Topic, keycount uint8) ([]string, error) {
 
 	var requestcount uint8
 	to := PssAddress{}
@@ -392,7 +393,7 @@ func (ctl *HandshakeController) sendKey(pubkeyid string, topic *Topic, keycount
 	recvkeyids := make([]string, keycount)
 	ctl.lock.Lock()
 	if _, ok := ctl.handshakes[pubkeyid]; !ok {
-		ctl.handshakes[pubkeyid] = make(map[Topic]*handshake)
+		ctl.handshakes[pubkeyid] = make(map[message.Topic]*handshake)
 	}
 	ctl.lock.Unlock()
 
@@ -479,7 +480,7 @@ type HandshakeAPI struct {
 //
 // Fails if the incoming symmetric key store is already full (and `flush` is false),
 // or if the underlying key dispatcher fails
-func (api *HandshakeAPI) Handshake(pubkeyid string, topic Topic, sync bool, flush bool) (keys []string, err error) {
+func (api *HandshakeAPI) Handshake(pubkeyid string, topic message.Topic, sync bool, flush bool) (keys []string, err error) {
 	var hsc chan []string
 	var keycount uint8
 	if flush {
@@ -512,13 +513,13 @@ func (api *HandshakeAPI) Handshake(pubkeyid string, topic Topic, sync bool, flus
 }
 
 // Activate handshake functionality on a topic
-func (api *HandshakeAPI) AddHandshake(topic Topic) error {
+func (api *HandshakeAPI) AddHandshake(topic message.Topic) error {
 	api.ctrl.deregisterFuncs[topic] = api.ctrl.pss.Register(&topic, NewHandler(api.ctrl.handler))
 	return nil
 }
 
 // Deactivate handshake functionality on a topic
-func (api *HandshakeAPI) RemoveHandshake(topic *Topic) error {
+func (api *HandshakeAPI) RemoveHandshake(topic *message.Topic) error {
 	if _, ok := api.ctrl.deregisterFuncs[*topic]; ok {
 		api.ctrl.deregisterFuncs[*topic]()
 	}
@@ -531,7 +532,7 @@ func (api *HandshakeAPI) RemoveHandshake(topic *Topic) error {
 // The `in` and `out` parameters indicate for which direction(s)
 // symmetric keys will be returned.
 // If both are false, no keys (and no error) will be returned.
-func (api *HandshakeAPI) GetHandshakeKeys(pubkeyid string, topic Topic, in bool, out bool) (keys []string, err error) {
+func (api *HandshakeAPI) GetHandshakeKeys(pubkeyid string, topic message.Topic, in bool, out bool) (keys []string, err error) {
 	if in {
 		for _, inkey := range api.ctrl.validKeys(pubkeyid, &topic, true) {
 			keys = append(keys, *inkey)
@@ -570,7 +571,7 @@ func (api *HandshakeAPI) GetHandshakePublicKey(symkeyid string) (string, error)
 // If `flush` is set, garbage collection will be performed before returning.
 //
 // Returns true on successful removal, false otherwise
-func (api *HandshakeAPI) ReleaseHandshakeKey(pubkeyid string, topic Topic, symkeyid string, flush bool) (removed bool, err error) {
+func (api *HandshakeAPI) ReleaseHandshakeKey(pubkeyid string, topic message.Topic, symkeyid string, flush bool) (removed bool, err error) {
 	removed = api.ctrl.releaseKey(symkeyid, &topic)
 	if removed && flush {
 		api.ctrl.cleanHandshake(pubkeyid, &topic, true, true)
@@ -582,7 +583,7 @@ func (api *HandshakeAPI) ReleaseHandshakeKey(pubkeyid string, topic Topic, symke
 //
 // Overloads the pss.SendSym() API call, adding symmetric key usage count
 // for message expiry control
-func (api *HandshakeAPI) SendSym(symkeyid string, topic Topic, msg hexutil.Bytes) (err error) {
+func (api *HandshakeAPI) SendSym(symkeyid string, topic message.Topic, msg hexutil.Bytes) (err error) {
 	err = api.ctrl.pss.SendSym(symkeyid, topic, msg[:])
 	if otherErr := api.ctrl.registerSymKeyUse(symkeyid); otherErr != nil {
 		return otherErr